func (dns *dnsPlugin) GapInStream(tcpTuple *common.TCPTuple, dir uint8, nbytes int, private protos.ProtocolData) (priv protos.ProtocolData, drop bool) {
	if private == nil {
		return private, true
	}
	conn, ok := private.(*dnsConnectionData)
	if !ok {
		return private, false
	}
	stream := conn.data[dir]

	if stream == nil || stream.message == nil {
		return private, false
	}

	decodedData, err := stream.handleTCPRawData()

	if err == nil {
		dns.messageComplete(conn, tcpTuple, dir, decodedData)
		return private, true
	}

	if dir == tcp.TCPDirectionReverse {
		dns.publishResponseError(conn, err)
	}

	debugf("%s addresses %s, length %d", err.Error(),
		tcpTuple.String(), len(stream.rawData))
	debugf("Dropping the stream %s", tcpTuple.String())

	// drop the stream because it is binary Data and it would be unexpected to have a decodable message later
	return private, true
}

func (http *httpPlugin) handleHTTP(
	conn *httpConnectionData,
	m *message,
	tcptuple *common.TCPTuple,
	dir uint8,
) {

	m.tcpTuple = *tcptuple
	m.direction = dir
	m.cmdlineTuple = procs.ProcWatcher.FindProcessesTuple(tcptuple.IPPort())
	http.hideHeaders(m)

	if m.isRequest {
		if isDebug {
			debugf("Received request with tuple: %s", m.tcpTuple)
		}
		conn.requests.append(m)
	} else {
		if isDebug {
			debugf("Received response with tuple: %s", m.tcpTuple)
		}
		conn.responses.append(m)
		http.correlate(conn)
	}
}

func (dns *dnsPlugin) ReceivedFin(tcpTuple *common.TCPTuple, dir uint8, private protos.ProtocolData) protos.ProtocolData {
	if private == nil {
		return nil
	}
	conn, ok := private.(*dnsConnectionData)
	if !ok {
		return private
	}
	stream := conn.data[dir]

	if stream == nil || stream.message == nil {
		return conn
	}

	decodedData, err := stream.handleTCPRawData()

	if err == nil {
		dns.messageComplete(conn, tcpTuple, dir, decodedData)
		return conn
	}

	if dir == tcp.TCPDirectionReverse {
		dns.publishResponseError(conn, err)
	}

	debugf("%s addresses %s, length %d", err.Error(),
		tcpTuple.String(), len(stream.rawData))

	return conn
}

func (pgsql *Pgsql) removeTransaction(transList []*PgsqlTransaction,
	tuple common.TCPTuple, index int) *PgsqlTransaction {

	trans := transList[index]
	transList = append(transList[:index], transList[index+1:]...)
	if len(transList) == 0 {
		pgsql.transactions.Delete(trans.tuple.Hashable())
	} else {
		pgsql.transactions.Put(tuple.Hashable(), transList)
	}

	return trans
}

func (thrift *thriftPlugin) ReceivedFin(tcptuple *common.TCPTuple, dir uint8,
	private protos.ProtocolData) protos.ProtocolData {

	trans := thrift.getTransaction(tcptuple.Hashable())
	if trans != nil {
		if trans.request != nil && trans.reply == nil {
			logp.Debug("thrift", "FIN and had only one transaction. Assuming one way")
			thrift.publishQueue <- trans
			thrift.transactions.Delete(trans.tuple.Hashable())
		}
	}

	return private
}

func handleMysql(mysql *mysqlPlugin, m *mysqlMessage, tcptuple *common.TCPTuple,
	dir uint8, rawMsg []byte) {

	m.tcpTuple = *tcptuple
	m.direction = dir
	m.cmdlineTuple = procs.ProcWatcher.FindProcessesTuple(tcptuple.IPPort())
	m.raw = rawMsg

	if m.isRequest {
		mysql.receivedMysqlRequest(m)
	} else {
		mysql.receivedMysqlResponse(m)
	}
}

