Define a bunch of interfaces and don't export any new structs from tap/api

tap/api/api.go

@@ -133,7 +133,7 @@ func (p *ReadProgress) Current() (n int) {
 type Dissector interface {
 	Register(*Extension)
 	Ping()
-	Dissect(b *bufio.Reader, reader *TcpReader, options *shared.TrafficFilteringOptions) error
+	Dissect(b *bufio.Reader, reader TcpReader, options *shared.TrafficFilteringOptions) error
 	Analyze(item *OutputChannelItem, resolvedSource string, resolvedDestination string, namespace string) *Entry
 	Summarize(entry *Entry) *BaseEntry
 	Represent(request map[string]interface{}, response map[string]interface{}) (object []byte, err error)

tap/api/tcp_reader.go

@@ -11,9 +11,21 @@ import (
 	"github.com/up9inc/mizu/shared/logger"
 )
 
-type TcpReaderDataMsg struct {
+type TcpReader interface {
-	bytes     []byte
+	Read(p []byte) (int, error)
-	timestamp time.Time
+	Close()
+	Run(options *shared.TrafficFilteringOptions, wg *sync.WaitGroup)
+	SendMsgIfNotClosed(msg TcpReaderDataMsg)
+	GetReqResMatcher() RequestResponseMatcher
+	GetIsClient() bool
+	GetReadProgress() *ReadProgress
+	GetParent() TcpStream
+	GetTcpID() *TcpID
+	GetCounterPair() *CounterPair
+	GetCaptureTime() time.Time
+	GetEmitter() Emitter
+	GetIsClosed() bool
+	GetExtension() *Extension
 }
 
 /* TcpReader gets reads from a channel of bytes of tcp payload, and parses it into requests and responses.
@@ -21,34 +33,53 @@ type TcpReaderDataMsg struct {
  * An TcpReader object is unidirectional: it parses either a client stream or a server stream.
  * Implements io.Reader interface (Read)
  */
-type TcpReader struct {
+type tcpReader struct {
-	Ident         string
+	ident         string
-	TcpID         *TcpID
+	tcpID         *TcpID
 	isClosed      bool
-	IsClient      bool
+	isClient      bool
-	IsOutgoing    bool
+	isOutgoing    bool
-	MsgQueue      chan TcpReaderDataMsg // Channel of captured reassembled tcp payload
+	msgQueue      chan TcpReaderDataMsg // Channel of captured reassembled tcp payload
 	data          []byte
-	Progress      *ReadProgress
+	progress      *ReadProgress
-	CaptureTime   time.Time
+	captureTime   time.Time
-	Parent        *TcpStream
+	parent        TcpStream
 	packetsSeen   uint
-	Extension     *Extension
+	extension     *Extension
-	Emitter       Emitter
+	emitter       Emitter
-	CounterPair   *CounterPair
+	counterPair   *CounterPair
-	ReqResMatcher RequestResponseMatcher
+	reqResMatcher RequestResponseMatcher
 	sync.Mutex
 }
 
-func (reader *TcpReader) Read(p []byte) (int, error) {
+func NewTcpReader(msgQueue chan TcpReaderDataMsg, progress *ReadProgress, ident string, tcpId *TcpID, captureTime time.Time, parent TcpStream, isClient bool, isOutgoing bool, extension *Extension, emitter Emitter, counterPair *CounterPair, reqResMatcher RequestResponseMatcher) TcpReader {
+	return &tcpReader{
+		msgQueue:      msgQueue,
+		progress:      progress,
+		ident:         ident,
+		tcpID:         tcpId,
+		captureTime:   captureTime,
+		parent:        parent,
+		isClient:      isClient,
+		isOutgoing:    isOutgoing,
+		extension:     extension,
+		emitter:       emitter,
+		counterPair:   counterPair,
+		reqResMatcher: reqResMatcher,
+	}
+}
+
+func (reader *tcpReader) Read(p []byte) (int, error) {
 	var msg TcpReaderDataMsg
 
 	ok := true
 	for ok && len(reader.data) == 0 {
-		msg, ok = <-reader.MsgQueue
+		msg, ok = <-reader.msgQueue
-		reader.data = msg.bytes
+		if msg != nil {
+			reader.data = msg.GetBytes()
+			reader.captureTime = msg.GetTimestamp()
+		}
 
-		reader.CaptureTime = msg.timestamp
 		if len(reader.data) > 0 {
 			reader.packetsSeen += 1
 		}
@@ -59,24 +90,24 @@ func (reader *TcpReader) Read(p []byte) (int, error) {
 
 	l := copy(p, reader.data)
 	reader.data = reader.data[l:]
-	reader.Progress.Feed(l)
+	reader.progress.Feed(l)
 
 	return l, nil
 }
 
-func (reader *TcpReader) Close() {
+func (reader *tcpReader) Close() {
 	reader.Lock()
 	if !reader.isClosed {
 		reader.isClosed = true
-		close(reader.MsgQueue)
+		close(reader.msgQueue)
 	}
 	reader.Unlock()
 }
 
-func (reader *TcpReader) Run(options *shared.TrafficFilteringOptions, wg *sync.WaitGroup) {
+func (reader *tcpReader) Run(options *shared.TrafficFilteringOptions, wg *sync.WaitGroup) {
 	defer wg.Done()
 	b := bufio.NewReader(reader)
-	err := reader.Extension.Dissector.Dissect(b, reader, options)
+	err := reader.extension.Dissector.Dissect(b, reader, options)
 	if err != nil {
 		_, err = io.Copy(ioutil.Discard, reader)
 		if err != nil {
@@ -84,3 +115,51 @@ func (reader *TcpReader) Run(options *shared.TrafficFilteringOptions, wg *sync.W
 		}
 	}
 }
+
+func (reader *tcpReader) SendMsgIfNotClosed(msg TcpReaderDataMsg) {
+	reader.Lock()
+	if !reader.isClosed {
+		reader.msgQueue <- msg
+	}
+	reader.Unlock()
+}
+
+func (reader *tcpReader) GetReqResMatcher() RequestResponseMatcher {
+	return reader.reqResMatcher
+}
+
+func (reader *tcpReader) GetIsClient() bool {
+	return reader.isClient
+}
+
+func (reader *tcpReader) GetReadProgress() *ReadProgress {
+	return reader.progress
+}
+
+func (reader *tcpReader) GetParent() TcpStream {
+	return reader.parent
+}
+
+func (reader *tcpReader) GetTcpID() *TcpID {
+	return reader.tcpID
+}
+
+func (reader *tcpReader) GetCounterPair() *CounterPair {
+	return reader.counterPair
+}
+
+func (reader *tcpReader) GetCaptureTime() time.Time {
+	return reader.captureTime
+}
+
+func (reader *tcpReader) GetEmitter() Emitter {
+	return reader.emitter
+}
+
+func (reader *tcpReader) GetIsClosed() bool {
+	return reader.isClosed
+}
+
+func (reader *tcpReader) GetExtension() *Extension {
+	return reader.extension
+}

tap/api/tcp_reader_data_msg.go

@@ -0,0 +1,25 @@
+package api
+
+import "time"
+
+type TcpReaderDataMsg interface {
+	GetBytes() []byte
+	GetTimestamp() time.Time
+}
+
+type tcpReaderDataMsg struct {
+	bytes     []byte
+	timestamp time.Time
+}
+
+func NewTcpReaderDataMsg(data []byte, timestamp time.Time) TcpReaderDataMsg {
+	return &tcpReaderDataMsg{data, timestamp}
+}
+
+func (dataMsg *tcpReaderDataMsg) GetBytes() []byte {
+	return dataMsg.bytes
+}
+
+func (dataMsg *tcpReaderDataMsg) GetTimestamp() time.Time {
+	return dataMsg.timestamp
+}

tap/api/tcp_stream.go

@@ -2,44 +2,86 @@ package api
 
 import (
 	"encoding/binary"
+	"fmt"
 	"sync"
 	"time"
 
 	"github.com/google/gopacket"
 	"github.com/google/gopacket/layers" // pulls in all layers decoders
 	"github.com/google/gopacket/reassembly"
+	"github.com/up9inc/mizu/shared"
 	"github.com/up9inc/mizu/tap/api/diagnose"
 )
 
+type TcpStream interface {
+	Accept(tcp *layers.TCP, ci gopacket.CaptureInfo, dir reassembly.TCPFlowDirection, nextSeq reassembly.Sequence, start *bool, ac reassembly.AssemblerContext) bool
+	ReassembledSG(sg reassembly.ScatterGather, ac reassembly.AssemblerContext)
+	ReassemblyComplete(ac reassembly.AssemblerContext) bool
+	Close()
+	CloseOtherProtocolDissectors(protocol *Protocol)
+	AddClient(reader TcpReader)
+	AddServer(reader TcpReader)
+	ClientRun(index int, filteringOptions *shared.TrafficFilteringOptions, wg *sync.WaitGroup)
+	ServerRun(index int, filteringOptions *shared.TrafficFilteringOptions, wg *sync.WaitGroup)
+	GetOrigin() Capture
+	GetProtoIdentifier() *ProtoIdentifier
+	GetReqResMatcher() RequestResponseMatcher
+	GetIsTapTarget() bool
+	GetId() int64
+	SetId(id int64)
+}
+
 /* It's a connection (bidirectional)
  * Implements gopacket.reassembly.Stream interface (Accept, ReassembledSG, ReassemblyComplete)
  * ReassembledSG gets called when new reassembled data is ready (i.e. bytes in order, no duplicates, complete)
  * In our implementation, we pass information from ReassembledSG to the TcpReader through a shared channel.
  */
-type TcpStream struct {
+type tcpStream struct {
-	Id              int64
+	id              int64
 	isClosed        bool
-	ProtoIdentifier *ProtoIdentifier
+	protoIdentifier *ProtoIdentifier
-	TcpState        *reassembly.TCPSimpleFSM
+	tcpState        *reassembly.TCPSimpleFSM
 	fsmerr          bool
-	Optchecker      reassembly.TCPOptionCheck
+	optchecker      reassembly.TCPOptionCheck
-	Net, Transport  gopacket.Flow
+	net, transport  gopacket.Flow
-	IsDNS           bool
+	isDNS           bool
-	IsTapTarget     bool
+	isTapTarget     bool
-	Clients         []TcpReader
+	clients         []TcpReader
-	Servers         []TcpReader
+	servers         []TcpReader
-	Ident           string
+	ident           string
-	Origin          Capture
+	origin          Capture
-	ReqResMatcher   RequestResponseMatcher
+	reqResMatcher   RequestResponseMatcher
 	createdAt       time.Time
-	StreamsMap      *TcpStreamMap
+	streamsMap      TcpStreamMap
 	sync.Mutex
 }
 
-func (t *TcpStream) Accept(tcp *layers.TCP, ci gopacket.CaptureInfo, dir reassembly.TCPFlowDirection, nextSeq reassembly.Sequence, start *bool, ac reassembly.AssemblerContext) bool {
+func NewTcpStream(net gopacket.Flow, transport gopacket.Flow, tcp *layers.TCP, isTapTarget bool, fsmOptions reassembly.TCPSimpleFSMOptions, streamsMap TcpStreamMap, capture Capture) TcpStream {
+	return &tcpStream{
+		net:             net,
+		transport:       transport,
+		isDNS:           tcp.SrcPort == 53 || tcp.DstPort == 53,
+		isTapTarget:     isTapTarget,
+		tcpState:        reassembly.NewTCPSimpleFSM(fsmOptions),
+		ident:           fmt.Sprintf("%s:%s", net, transport),
+		optchecker:      reassembly.NewTCPOptionCheck(),
+		protoIdentifier: &ProtoIdentifier{},
+		streamsMap:      streamsMap,
+		origin:          capture,
+	}
+}
+
+func NewTcpStreamDummy(capture Capture) TcpStream {
+	return &tcpStream{
+		origin:          capture,
+		protoIdentifier: &ProtoIdentifier{},
+	}
+}
+
+func (t *tcpStream) Accept(tcp *layers.TCP, ci gopacket.CaptureInfo, dir reassembly.TCPFlowDirection, nextSeq reassembly.Sequence, start *bool, ac reassembly.AssemblerContext) bool {
 	// FSM
-	if !t.TcpState.CheckState(tcp, dir) {
+	if !t.tcpState.CheckState(tcp, dir) {
-		diagnose.TapErrors.SilentError("FSM-rejection", "%s: Packet rejected by FSM (state:%s)", t.Ident, t.TcpState.String())
+		diagnose.TapErrors.SilentError("FSM-rejection", "%s: Packet rejected by FSM (state:%s)", t.ident, t.tcpState.String())
 		diagnose.InternalStats.RejectFsm++
 		if !t.fsmerr {
 			t.fsmerr = true
@@ -50,9 +92,9 @@ func (t *TcpStream) Accept(tcp *layers.TCP, ci gopacket.CaptureInfo, dir reassem
 		}
 	}
 	// Options
-	err := t.Optchecker.Accept(tcp, ci, dir, nextSeq, start)
+	err := t.optchecker.Accept(tcp, ci, dir, nextSeq, start)
 	if err != nil {
-		diagnose.TapErrors.SilentError("OptionChecker-rejection", "%s: Packet rejected by OptionChecker: %s", t.Ident, err)
+		diagnose.TapErrors.SilentError("OptionChecker-rejection", "%s: Packet rejected by OptionChecker: %s", t.ident, err)
 		diagnose.InternalStats.RejectOpt++
 		if !*nooptcheck {
 			return false
@@ -63,10 +105,10 @@ func (t *TcpStream) Accept(tcp *layers.TCP, ci gopacket.CaptureInfo, dir reassem
 	if *checksum {
 		c, err := tcp.ComputeChecksum()
 		if err != nil {
-			diagnose.TapErrors.SilentError("ChecksumCompute", "%s: Got error computing checksum: %s", t.Ident, err)
+			diagnose.TapErrors.SilentError("ChecksumCompute", "%s: Got error computing checksum: %s", t.ident, err)
 			accept = false
 		} else if c != 0x0 {
-			diagnose.TapErrors.SilentError("Checksum", "%s: Invalid checksum: 0x%x", t.Ident, c)
+			diagnose.TapErrors.SilentError("Checksum", "%s: Invalid checksum: 0x%x", t.ident, c)
 			accept = false
 		}
 	}
@@ -79,7 +121,7 @@ func (t *TcpStream) Accept(tcp *layers.TCP, ci gopacket.CaptureInfo, dir reassem
 	return accept
 }
 
-func (t *TcpStream) ReassembledSG(sg reassembly.ScatterGather, ac reassembly.AssemblerContext) {
+func (t *tcpStream) ReassembledSG(sg reassembly.ScatterGather, ac reassembly.AssemblerContext) {
 	dir, _, _, skip := sg.Info()
 	length, saved := sg.Lengths()
 	// update stats
@@ -113,7 +155,7 @@ func (t *TcpStream) ReassembledSG(sg reassembly.ScatterGather, ac reassembly.Ass
 		return
 	}
 	data := sg.Fetch(length)
-	if t.IsDNS {
+	if t.isDNS {
 		dns := &layers.DNS{}
 		var decoded []gopacket.LayerType
 		if len(data) < 2 {
@@ -140,44 +182,36 @@ func (t *TcpStream) ReassembledSG(sg reassembly.ScatterGather, ac reassembly.Ass
 		if len(data) > 2+int(dnsSize) {
 			sg.KeepFrom(2 + int(dnsSize))
 		}
-	} else if t.IsTapTarget {
+	} else if t.isTapTarget {
 		if length > 0 {
 			// This is where we pass the reassembled information onwards
 			// This channel is read by an tcpReader object
 			diagnose.AppStatsInst.IncReassembledTcpPayloadsCount()
 			timestamp := ac.GetCaptureInfo().Timestamp
 			if dir == reassembly.TCPDirClientToServer {
-				for i := range t.Clients {
+				for i := range t.clients {
-					reader := &t.Clients[i]
+					reader := t.clients[i]
-					reader.Lock()
+					reader.SendMsgIfNotClosed(NewTcpReaderDataMsg(data, timestamp))
-					if !reader.isClosed {
-						reader.MsgQueue <- TcpReaderDataMsg{data, timestamp}
-					}
-					reader.Unlock()
 				}
 			} else {
-				for i := range t.Servers {
+				for i := range t.servers {
-					reader := &t.Servers[i]
+					reader := t.servers[i]
-					reader.Lock()
+					reader.SendMsgIfNotClosed(NewTcpReaderDataMsg(data, timestamp))
-					if !reader.isClosed {
-						reader.MsgQueue <- TcpReaderDataMsg{data, timestamp}
-					}
-					reader.Unlock()
 				}
 			}
 		}
 	}
 }
 
-func (t *TcpStream) ReassemblyComplete(ac reassembly.AssemblerContext) bool {
+func (t *tcpStream) ReassemblyComplete(ac reassembly.AssemblerContext) bool {
-	if t.IsTapTarget && !t.isClosed {
+	if t.isTapTarget && !t.isClosed {
 		t.Close()
 	}
 	// do not remove the connection to allow last ACK
 	return false
 }
 
-func (t *TcpStream) Close() {
+func (t *tcpStream) Close() {
 	t.Lock()
 	defer t.Unlock()
 
@@ -187,40 +221,80 @@ func (t *TcpStream) Close() {
 
 	t.isClosed = true
 
-	t.StreamsMap.Delete(t.Id)
+	t.streamsMap.Delete(t.id)
 
-	for i := range t.Clients {
+	for i := range t.clients {
-		reader := &t.Clients[i]
+		reader := t.clients[i]
 		reader.Close()
 	}
-	for i := range t.Servers {
+	for i := range t.servers {
-		reader := &t.Servers[i]
+		reader := t.servers[i]
 		reader.Close()
 	}
 }
 
-func (t *TcpStream) CloseOtherProtocolDissectors(protocol *Protocol) {
+func (t *tcpStream) CloseOtherProtocolDissectors(protocol *Protocol) {
 	t.Lock()
 	defer t.Unlock()
 
-	if t.ProtoIdentifier.IsClosedOthers {
+	if t.protoIdentifier.IsClosedOthers {
 		return
 	}
 
-	t.ProtoIdentifier.Protocol = protocol
+	t.protoIdentifier.Protocol = protocol
 
-	for i := range t.Clients {
+	for i := range t.clients {
-		reader := &t.Clients[i]
+		reader := t.clients[i]
-		if reader.Extension.Protocol != t.ProtoIdentifier.Protocol {
+		if reader.GetExtension().Protocol != t.protoIdentifier.Protocol {
 			reader.Close()
 		}
 	}
-	for i := range t.Servers {
+	for i := range t.servers {
-		reader := &t.Servers[i]
+		reader := t.servers[i]
-		if reader.Extension.Protocol != t.ProtoIdentifier.Protocol {
+		if reader.GetExtension().Protocol != t.protoIdentifier.Protocol {
 			reader.Close()
 		}
 	}
 
-	t.ProtoIdentifier.IsClosedOthers = true
+	t.protoIdentifier.IsClosedOthers = true
+}
+
+func (t *tcpStream) AddClient(reader TcpReader) {
+	t.clients = append(t.clients, reader)
+}
+
+func (t *tcpStream) AddServer(reader TcpReader) {
+	t.servers = append(t.servers, reader)
+}
+
+func (t *tcpStream) ClientRun(index int, filteringOptions *shared.TrafficFilteringOptions, wg *sync.WaitGroup) {
+	t.clients[index].Run(filteringOptions, wg)
+}
+
+func (t *tcpStream) ServerRun(index int, filteringOptions *shared.TrafficFilteringOptions, wg *sync.WaitGroup) {
+	t.servers[index].Run(filteringOptions, wg)
+}
+
+func (t *tcpStream) GetOrigin() Capture {
+	return t.origin
+}
+
+func (t *tcpStream) GetProtoIdentifier() *ProtoIdentifier {
+	return t.protoIdentifier
+}
+
+func (t *tcpStream) GetReqResMatcher() RequestResponseMatcher {
+	return t.reqResMatcher
+}
+
+func (t *tcpStream) GetIsTapTarget() bool {
+	return t.isTapTarget
+}
+
+func (t *tcpStream) GetId() int64 {
+	return t.id
+}
+
+func (t *tcpStream) SetId(id int64) {
+	t.id = id
 }

tap/api/tcp_streams_map.go

@@ -10,31 +10,43 @@ import (
 	"github.com/up9inc/mizu/tap/api/diagnose"
 )
 
-type TcpStreamMap struct {
+type TcpStreamMap interface {
-	Streams  *sync.Map
+	Range(f func(key, value interface{}) bool)
+	Store(key, value interface{})
+	Delete(key interface{})
+	NextId() int64
+	CloseTimedoutTcpStreamChannels()
+}
+
+type tcpStreamMap struct {
+	streams  *sync.Map
 	streamId int64
 }
 
-func NewTcpStreamMap() *TcpStreamMap {
+func NewTcpStreamMap() TcpStreamMap {
-	return &TcpStreamMap{
+	return &tcpStreamMap{
-		Streams: &sync.Map{},
+		streams: &sync.Map{},
 	}
 }
 
-func (streamMap *TcpStreamMap) Store(key, value interface{}) {
+func (streamMap *tcpStreamMap) Range(f func(key, value interface{}) bool) {
-	streamMap.Streams.Store(key, value)
+	streamMap.streams.Range(f)
 }
 
-func (streamMap *TcpStreamMap) Delete(key interface{}) {
+func (streamMap *tcpStreamMap) Store(key, value interface{}) {
-	streamMap.Streams.Delete(key)
+	streamMap.streams.Store(key, value)
 }
 
-func (streamMap *TcpStreamMap) NextId() int64 {
+func (streamMap *tcpStreamMap) Delete(key interface{}) {
+	streamMap.streams.Delete(key)
+}
+
+func (streamMap *tcpStreamMap) NextId() int64 {
 	streamMap.streamId++
 	return streamMap.streamId
 }
 
-func (streamMap *TcpStreamMap) CloseTimedoutTcpStreamChannels() {
+func (streamMap *tcpStreamMap) CloseTimedoutTcpStreamChannels() {
 	tcpStreamChannelTimeoutMs := GetTcpChannelTimeoutMs()
 	closeTimedoutTcpChannelsIntervalMs := GetCloseTimedoutTcpChannelsInterval()
 	logger.Log.Infof("Using %d ms as the close timedout TCP stream channels interval", closeTimedoutTcpChannelsIntervalMs/time.Millisecond)
@@ -44,9 +56,9 @@ func (streamMap *TcpStreamMap) CloseTimedoutTcpStreamChannels() {
 		<-ticker.C
 
 		debug.FreeOSMemory()
-		streamMap.Streams.Range(func(key interface{}, value interface{}) bool {
+		streamMap.streams.Range(func(key interface{}, value interface{}) bool {
-			stream := value.(*TcpStream)
+			stream := value.(*tcpStream)
-			if stream.ProtoIdentifier.Protocol == nil {
+			if stream.protoIdentifier.Protocol == nil {
 				if !stream.isClosed && time.Now().After(stream.createdAt.Add(tcpStreamChannelTimeoutMs)) {
 					stream.Close()
 					diagnose.AppStatsInst.IncDroppedTcpStreams()

tap/cleaner.go

@@ -22,7 +22,7 @@ type Cleaner struct {
 	connectionTimeout time.Duration
 	stats             CleanerStats
 	statsMutex        sync.Mutex
-	streamsMap        *api.TcpStreamMap
+	streamsMap        api.TcpStreamMap
 }
 
 func (cl *Cleaner) clean() {
@@ -33,8 +33,8 @@ func (cl *Cleaner) clean() {
 	flushed, closed := cl.assembler.FlushCloseOlderThan(startCleanTime.Add(-cl.connectionTimeout))
 	cl.assemblerMutex.Unlock()
 
-	cl.streamsMap.Streams.Range(func(k, v interface{}) bool {
+	cl.streamsMap.Range(func(k, v interface{}) bool {
-		reqResMatcher := v.(*api.TcpStream).ReqResMatcher
+		reqResMatcher := v.(api.TcpStream).GetReqResMatcher()
 		if reqResMatcher == nil {
 			return true
 		}

tap/extensions/amqp/main.go

@@ -40,17 +40,17 @@ func (d dissecting) Ping() {
 
 const amqpRequest string = "amqp_request"
 
-func (d dissecting) Dissect(b *bufio.Reader, reader *api.TcpReader, options *shared.TrafficFilteringOptions) error {
+func (d dissecting) Dissect(b *bufio.Reader, reader api.TcpReader, options *shared.TrafficFilteringOptions) error {
 	r := AmqpReader{b}
 
 	var remaining int
 	var header *HeaderFrame
 
 	connectionInfo := &api.ConnectionInfo{
-		ClientIP:   reader.TcpID.SrcIP,
+		ClientIP:   reader.GetTcpID().SrcIP,
-		ClientPort: reader.TcpID.SrcPort,
+		ClientPort: reader.GetTcpID().SrcPort,
-		ServerIP:   reader.TcpID.DstIP,
+		ServerIP:   reader.GetTcpID().DstIP,
-		ServerPort: reader.TcpID.DstPort,
+		ServerPort: reader.GetTcpID().DstPort,
 		IsOutgoing: true,
 	}
 
@@ -76,7 +76,7 @@ func (d dissecting) Dissect(b *bufio.Reader, reader *api.TcpReader, options *sha
 	var lastMethodFrameMessage Message
 
 	for {
-		if reader.Parent.ProtoIdentifier.Protocol != nil && reader.Parent.ProtoIdentifier.Protocol != &protocol {
+		if reader.GetParent().GetProtoIdentifier().Protocol != nil && reader.GetParent().GetProtoIdentifier().Protocol != &protocol {
 			return errors.New("Identified by another protocol")
 		}
 
@@ -113,12 +113,12 @@ func (d dissecting) Dissect(b *bufio.Reader, reader *api.TcpReader, options *sha
 			switch lastMethodFrameMessage.(type) {
 			case *BasicPublish:
 				eventBasicPublish.Body = f.Body
-				reader.Parent.CloseOtherProtocolDissectors(&protocol)
+				reader.GetParent().CloseOtherProtocolDissectors(&protocol)
-				emitAMQP(*eventBasicPublish, amqpRequest, basicMethodMap[40], connectionInfo, reader.CaptureTime, reader.Progress.Current(), reader.Emitter, reader.Parent.Origin)
+				emitAMQP(*eventBasicPublish, amqpRequest, basicMethodMap[40], connectionInfo, reader.GetCaptureTime(), reader.GetReadProgress().Current(), reader.GetEmitter(), reader.GetParent().GetOrigin())
 			case *BasicDeliver:
 				eventBasicDeliver.Body = f.Body
-				reader.Parent.CloseOtherProtocolDissectors(&protocol)
+				reader.GetParent().CloseOtherProtocolDissectors(&protocol)
-				emitAMQP(*eventBasicDeliver, amqpRequest, basicMethodMap[60], connectionInfo, reader.CaptureTime, reader.Progress.Current(), reader.Emitter, reader.Parent.Origin)
+				emitAMQP(*eventBasicDeliver, amqpRequest, basicMethodMap[60], connectionInfo, reader.GetCaptureTime(), reader.GetReadProgress().Current(), reader.GetEmitter(), reader.GetParent().GetOrigin())
 			}
 
 		case *MethodFrame:
@@ -138,8 +138,8 @@ func (d dissecting) Dissect(b *bufio.Reader, reader *api.TcpReader, options *sha
 					NoWait:     m.NoWait,
 					Arguments:  m.Arguments,
 				}
-				reader.Parent.CloseOtherProtocolDissectors(&protocol)
+				reader.GetParent().CloseOtherProtocolDissectors(&protocol)
-				emitAMQP(*eventQueueBind, amqpRequest, queueMethodMap[20], connectionInfo, reader.CaptureTime, reader.Progress.Current(), reader.Emitter, reader.Parent.Origin)
+				emitAMQP(*eventQueueBind, amqpRequest, queueMethodMap[20], connectionInfo, reader.GetCaptureTime(), reader.GetReadProgress().Current(), reader.GetEmitter(), reader.GetParent().GetOrigin())
 
 			case *BasicConsume:
 				eventBasicConsume := &BasicConsume{
@@ -151,8 +151,8 @@ func (d dissecting) Dissect(b *bufio.Reader, reader *api.TcpReader, options *sha
 					NoWait:      m.NoWait,
 					Arguments:   m.Arguments,
 				}
-				reader.Parent.CloseOtherProtocolDissectors(&protocol)
+				reader.GetParent().CloseOtherProtocolDissectors(&protocol)
-				emitAMQP(*eventBasicConsume, amqpRequest, basicMethodMap[20], connectionInfo, reader.CaptureTime, reader.Progress.Current(), reader.Emitter, reader.Parent.Origin)
+				emitAMQP(*eventBasicConsume, amqpRequest, basicMethodMap[20], connectionInfo, reader.GetCaptureTime(), reader.GetReadProgress().Current(), reader.GetEmitter(), reader.GetParent().GetOrigin())
 
 			case *BasicDeliver:
 				eventBasicDeliver.ConsumerTag = m.ConsumerTag
@@ -171,8 +171,8 @@ func (d dissecting) Dissect(b *bufio.Reader, reader *api.TcpReader, options *sha
 					NoWait:     m.NoWait,
 					Arguments:  m.Arguments,
 				}
-				reader.Parent.CloseOtherProtocolDissectors(&protocol)
+				reader.GetParent().CloseOtherProtocolDissectors(&protocol)
-				emitAMQP(*eventQueueDeclare, amqpRequest, queueMethodMap[10], connectionInfo, reader.CaptureTime, reader.Progress.Current(), reader.Emitter, reader.Parent.Origin)
+				emitAMQP(*eventQueueDeclare, amqpRequest, queueMethodMap[10], connectionInfo, reader.GetCaptureTime(), reader.GetReadProgress().Current(), reader.GetEmitter(), reader.GetParent().GetOrigin())
 
 			case *ExchangeDeclare:
 				eventExchangeDeclare := &ExchangeDeclare{
@@ -185,8 +185,8 @@ func (d dissecting) Dissect(b *bufio.Reader, reader *api.TcpReader, options *sha
 					NoWait:     m.NoWait,
 					Arguments:  m.Arguments,
 				}
-				reader.Parent.CloseOtherProtocolDissectors(&protocol)
+				reader.GetParent().CloseOtherProtocolDissectors(&protocol)
-				emitAMQP(*eventExchangeDeclare, amqpRequest, exchangeMethodMap[10], connectionInfo, reader.CaptureTime, reader.Progress.Current(), reader.Emitter, reader.Parent.Origin)
+				emitAMQP(*eventExchangeDeclare, amqpRequest, exchangeMethodMap[10], connectionInfo, reader.GetCaptureTime(), reader.GetReadProgress().Current(), reader.GetEmitter(), reader.GetParent().GetOrigin())
 
 			case *ConnectionStart:
 				eventConnectionStart := &ConnectionStart{
@@ -196,8 +196,8 @@ func (d dissecting) Dissect(b *bufio.Reader, reader *api.TcpReader, options *sha
 					Mechanisms:       m.Mechanisms,
 					Locales:          m.Locales,
 				}
-				reader.Parent.CloseOtherProtocolDissectors(&protocol)
+				reader.GetParent().CloseOtherProtocolDissectors(&protocol)
-				emitAMQP(*eventConnectionStart, amqpRequest, connectionMethodMap[10], connectionInfo, reader.CaptureTime, reader.Progress.Current(), reader.Emitter, reader.Parent.Origin)
+				emitAMQP(*eventConnectionStart, amqpRequest, connectionMethodMap[10], connectionInfo, reader.GetCaptureTime(), reader.GetReadProgress().Current(), reader.GetEmitter(), reader.GetParent().GetOrigin())
 
 			case *ConnectionClose:
 				eventConnectionClose := &ConnectionClose{
@@ -206,8 +206,8 @@ func (d dissecting) Dissect(b *bufio.Reader, reader *api.TcpReader, options *sha
 					ClassId:   m.ClassId,
 					MethodId:  m.MethodId,
 				}
-				reader.Parent.CloseOtherProtocolDissectors(&protocol)
+				reader.GetParent().CloseOtherProtocolDissectors(&protocol)
-				emitAMQP(*eventConnectionClose, amqpRequest, connectionMethodMap[50], connectionInfo, reader.CaptureTime, reader.Progress.Current(), reader.Emitter, reader.Parent.Origin)
+				emitAMQP(*eventConnectionClose, amqpRequest, connectionMethodMap[50], connectionInfo, reader.GetCaptureTime(), reader.GetReadProgress().Current(), reader.GetEmitter(), reader.GetParent().GetOrigin())
 			}
 
 		default:

tap/extensions/amqp/main_test.go

@@ -108,7 +108,6 @@ func TestDissect(t *testing.T) {
 			Request:  0,
 			Response: 0,
 		}
-		protoIdentifier := &api.ProtoIdentifier{}
 
 		// Request
 		pathClient := _path
@@ -124,18 +123,21 @@ func TestDissect(t *testing.T) {
 			DstPort: "2",
 		}
 		reqResMatcher := dissector.NewResponseRequestMatcher()
-		reader := &api.TcpReader{
+		stream := api.NewTcpStreamDummy(api.Pcap)
-			Progress: &api.ReadProgress{},
+		reader := api.NewTcpReader(
-			Parent: &api.TcpStream{
+			make(chan api.TcpReaderDataMsg),
-				Origin:          api.Pcap,
+			&api.ReadProgress{},
-				ProtoIdentifier: protoIdentifier,
+			"",
-			},
+			tcpIDClient,
-			IsClient:      true,
+			time.Time{},
-			TcpID:         tcpIDClient,
+			stream,
-			Emitter:       emitter,
+			true,
-			CounterPair:   counterPair,
+			false,
-			ReqResMatcher: reqResMatcher,
+			nil,
-		}
+			emitter,
+			counterPair,
+			reqResMatcher,
+		)
 		err = dissector.Dissect(bufferClient, reader, options)
 		if err != nil && err != io.EOF && err != io.ErrUnexpectedEOF {
 			panic(err)
@@ -154,18 +156,20 @@ func TestDissect(t *testing.T) {
 			SrcPort: "2",
 			DstPort: "1",
 		}
-		reader = &api.TcpReader{
+		reader = api.NewTcpReader(
-			Progress: &api.ReadProgress{},
+			make(chan api.TcpReaderDataMsg),
-			Parent: &api.TcpStream{
+			&api.ReadProgress{},
-				Origin:          api.Pcap,
+			"",
-				ProtoIdentifier: protoIdentifier,
+			tcpIDServer,
-			},
+			time.Time{},
-			IsClient:      false,
+			stream,
-			TcpID:         tcpIDServer,
+			false,
-			Emitter:       emitter,
+			false,
-			CounterPair:   counterPair,
+			nil,
-			ReqResMatcher: reqResMatcher,
+			emitter,
-		}
+			counterPair,
+			reqResMatcher,
+		)
 		err = dissector.Dissect(bufferServer, reader, options)
 		if err != nil && err != io.EOF && err != io.ErrUnexpectedEOF {
 			panic(err)

tap/extensions/http/main.go

@@ -87,15 +87,15 @@ func (d dissecting) Ping() {
 	log.Printf("pong %s", http11protocol.Name)
 }
 
-func (d dissecting) Dissect(b *bufio.Reader, reader *api.TcpReader, options *shared.TrafficFilteringOptions) error {
+func (d dissecting) Dissect(b *bufio.Reader, reader api.TcpReader, options *shared.TrafficFilteringOptions) error {
-	reqResMatcher := reader.ReqResMatcher.(*requestResponseMatcher)
+	reqResMatcher := reader.GetReqResMatcher().(*requestResponseMatcher)
 
 	var err error
-	isHTTP2, _ := checkIsHTTP2Connection(b, reader.IsClient)
+	isHTTP2, _ := checkIsHTTP2Connection(b, reader.GetIsClient())
 
 	var http2Assembler *Http2Assembler
 	if isHTTP2 {
-		err = prepareHTTP2Connection(b, reader.IsClient)
+		err = prepareHTTP2Connection(b, reader.GetIsClient())
 		if err != nil {
 			return err
 		}
@@ -106,74 +106,74 @@ func (d dissecting) Dissect(b *bufio.Reader, reader *api.TcpReader, options *sha
 	for {
 		if switchingProtocolsHTTP2 {
 			switchingProtocolsHTTP2 = false
-			isHTTP2, err = checkIsHTTP2Connection(b, reader.IsClient)
+			isHTTP2, err = checkIsHTTP2Connection(b, reader.GetIsClient())
 			if err != nil {
 				break
 			}
-			err = prepareHTTP2Connection(b, reader.IsClient)
+			err = prepareHTTP2Connection(b, reader.GetIsClient())
 			if err != nil {
 				break
 			}
 			http2Assembler = createHTTP2Assembler(b)
 		}
 
-		if reader.Parent.ProtoIdentifier.Protocol != nil && reader.Parent.ProtoIdentifier.Protocol != &http11protocol {
+		if reader.GetParent().GetProtoIdentifier().Protocol != nil && reader.GetParent().GetProtoIdentifier().Protocol != &http11protocol {
 			return errors.New("Identified by another protocol")
 		}
 
 		if isHTTP2 {
-			err = handleHTTP2Stream(http2Assembler, reader.Progress, reader.Parent.Origin, reader.TcpID, reader.CaptureTime, reader.Emitter, options, reqResMatcher)
+			err = handleHTTP2Stream(http2Assembler, reader.GetReadProgress(), reader.GetParent().GetOrigin(), reader.GetTcpID(), reader.GetCaptureTime(), reader.GetEmitter(), options, reqResMatcher)
 			if err == io.EOF || err == io.ErrUnexpectedEOF {
 				break
 			} else if err != nil {
 				continue
 			}
-			reader.Parent.CloseOtherProtocolDissectors(&http11protocol)
+			reader.GetParent().CloseOtherProtocolDissectors(&http11protocol)
-		} else if reader.IsClient {
+		} else if reader.GetIsClient() {
 			var req *http.Request
-			switchingProtocolsHTTP2, req, err = handleHTTP1ClientStream(b, reader.Progress, reader.Parent.Origin, reader.TcpID, reader.CounterPair, reader.CaptureTime, reader.Emitter, options, reqResMatcher)
+			switchingProtocolsHTTP2, req, err = handleHTTP1ClientStream(b, reader.GetReadProgress(), reader.GetParent().GetOrigin(), reader.GetTcpID(), reader.GetCounterPair(), reader.GetCaptureTime(), reader.GetEmitter(), options, reqResMatcher)
 			if err == io.EOF || err == io.ErrUnexpectedEOF {
 				break
 			} else if err != nil {
 				continue
 			}
-			reader.Parent.CloseOtherProtocolDissectors(&http11protocol)
+			reader.GetParent().CloseOtherProtocolDissectors(&http11protocol)
 
 			// In case of an HTTP2 upgrade, duplicate the HTTP1 request into HTTP2 with stream ID 1
 			if switchingProtocolsHTTP2 {
 				ident := fmt.Sprintf(
 					"%s_%s_%s_%s_1_%s",
-					reader.TcpID.SrcIP,
+					reader.GetTcpID().SrcIP,
-					reader.TcpID.DstIP,
+					reader.GetTcpID().DstIP,
-					reader.TcpID.SrcPort,
+					reader.GetTcpID().SrcPort,
-					reader.TcpID.DstPort,
+					reader.GetTcpID().DstPort,
 					"HTTP2",
 				)
-				item := reqResMatcher.registerRequest(ident, req, reader.CaptureTime, reader.Progress.Current(), req.ProtoMinor)
+				item := reqResMatcher.registerRequest(ident, req, reader.GetCaptureTime(), reader.GetReadProgress().Current(), req.ProtoMinor)
 				if item != nil {
 					item.ConnectionInfo = &api.ConnectionInfo{
-						ClientIP:   reader.TcpID.SrcIP,
+						ClientIP:   reader.GetTcpID().SrcIP,
-						ClientPort: reader.TcpID.SrcPort,
+						ClientPort: reader.GetTcpID().SrcPort,
-						ServerIP:   reader.TcpID.DstIP,
+						ServerIP:   reader.GetTcpID().DstIP,
-						ServerPort: reader.TcpID.DstPort,
+						ServerPort: reader.GetTcpID().DstPort,
 						IsOutgoing: true,
 					}
-					item.Capture = reader.Parent.Origin
+					item.Capture = reader.GetParent().GetOrigin()
-					filterAndEmit(item, reader.Emitter, options)
+					filterAndEmit(item, reader.GetEmitter(), options)
 				}
 			}
 		} else {
-			switchingProtocolsHTTP2, err = handleHTTP1ServerStream(b, reader.Progress, reader.Parent.Origin, reader.TcpID, reader.CounterPair, reader.CaptureTime, reader.Emitter, options, reqResMatcher)
+			switchingProtocolsHTTP2, err = handleHTTP1ServerStream(b, reader.GetReadProgress(), reader.GetParent().GetOrigin(), reader.GetTcpID(), reader.GetCounterPair(), reader.GetCaptureTime(), reader.GetEmitter(), options, reqResMatcher)
 			if err == io.EOF || err == io.ErrUnexpectedEOF {
 				break
 			} else if err != nil {
 				continue
 			}
-			reader.Parent.CloseOtherProtocolDissectors(&http11protocol)
+			reader.GetParent().CloseOtherProtocolDissectors(&http11protocol)
 		}
 	}
 
-	if reader.Parent.ProtoIdentifier.Protocol == nil {
+	if reader.GetParent().GetProtoIdentifier().Protocol == nil {
 		return err
 	}
 

tap/extensions/http/main_test.go

@@ -110,7 +110,6 @@ func TestDissect(t *testing.T) {
 			Request:  0,
 			Response: 0,
 		}
-		protoIdentifier := &api.ProtoIdentifier{}
 
 		// Request
 		pathClient := _path
@@ -126,18 +125,21 @@ func TestDissect(t *testing.T) {
 			DstPort: "2",
 		}
 		reqResMatcher := dissector.NewResponseRequestMatcher()
-		reader := &api.TcpReader{
+		stream := api.NewTcpStreamDummy(api.Pcap)
-			Progress: &api.ReadProgress{},
+		reader := api.NewTcpReader(
-			Parent: &api.TcpStream{
+			make(chan api.TcpReaderDataMsg),
-				Origin:          api.Pcap,
+			&api.ReadProgress{},
-				ProtoIdentifier: protoIdentifier,
+			"",
-			},
+			tcpIDClient,
-			IsClient:      true,
+			time.Time{},
-			TcpID:         tcpIDClient,
+			stream,
-			Emitter:       emitter,
+			true,
-			CounterPair:   counterPair,
+			false,
-			ReqResMatcher: reqResMatcher,
+			nil,
-		}
+			emitter,
+			counterPair,
+			reqResMatcher,
+		)
 		err = dissector.Dissect(bufferClient, reader, options)
 		if err != nil && err != io.EOF && err != io.ErrUnexpectedEOF {
 			panic(err)
@@ -156,18 +158,20 @@ func TestDissect(t *testing.T) {
 			SrcPort: "2",
 			DstPort: "1",
 		}
-		reader = &api.TcpReader{
+		reader = api.NewTcpReader(
-			Progress: &api.ReadProgress{},
+			make(chan api.TcpReaderDataMsg),
-			Parent: &api.TcpStream{
+			&api.ReadProgress{},
-				Origin:          api.Pcap,
+			"",
-				ProtoIdentifier: protoIdentifier,
+			tcpIDServer,
-			},
+			time.Time{},
-			IsClient:      false,
+			stream,
-			TcpID:         tcpIDServer,
+			false,
-			Emitter:       emitter,
+			false,
-			CounterPair:   counterPair,
+			nil,
-			ReqResMatcher: reqResMatcher,
+			emitter,
-		}
+			counterPair,
+			reqResMatcher,
+		)
 		err = dissector.Dissect(bufferServer, reader, options)
 		if err != nil && err != io.EOF && err != io.ErrUnexpectedEOF {
 			panic(err)

tap/extensions/kafka/main.go

@@ -36,25 +36,25 @@ func (d dissecting) Ping() {
 	log.Printf("pong %s", _protocol.Name)
 }
 
-func (d dissecting) Dissect(b *bufio.Reader, reader *api.TcpReader, options *shared.TrafficFilteringOptions) error {
+func (d dissecting) Dissect(b *bufio.Reader, reader api.TcpReader, options *shared.TrafficFilteringOptions) error {
-	reqResMatcher := reader.ReqResMatcher.(*requestResponseMatcher)
+	reqResMatcher := reader.GetReqResMatcher().(*requestResponseMatcher)
 	for {
-		if reader.Parent.ProtoIdentifier.Protocol != nil && reader.Parent.ProtoIdentifier.Protocol != &_protocol {
+		if reader.GetParent().GetProtoIdentifier().Protocol != nil && reader.GetParent().GetProtoIdentifier().Protocol != &_protocol {
 			return errors.New("Identified by another protocol")
 		}
 
-		if reader.IsClient {
+		if reader.GetIsClient() {
-			_, _, err := ReadRequest(b, reader.TcpID, reader.CounterPair, reader.CaptureTime, reqResMatcher)
+			_, _, err := ReadRequest(b, reader.GetTcpID(), reader.GetCounterPair(), reader.GetCaptureTime(), reqResMatcher)
 			if err != nil {
 				return err
 			}
-			reader.Parent.CloseOtherProtocolDissectors(&_protocol)
+			reader.GetParent().CloseOtherProtocolDissectors(&_protocol)
 		} else {
-			err := ReadResponse(b, reader.Parent.Origin, reader.TcpID, reader.CounterPair, reader.CaptureTime, reader.Emitter, reqResMatcher)
+			err := ReadResponse(b, reader.GetParent().GetOrigin(), reader.GetTcpID(), reader.GetCounterPair(), reader.GetCaptureTime(), reader.GetEmitter(), reqResMatcher)
 			if err != nil {
 				return err
 			}
-			reader.Parent.CloseOtherProtocolDissectors(&_protocol)
+			reader.GetParent().CloseOtherProtocolDissectors(&_protocol)
 		}
 	}
 }

tap/extensions/kafka/main_test.go

@@ -108,7 +108,6 @@ func TestDissect(t *testing.T) {
 			Request:  0,
 			Response: 0,
 		}
-		protoIdentifier := &api.ProtoIdentifier{}
 
 		// Request
 		pathClient := _path
@@ -125,18 +124,21 @@ func TestDissect(t *testing.T) {
 		}
 		reqResMatcher := dissector.NewResponseRequestMatcher()
 		reqResMatcher.SetMaxTry(10)
-		reader := &api.TcpReader{
+		stream := api.NewTcpStreamDummy(api.Pcap)
-			Progress: &api.ReadProgress{},
+		reader := api.NewTcpReader(
-			Parent: &api.TcpStream{
+			make(chan api.TcpReaderDataMsg),
-				Origin:          api.Pcap,
+			&api.ReadProgress{},
-				ProtoIdentifier: protoIdentifier,
+			"",
-			},
+			tcpIDClient,
-			IsClient:      true,
+			time.Time{},
-			TcpID:         tcpIDClient,
+			stream,
-			Emitter:       emitter,
+			true,
-			CounterPair:   counterPair,
+			false,
-			ReqResMatcher: reqResMatcher,
+			nil,
-		}
+			emitter,
+			counterPair,
+			reqResMatcher,
+		)
 		err = dissector.Dissect(bufferClient, reader, options)
 		if err != nil && err != io.EOF && err != io.ErrUnexpectedEOF {
 			log.Println(err)
@@ -155,18 +157,20 @@ func TestDissect(t *testing.T) {
 			SrcPort: "2",
 			DstPort: "1",
 		}
-		reader = &api.TcpReader{
+		reader = api.NewTcpReader(
-			Progress: &api.ReadProgress{},
+			make(chan api.TcpReaderDataMsg),
-			Parent: &api.TcpStream{
+			&api.ReadProgress{},
-				Origin:          api.Pcap,
+			"",
-				ProtoIdentifier: protoIdentifier,
+			tcpIDServer,
-			},
+			time.Time{},
-			IsClient:      false,
+			stream,
-			TcpID:         tcpIDServer,
+			false,
-			Emitter:       emitter,
+			false,
-			CounterPair:   counterPair,
+			nil,
-			ReqResMatcher: reqResMatcher,
+			emitter,
-		}
+			counterPair,
+			reqResMatcher,
+		)
 		err = dissector.Dissect(bufferServer, reader, options)
 		if err != nil && err != io.EOF && err != io.ErrUnexpectedEOF {
 			log.Println(err)

tap/extensions/redis/main.go

@@ -35,8 +35,8 @@ func (d dissecting) Ping() {
 	log.Printf("pong %s", protocol.Name)
 }
 
-func (d dissecting) Dissect(b *bufio.Reader, reader *api.TcpReader, options *shared.TrafficFilteringOptions) error {
+func (d dissecting) Dissect(b *bufio.Reader, reader api.TcpReader, options *shared.TrafficFilteringOptions) error {
-	reqResMatcher := reader.ReqResMatcher.(*requestResponseMatcher)
+	reqResMatcher := reader.GetReqResMatcher().(*requestResponseMatcher)
 	is := &RedisInputStream{
 		Reader: b,
 		Buf:    make([]byte, 8192),
@@ -48,10 +48,10 @@ func (d dissecting) Dissect(b *bufio.Reader, reader *api.TcpReader, options *sha
 			return err
 		}
 
-		if reader.IsClient {
+		if reader.GetIsClient() {
-			err = handleClientStream(reader.Progress, reader.Parent.Origin, reader.TcpID, reader.CounterPair, reader.CaptureTime, reader.Emitter, redisPacket, reqResMatcher)
+			err = handleClientStream(reader.GetReadProgress(), reader.GetParent().GetOrigin(), reader.GetTcpID(), reader.GetCounterPair(), reader.GetCaptureTime(), reader.GetEmitter(), redisPacket, reqResMatcher)
 		} else {
-			err = handleServerStream(reader.Progress, reader.Parent.Origin, reader.TcpID, reader.CounterPair, reader.CaptureTime, reader.Emitter, redisPacket, reqResMatcher)
+			err = handleServerStream(reader.GetReadProgress(), reader.GetParent().GetOrigin(), reader.GetTcpID(), reader.GetCounterPair(), reader.GetCaptureTime(), reader.GetEmitter(), redisPacket, reqResMatcher)
 		}
 
 		if err != nil {

tap/extensions/redis/main_test.go

@@ -109,7 +109,6 @@ func TestDissect(t *testing.T) {
 			Request:  0,
 			Response: 0,
 		}
-		protoIdentifier := &api.ProtoIdentifier{}
 
 		// Request
 		pathClient := _path
@@ -125,18 +124,21 @@ func TestDissect(t *testing.T) {
 			DstPort: "2",
 		}
 		reqResMatcher := dissector.NewResponseRequestMatcher()
-		reader := &api.TcpReader{
+		stream := api.NewTcpStreamDummy(api.Pcap)
-			Progress: &api.ReadProgress{},
+		reader := api.NewTcpReader(
-			Parent: &api.TcpStream{
+			make(chan api.TcpReaderDataMsg),
-				Origin:          api.Pcap,
+			&api.ReadProgress{},
-				ProtoIdentifier: protoIdentifier,
+			"",
-			},
+			tcpIDClient,
-			IsClient:      true,
+			time.Time{},
-			TcpID:         tcpIDClient,
+			stream,
-			Emitter:       emitter,
+			true,
-			CounterPair:   counterPair,
+			false,
-			ReqResMatcher: reqResMatcher,
+			nil,
-		}
+			emitter,
+			counterPair,
+			reqResMatcher,
+		)
 		err = dissector.Dissect(bufferClient, reader, options)
 		if err != nil && reflect.TypeOf(err) != reflect.TypeOf(&ConnectError{}) && err != io.EOF && err != io.ErrUnexpectedEOF {
 			log.Println(err)
@@ -155,18 +157,20 @@ func TestDissect(t *testing.T) {
 			SrcPort: "2",
 			DstPort: "1",
 		}
-		reader = &api.TcpReader{
+		reader = api.NewTcpReader(
-			Progress: &api.ReadProgress{},
+			make(chan api.TcpReaderDataMsg),
-			Parent: &api.TcpStream{
+			&api.ReadProgress{},
-				Origin:          api.Pcap,
+			"",
-				ProtoIdentifier: protoIdentifier,
+			tcpIDServer,
-			},
+			time.Time{},
-			IsClient:      false,
+			stream,
-			TcpID:         tcpIDServer,
+			false,
-			Emitter:       emitter,
+			false,
-			CounterPair:   counterPair,
+			nil,
-			ReqResMatcher: reqResMatcher,
+			emitter,
-		}
+			counterPair,
+			reqResMatcher,
+		)
 		err = dissector.Dissect(bufferServer, reader, options)
 		if err != nil && reflect.TypeOf(err) != reflect.TypeOf(&ConnectError{}) && err != io.EOF && err != io.ErrUnexpectedEOF {
 			log.Println(err)

tap/passive_tapper.go

@@ -179,7 +179,7 @@ func initializePacketSources() error {
 	return err
 }
 
-func initializePassiveTapper(opts *TapOpts, outputItems chan *api.OutputChannelItem) (*api.TcpStreamMap, *tcpAssembler) {
+func initializePassiveTapper(opts *TapOpts, outputItems chan *api.OutputChannelItem) (api.TcpStreamMap, *tcpAssembler) {
 	streamsMap := api.NewTcpStreamMap()
 
 	diagnose.InitializeErrorsMap(*debug, *verbose, *quiet)
@@ -196,7 +196,7 @@ func initializePassiveTapper(opts *TapOpts, outputItems chan *api.OutputChannelI
 	return streamsMap, assembler
 }
 
-func startPassiveTapper(streamsMap *api.TcpStreamMap, assembler *tcpAssembler) {
+func startPassiveTapper(streamsMap api.TcpStreamMap, assembler *tcpAssembler) {
 	go streamsMap.CloseTimedoutTcpStreamChannels()
 
 	diagnose.AppStatsInst.SetStartTime(time.Now())

tap/tcp_assembler.go

@@ -36,7 +36,7 @@ func (c *context) GetCaptureInfo() gopacket.CaptureInfo {
 	return c.CaptureInfo
 }
 
-func NewTcpAssembler(outputItems chan *api.OutputChannelItem, streamsMap *api.TcpStreamMap, opts *TapOpts) *tcpAssembler {
+func NewTcpAssembler(outputItems chan *api.OutputChannelItem, streamsMap api.TcpStreamMap, opts *TapOpts) *tcpAssembler {
 	var emitter api.Emitter = &api.Emitting{
 		AppStats:      &diagnose.AppStatsInst,
 		OutputChannel: outputItems,

tap/tcp_stream_factory.go

@@ -3,6 +3,7 @@ package tap
 import (
 	"fmt"
 	"sync"
+	"time"
 
 	"github.com/up9inc/mizu/shared/logger"
 	"github.com/up9inc/mizu/tap/api"
@@ -20,13 +21,13 @@ import (
  */
 type tcpStreamFactory struct {
 	wg         sync.WaitGroup
-	Emitter    api.Emitter
+	emitter    api.Emitter
-	streamsMap *api.TcpStreamMap
+	streamsMap api.TcpStreamMap
 	ownIps     []string
 	opts       *TapOpts
 }
 
-func NewTcpStreamFactory(emitter api.Emitter, streamsMap *api.TcpStreamMap, opts *TapOpts) *tcpStreamFactory {
+func NewTcpStreamFactory(emitter api.Emitter, streamsMap api.TcpStreamMap, opts *TapOpts) *tcpStreamFactory {
 	var ownIps []string
 
 	if localhostIPs, err := getLocalhostIPs(); err != nil {
@@ -39,7 +40,7 @@ func NewTcpStreamFactory(emitter api.Emitter, streamsMap *api.TcpStreamMap, opts
 	}
 
 	return &tcpStreamFactory{
-		Emitter:    emitter,
+		emitter:    emitter,
 		streamsMap: streamsMap,
 		ownIps:     ownIps,
 		opts:       opts,
@@ -57,69 +58,64 @@ func (factory *tcpStreamFactory) New(net, transport gopacket.Flow, tcp *layers.T
 
 	props := factory.getStreamProps(srcIp, srcPort, dstIp, dstPort)
 	isTapTarget := props.isTapTarget
-	stream := &api.TcpStream{
+	stream := api.NewTcpStream(net, transport, tcp, isTapTarget, fsmOptions, factory.streamsMap, getPacketOrigin(ac))
-		Net:             net,
+	if stream.GetIsTapTarget() {
-		Transport:       transport,
+		stream.SetId(factory.streamsMap.NextId())
-		IsDNS:           tcp.SrcPort == 53 || tcp.DstPort == 53,
-		IsTapTarget:     isTapTarget,
-		TcpState:        reassembly.NewTCPSimpleFSM(fsmOptions),
-		Ident:           fmt.Sprintf("%s:%s", net, transport),
-		Optchecker:      reassembly.NewTCPOptionCheck(),
-		ProtoIdentifier: &api.ProtoIdentifier{},
-		StreamsMap:      factory.streamsMap,
-		Origin:          getPacketOrigin(ac),
-	}
-	if stream.IsTapTarget {
-		stream.Id = factory.streamsMap.NextId()
 		for i, extension := range extensions {
 			reqResMatcher := extension.Dissector.NewResponseRequestMatcher()
 			counterPair := &api.CounterPair{
 				Request:  0,
 				Response: 0,
 			}
-			stream.Clients = append(stream.Clients, api.TcpReader{
+			stream.AddClient(
-				MsgQueue: make(chan api.TcpReaderDataMsg),
+				api.NewTcpReader(
-				Progress: &api.ReadProgress{},
+					make(chan api.TcpReaderDataMsg),
-				Ident:    fmt.Sprintf("%s %s", net, transport),
+					&api.ReadProgress{},
-				TcpID: &api.TcpID{
+					fmt.Sprintf("%s %s", net, transport),
-					SrcIP:   srcIp,
+					&api.TcpID{
-					DstIP:   dstIp,
+						SrcIP:   srcIp,
-					SrcPort: srcPort,
+						DstIP:   dstIp,
-					DstPort: dstPort,
+						SrcPort: srcPort,
-				},
+						DstPort: dstPort,
-				Parent:        stream,
+					},
-				IsClient:      true,
+					time.Time{},
-				IsOutgoing:    props.isOutgoing,
+					stream,
-				Extension:     extension,
+					true,
-				Emitter:       factory.Emitter,
+					props.isOutgoing,
-				CounterPair:   counterPair,
+					extension,
-				ReqResMatcher: reqResMatcher,
+					factory.emitter,
-			})
+					counterPair,
-			stream.Servers = append(stream.Servers, api.TcpReader{
+					reqResMatcher,
-				MsgQueue: make(chan api.TcpReaderDataMsg),
+				),
-				Progress: &api.ReadProgress{},
+			)
-				Ident:    fmt.Sprintf("%s %s", net, transport),
+			stream.AddServer(
-				TcpID: &api.TcpID{
+				api.NewTcpReader(
-					SrcIP:   net.Dst().String(),
+					make(chan api.TcpReaderDataMsg),
-					DstIP:   net.Src().String(),
+					&api.ReadProgress{},
-					SrcPort: transport.Dst().String(),
+					fmt.Sprintf("%s %s", net, transport),
-					DstPort: transport.Src().String(),
+					&api.TcpID{
-				},
+						SrcIP:   net.Dst().String(),
-				Parent:        stream,
+						DstIP:   net.Src().String(),
-				IsClient:      false,
+						SrcPort: transport.Dst().String(),
-				IsOutgoing:    props.isOutgoing,
+						DstPort: transport.Src().String(),
-				Extension:     extension,
+					},
-				Emitter:       factory.Emitter,
+					time.Time{},
-				CounterPair:   counterPair,
+					stream,
-				ReqResMatcher: reqResMatcher,
+					false,
-			})
+					props.isOutgoing,
+					extension,
+					factory.emitter,
+					counterPair,
+					reqResMatcher,
+				),
+			)
 
-			factory.streamsMap.Store(stream.Id, stream)
+			factory.streamsMap.Store(stream.GetId(), stream)
 
 			factory.wg.Add(2)
 			// Start reading from channel stream.reader.bytes
-			go stream.Clients[i].Run(filteringOptions, &factory.wg)
+			go stream.ClientRun(i, filteringOptions, &factory.wg)
-			go stream.Servers[i].Run(filteringOptions, &factory.wg)
+			go stream.ServerRun(i, filteringOptions, &factory.wg)
 		}
 	}
 	return stream

tap/tlstapper/tls_poller.go

@@ -167,18 +167,21 @@ func dissect(extension *api.Extension, reader *tlsReader, isRequest bool, tcpid
 	tlsEmitter *tlsEmitter, options *shared.TrafficFilteringOptions, reqResMatcher api.RequestResponseMatcher) {
 	b := bufio.NewReader(reader)
 
-	tcpReader := &api.TcpReader{
+	tcpStream := api.NewTcpStreamDummy(api.Ebpf)
-		Progress:    reader.progress,
+	tcpReader := api.NewTcpReader(
-		CaptureTime: time.Now(),
+		make(chan api.TcpReaderDataMsg),
-		Parent: &api.TcpStream{
+		reader.progress,
-			Origin:          api.Ebpf,
+		"",
-			ProtoIdentifier: &api.ProtoIdentifier{},
+		tcpid,
-		},
+		time.Now(),
-		IsClient:      isRequest,
+		tcpStream,
-		TcpID:         tcpid,
+		isRequest,
-		Emitter:       tlsEmitter,
+		false,
-		ReqResMatcher: reqResMatcher,
+		nil,
-	}
+		tlsEmitter,
+		&api.CounterPair{},
+		reqResMatcher,
+	)
 
 	err := extension.Dissector.Dissect(b, tcpReader, options)