From 2bc441e0beca74240e300208ada0b50e40aa574f Mon Sep 17 00:00:00 2001 From: Leonid Bugaev Date: Wed, 11 May 2016 13:16:05 +0500 Subject: [PATCH] More fixes --- Makefile | 2 +- raw_socket_listener/listener.go | 51 +++++++++++++++---------- raw_socket_listener/listener_test.go | 29 ++++++++++++-- raw_socket_listener/tcp_message.go | 6 +++ raw_socket_listener/tcp_message_test.go | 2 +- raw_socket_listener/tcp_packet.go | 15 +++++++- 6 files changed, 78 insertions(+), 27 deletions(-) diff --git a/Makefile b/Makefile index 420fbe3..559eb4f 100644 --- a/Makefile +++ b/Makefile @@ -29,7 +29,7 @@ test: $(RUN) go test ./. -timeout 30s $(ARGS) -v test_all: - $(RUN) go test ./... -timeout 30s $(ARGS) -v + $(RUN) go test ./... -timeout 60s $(ARGS) -v testone: $(RUN) go test ./... -timeout 4s -run $(TEST) $(ARGS) -v diff --git a/raw_socket_listener/listener.go b/raw_socket_listener/listener.go index c3f70a9..b2bf26e 100644 --- a/raw_socket_listener/listener.go +++ b/raw_socket_listener/listener.go @@ -146,17 +146,28 @@ func (t *Listener) listen() { } } +func (t *Listener) deleteMessage(message *TCPMessage) { + delete(t.messages, message.ID()) + delete(t.ackAliases, message.Ack) + if message.DataAck != 0 { + delete(t.ackAliases, message.DataAck) + } + if message.DataSeq != 0 { + delete(t.seqWithData, message.DataSeq) + } + + delete(t.respAliases, message.ResponseAck) +} + func (t *Listener) dispatchMessage(message *TCPMessage) { // If already dispatched if _, ok := t.messages[message.ID()]; !ok { return } - delete(t.ackAliases, message.Ack) - delete(t.messages, message.ID()) - delete(t.respAliases, message.ResponseAck) + t.deleteMessage(message) - // log.Println("Dispatching, message", message.Seq, message.Ack, string(message.Bytes())) + log.Println("Dispatching, message", message.Seq, message.Ack, message.RequestAck, string(message.Bytes())) if message.IsIncoming { // If there were response before request @@ -375,7 +386,7 @@ func (t *Listener) processTCPPacket(packet *TCPPacket) { } }() - // log.Println("Processing packet:", packet.Ack, packet.Seq, string(packet.Data)) + // log.Println("Processing packet:", packet.Ack, packet.Seq, packet.ID) var message *TCPMessage @@ -383,28 +394,26 @@ func (t *Listener) processTCPPacket(packet *TCPPacket) { // Seek for 100-expect chunks if parentAck, ok := t.seqWithData[packet.Seq]; ok { - // log.Println("Found data package with Ack:", packet.Ack) // In case if non-first data chunks comes first - for _id, m := range t.messages { - // log.Println("Message ack:", m.Ack, m.packets[0].Addr, packet.Addr) + for _, m := range t.messages { if m.Ack == packet.Ack && bytes.Equal(m.packets[0].Addr, packet.Addr) { - delete(t.messages, _id) + t.deleteMessage(m) for _, pkt := range m.packets { - pkt.Ack = parentAck + // log.Println("Updating ack", parentAck, pkt.Ack) + pkt.UpdateAck(parentAck) // Re-queue this packets t.processTCPPacket(pkt) } } } - delete(t.seqWithData, packet.Seq) t.ackAliases[packet.Ack] = parentAck - packet.Ack = parentAck + packet.UpdateAck(parentAck) } if alias, ok := t.ackAliases[packet.Ack]; ok { - packet.Ack = alias + packet.UpdateAck(alias) } var responseRequest *request @@ -430,25 +439,28 @@ func (t *Listener) processTCPPacket(packet *TCPPacket) { } } + // Adding packet to message + message.AddPacket(packet) + // Handling Expect: 100-continue requests if len(packet.Data) > 4 && bytes.Equal(packet.Data[0:4], bPOST) { // reading last 20 bytes (not counting CRLF): last header value (if no body presented) if bytes.Equal(packet.Data[len(packet.Data)-24:len(packet.Data)-4], bExpect100ContinueCheck) { seq := packet.Seq + uint32(len(packet.Data)) t.seqWithData[seq] = packet.Ack + message.DataSeq = seq // In case if sequence packet came first - // log.Println("Looking for sequences:", seq, t.messages) - for _id, m := range t.messages { - // log.Println("SeqSEQ", m.Seq, len(m.packets)) + for _, m := range t.messages { if m.Seq == seq { + t.deleteMessage(m) + // log.Println("2: Adding ack alias:", m.Ack, packet.Ack) t.ackAliases[m.Ack] = packet.Ack for _, pkt := range m.packets { + pkt.UpdateAck(packet.Ack) message.AddPacket(pkt) } - - delete(t.messages, _id) } } @@ -459,8 +471,7 @@ func (t *Listener) processTCPPacket(packet *TCPPacket) { } } - // Adding packet to message - message.AddPacket(packet) + // log.Println("Received message:", string(message.Bytes()), message.ID(), t.messages) if isIncoming { // If message have multiple packets, delete previous alias diff --git a/raw_socket_listener/listener_test.go b/raw_socket_listener/listener_test.go index 99e0caf..ff6a6f7 100644 --- a/raw_socket_listener/listener_test.go +++ b/raw_socket_listener/listener_test.go @@ -203,7 +203,7 @@ func testChunkedSequence(t *testing.T, listener *Listener, packets ...*TCPPacket var r, req, resp *TCPMessage for _, p := range packets { - listener.processTCPPacket(p) + listener.packetsChan <- p.Dump() } select { @@ -255,10 +255,26 @@ func testChunkedSequence(t *testing.T, listener *Listener, packets ...*TCPPacket t.Error("Resp and Req UUID should be equal", string(resp.UUID()), string(req.UUID())) } - time.Sleep(15 * time.Millisecond) + time.Sleep(20 * time.Millisecond) if len(listener.messages) != 0 { - t.Error("Messages non empty:", listener.messages) + t.Fatal("Messages non empty:", listener.messages) + } + + if len(listener.ackAliases) != 0 { + t.Fatal("ackAliases non empty:", listener.ackAliases) + } + + if len(listener.seqWithData) != 0 { + t.Fatal("seqWithData non empty:", listener.seqWithData) + } + + if len(listener.respAliases) != 0 { + t.Fatal("respAliases non empty:", listener.respAliases) + } + + if len(listener.respWithoutReq) != 0 { + t.Fatal("respWithoutReq non empty:", listener.respWithoutReq) } } @@ -295,7 +311,14 @@ func TestRawListenerChunkedWrongOrder(t *testing.T) { // Should re-construct message from all possible combinations for i := 0; i < 6*5*4*3*2*1; i++ { + + if i != 87 { + continue + } + packets := permutation(i, []*TCPPacket{reqPacket1, reqPacket2, reqPacket3, reqPacket4, respPacket1, respPacket2}) + + t.Log("permutation:", i, packets) testChunkedSequence(t, listener, packets...) } } diff --git a/raw_socket_listener/tcp_message.go b/raw_socket_listener/tcp_message.go index 7380dcd..aea2b54 100644 --- a/raw_socket_listener/tcp_message.go +++ b/raw_socket_listener/tcp_message.go @@ -24,6 +24,8 @@ type TCPMessage struct { Ack uint32 ResponseAck uint32 RequestStart time.Time + DataAck uint32 + DataSeq uint32 RequestAck uint32 RequestID tcpID ResponseID tcpID @@ -114,6 +116,10 @@ func (t *TCPMessage) AddPacket(packet *TCPPacket) { } else { t.End = time.Now().Add(time.Millisecond) } + + if packet.OrigAck != 0 { + t.DataAck = packet.OrigAck + } } } diff --git a/raw_socket_listener/tcp_message_test.go b/raw_socket_listener/tcp_message_test.go index fad79a2..9bc4a50 100644 --- a/raw_socket_listener/tcp_message_test.go +++ b/raw_socket_listener/tcp_message_test.go @@ -25,7 +25,7 @@ func buildPacket(isIncoming bool, Ack, Seq uint32, Data []byte) (packet *TCPPack buf[12] = 64 buf = append(buf, Data...) - packet = ParseTCPPacket([]byte(""), buf) + packet = ParseTCPPacket([]byte("123"), buf) return packet } diff --git a/raw_socket_listener/tcp_packet.go b/raw_socket_listener/tcp_packet.go index 61ab4df..d31ed89 100644 --- a/raw_socket_listener/tcp_packet.go +++ b/raw_socket_listener/tcp_packet.go @@ -28,6 +28,7 @@ type TCPPacket struct { DestPort uint16 Seq uint32 Ack uint32 + OrigAck uint32 DataOffset uint8 Raw []byte @@ -41,13 +42,23 @@ func ParseTCPPacket(addr []byte, data []byte) (p *TCPPacket) { p = &TCPPacket{Raw: data} p.ParseBasic() p.Addr = addr + p.GenID() - copy(p.ID[:4], addr) + return +} + +func (p *TCPPacket) GenID() { + copy(p.ID[:4], p.Addr) copy(p.ID[4:], p.Raw[0:2]) // Src port copy(p.ID[6:], p.Raw[2:4]) // Dest port copy(p.ID[8:], p.Raw[8:12]) // Ack +} - return +func (p *TCPPacket) UpdateAck(ack uint32) { + p.OrigAck = p.Ack + p.Ack = ack + binary.BigEndian.PutUint32(p.Raw[8:12], ack) + p.GenID() } // ParseBasic set of fields