func handleMysql(mysql *Mysql, m *MysqlMessage, tcptuple *common.TCPTuple,
	dir uint8, raw_msg []byte) {

	m.TcpTuple = *tcptuple
	m.Direction = dir
	m.CmdlineTuple = procs.ProcWatcher.FindProcessesTuple(tcptuple.IPPort())
	m.Raw = raw_msg

	if m.IsRequest {
		mysql.receivedMysqlRequest(m)
	} else {
		mysql.receivedMysqlResponse(m)
	}
}

func (dns *dnsPlugin) handleDNS(conn *dnsConnectionData, tcpTuple *common.TCPTuple, decodedData *mkdns.Msg, dir uint8) {
	message := conn.data[dir].message
	dnsTuple := dnsTupleFromIPPort(&message.tuple, transportTCP, decodedData.Id)

	message.cmdlineTuple = procs.ProcWatcher.FindProcessesTuple(tcpTuple.IPPort())
	message.data = decodedData
	message.length += decodeOffset

	if decodedData.Response {
		dns.receivedDNSResponse(&dnsTuple, message)
		conn.prevRequest = nil
	} else /* Query */ {
		dns.receivedDNSRequest(&dnsTuple, message)
		conn.prevRequest = message
	}
}

func (redis *Redis) handleRedis(
	conn *redisConnectionData,
	m *redisMessage,
	tcptuple *common.TCPTuple,
	dir uint8,
) {
	m.TcpTuple = *tcptuple
	m.Direction = dir
	m.CmdlineTuple = procs.ProcWatcher.FindProcessesTuple(tcptuple.IPPort())

	if m.IsRequest {
		conn.requests.append(m) // wait for response
	} else {
		conn.responses.append(m)
		redis.correlate(conn)
	}
}

func (dns *dnsPlugin) doParse(conn *dnsConnectionData, pkt *protos.Packet, tcpTuple *common.TCPTuple, dir uint8) *dnsConnectionData {
	stream := conn.data[dir]
	payload := pkt.Payload

	if stream == nil {
		stream = newStream(pkt, tcpTuple)
		conn.data[dir] = stream
	} else {
		if stream.message == nil { // nth message of the same stream
			stream.message = &dnsMessage{ts: pkt.Ts, tuple: pkt.Tuple}
		}

		stream.rawData = append(stream.rawData, payload...)
		if len(stream.rawData) > tcp.TCPMaxDataInStream {
			debugf("Stream data too large, dropping DNS stream")
			conn.data[dir] = nil
			return conn
		}
	}
	decodedData, err := stream.handleTCPRawData()

	if err != nil {

		if err == incompleteMsg {
			debugf("Waiting for more raw data")
			return conn
		}

		if dir == tcp.TCPDirectionReverse {
			dns.publishResponseError(conn, err)
		}

		debugf("%s addresses %s, length %d", err.Error(),
			tcpTuple.String(), len(stream.rawData))

		// This means that malformed requests or responses are being sent...
		// TODO: publish the situation also if Request
		conn.data[dir] = nil
		return conn
	}

	dns.messageComplete(conn, tcpTuple, dir, decodedData)
	stream.prepareForNewMessage()
	return conn
}

func (mongodb *mongodbPlugin) handleMongodb(
	conn *mongodbConnectionData,
	m *mongodbMessage,
	tcptuple *common.TCPTuple,
	dir uint8,
) {

	m.tcpTuple = *tcptuple
	m.direction = dir
	m.cmdlineTuple = procs.ProcWatcher.FindProcessesTuple(tcptuple.IPPort())

	if m.isResponse {
		debugf("MongoDB response message")
		mongodb.onResponse(conn, m)
	} else {
		debugf("MongoDB request message")
		mongodb.onRequest(conn, m)
	}
}

func (amqp *amqpPlugin) handleAmqp(m *amqpMessage, tcptuple *common.TCPTuple, dir uint8) {
	if amqp.mustHideCloseMethod(m) {
		return
	}
	debugf("A message is ready to be handled")
	m.tcpTuple = *tcptuple
	m.direction = dir
	m.cmdlineTuple = procs.ProcWatcher.FindProcessesTuple(tcptuple.IPPort())

	if m.method == "basic.publish" {
		amqp.handlePublishing(m)
	} else if m.method == "basic.deliver" || m.method == "basic.return" ||
		m.method == "basic.get-ok" {
		amqp.handleDelivering(m)
	} else if m.isRequest == true {
		amqp.handleAmqpRequest(m)
	} else if m.isRequest == false {
		amqp.handleAmqpResponse(m)
	}
}

