Remove tcpStreamWrapper struct

tap/cleaner.go

@@ -34,7 +34,7 @@ func (cl *Cleaner) clean() {
 	cl.assemblerMutex.Unlock()
 
 	cl.streamsMap.streams.Range(func(k, v interface{}) bool {
-		reqResMatcher := v.(*tcpStreamWrapper).reqResMatcher
+		reqResMatcher := v.(*tcpStream).reqResMatcher
 		if reqResMatcher == nil {
 			return true
 		}

tap/tcp_stream.go

@@ -3,6 +3,7 @@ package tap
 import (
 	"encoding/binary"
 	"sync"
+	"time"
 
 	"github.com/google/gopacket"
 	"github.com/google/gopacket/layers" // pulls in all layers decoders
@@ -30,6 +31,8 @@ type tcpStream struct {
 	servers         []tcpReader
 	ident           string
 	origin          api.Capture
+	reqResMatcher   api.RequestResponseMatcher
+	createdAt       time.Time
 	sync.Mutex
 	streamsMap *tcpStreamMap
 }
@@ -71,9 +74,9 @@ func (t *tcpStream) Accept(tcp *layers.TCP, ci gopacket.CaptureInfo, dir reassem
 	if !accept {
 		diagnose.InternalStats.RejectOpt++
 	}
-	
+
 	*start = true
-	
+
 	return accept
 }
 

tap/tcp_stream_factory.go

@@ -3,7 +3,6 @@ package tap
 import (
 	"fmt"
 	"sync"
-	"time"
 
 	"github.com/up9inc/mizu/shared/logger"
 	"github.com/up9inc/mizu/tap/api"
@@ -27,12 +26,6 @@ type tcpStreamFactory struct {
 	opts       *TapOpts
 }
 
-type tcpStreamWrapper struct {
-	stream        *tcpStream
-	reqResMatcher api.RequestResponseMatcher
-	createdAt     time.Time
-}
-
 func NewTcpStreamFactory(emitter api.Emitter, streamsMap *tcpStreamMap, opts *TapOpts) *tcpStreamFactory {
 	var ownIps []string
 
@@ -123,11 +116,7 @@ func (factory *tcpStreamFactory) New(net, transport gopacket.Flow, tcp *layers.T
 				reqResMatcher: reqResMatcher,
 			})
 
-			factory.streamsMap.Store(stream.id, &tcpStreamWrapper{
-				stream:        stream,
-				reqResMatcher: reqResMatcher,
-				createdAt:     time.Now(),
-			})
+			factory.streamsMap.Store(stream.id, stream)
 
 			factory.wg.Add(2)
 			// Start reading from channel stream.reader.bytes

tap/tcp_streams_map.go

@@ -67,10 +67,9 @@ func (streamMap *tcpStreamMap) closeTimedoutTcpStreamChannels() {
 		time.Sleep(closeTimedoutTcpChannelsIntervalMs)
 		_debug.FreeOSMemory()
 		streamMap.streams.Range(func(key interface{}, value interface{}) bool {
-			streamWrapper := value.(*tcpStreamWrapper)
-			stream := streamWrapper.stream
+			stream := value.(*tcpStream)
 			if stream.superIdentifier.Protocol == nil {
-				if !stream.isClosed && time.Now().After(streamWrapper.createdAt.Add(tcpStreamChannelTimeout)) {
+				if !stream.isClosed && time.Now().After(stream.createdAt.Add(tcpStreamChannelTimeout)) {
 					stream.Close()
 					diagnose.AppStats.IncDroppedTcpStreams()
 					logger.Log.Debugf("Dropped an unidentified TCP stream because of timeout. Total dropped: %d Total Goroutines: %d Timeout (ms): %d",