func (thrift *thriftPlugin) messageComplete(tcptuple *common.TCPTuple, dir uint8,
	stream *thriftStream, priv *thriftPrivateData) {

	flush := false

	if stream.message.isRequest {
		logp.Debug("thrift", "Thrift request message: %s", stream.message.method)
		if !thrift.captureReply {
			// enable the stream in the other direction to get the reply
			streamRev := priv.data[1-dir]
			if streamRev != nil {
				streamRev.skipInput = false
			}
		}
	} else {
		logp.Debug("thrift", "Thrift response message: %s", stream.message.method)
		if !thrift.captureReply {
			// disable stream in this direction
			stream.skipInput = true

			// and flush current data
			flush = true
		}
	}

	// all ok, go to next level
	stream.message.tcpTuple = *tcptuple
	stream.message.direction = dir
	stream.message.cmdlineTuple = procs.ProcWatcher.FindProcessesTuple(tcptuple.IPPort())
	if stream.message.frameSize == 0 {
		stream.message.frameSize = uint32(stream.parseOffset - stream.message.start)
	}
	thrift.handleThrift(stream.message)

	// and reset message
	stream.prepareForNewMessage(flush)

}

// Parse processes a TCP packet. Return nil if connection
// state shall be dropped (e.g. parser not in sync with tcp stream)
func (cassandra *cassandra) Parse(
	pkt *protos.Packet,
	tcptuple *common.TCPTuple, dir uint8,
	private protos.ProtocolData,
) protos.ProtocolData {
	defer logp.Recover("Parse cassandra exception")

	conn := cassandra.ensureConnection(private)
	st := conn.streams[dir]
	if st == nil {
		st = &stream{}
		st.parser.init(&cassandra.parserConfig, func(msg *message) error {
			return conn.trans.onMessage(tcptuple.IPPort(), dir, msg)
		})
		conn.streams[dir] = st
	}

	if err := st.parser.feed(pkt.Ts, pkt.Payload); err != nil {
		debugf("%v, dropping TCP stream for error in direction %v.", err, dir)
		cassandra.onDropConnection(conn)
		return nil
	}
	return conn
}

func (mc *Memcache) memcacheParseTCP(
	tcpConn *tcpConnectionData,
	pkt *protos.Packet,
	tcptuple *common.TCPTuple,
	dir uint8,
) *tcpConnectionData {
	// assume we are in sync
	stream := tcpConn.Streams[dir]
	if stream == nil {
		stream = mc.newStream(tcptuple)
		tcpConn.Streams[dir] = stream
	}

	debug("add payload to stream(%p): %v", stream, dir)
	if err := stream.Append(pkt.Payload); err != nil {
		debug("%v, dropping TCP streams", err)
		mc.pushAllTCPTrans(tcpConn.connection)
		tcpConn.drop(dir)
		return nil
	}

	if tcpConn.connection == nil {
		tcpConn.connection = &connection{}
	} else {
		stopped := tcpConn.connection.timer.Stop()
		if !stopped {
			// timer was stopped by someone else, create new connection
			tcpConn.connection = &connection{}
		}
	}
	conn := tcpConn.connection

	for stream.Buf.Total() > 0 {
		debug("stream(%p) try to content", stream)
		msg, err := stream.parse(pkt.Ts)
		if err != nil {
			// parsing error, drop tcp stream and retry with next segement
			debug("Ignore Memcache message, drop tcp stream: %v", err)
			mc.pushAllTCPTrans(conn)
			tcpConn.drop(dir)
			return nil
		}

		if msg == nil {
			// wait for more data
			break
		}
		stream.reset()

		tuple := tcptuple.IPPort()
		err = mc.onTCPMessage(conn, tuple, dir, msg)
		if err != nil {
			logp.Warn("error processing memcache message: %s", err)
		}
	}

	conn.timer = time.AfterFunc(mc.tcpTransTimeout, func() {
		debug("connection=%p timed out", conn)
		mc.pushAllTCPTrans(conn)
	})

	return tcpConn
}