func (d *EndpointPayloadDetector) detectByFlow(payload []byte, flow gopacket.Flow) (layer gopacket.LayerType, err error) {
	var ok bool
	if layer, ok = d.srcEndpointMap[flow.Src()]; ok {
		return
	}
	if layer, ok = d.dstEndpointMap[flow.Dst()]; ok {
		return
	}
	return gopacket.LayerTypeZero, EndpointPayloadDetectorFailedError
}

func (http MySql) New(netFlow, tcpFlow gopacket.Flow) (ret tcpassembly.Stream) {
	destPort, _ := strconv.Atoi(fmt.Sprintf("%v", tcpFlow.Dst()))

	//
	if isSupport(destPort) {
		return NewMySqlStream()
	}

	//
	return nil
}

// New is the factory method called by the tcpassembly to generate new tcpassembly.Stream.
func (os *MongoOpStream) New(netFlow, tcpFlow gopacket.Flow) tcpassembly.Stream {
	key := bidiKey{netFlow, tcpFlow}
	rkey := bidiKey{netFlow.Reverse(), tcpFlow.Reverse()}
	if bidi, ok := os.bidiMap[key]; ok {
		atomic.AddInt32(&bidi.openStreamCount, 1)
		delete(os.bidiMap, key)
		return bidi.streams[1]
	}
	bidi := newBidi(netFlow, tcpFlow, os, <-os.connectionCounter)
	os.bidiMap[rkey] = bidi
	atomic.AddInt32(&bidi.openStreamCount, 1)
	go bidi.streamOps()
	return bidi.streams[0]
}

// New handles creating a new tcpassembly.Stream.
func (f *myFactory) New(netFlow, tcpFlow gopacket.Flow) tcpassembly.Stream {
	// Create a new stream.
	s := &myStream{}

	// Find the bidi bidirectional struct for this stream, creating a new one if
	// one doesn't already exist in the map.
	k := key{netFlow, tcpFlow}
	bd := f.bidiMap[k]
	if bd == nil {
		bd = &bidi{a: s, key: k}
		// Register bidirectional with the reverse key, so the matching stream going
		// the other direction will find it.
		f.bidiMap[key{netFlow.Reverse(), tcpFlow.Reverse()}] = bd
	} else {
		bd.b = s
		// Clear out the bidi we're using from the map, just in case.
		delete(f.bidiMap, k)
	}
	s.bidi = bd
	return s
}

func newBidi(netFlow, tcpFlow gopacket.Flow, opStream *MongoOpStream, num int64) *bidi {
	bidi := &bidi{connectionNumber: num}
	bidi.streams[0] = &stream{
		bidi:        bidi,
		reassembled: make(chan []tcpassembly.Reassembly),
		done:        make(chan interface{}),
		op:          &RawOp{},
		netFlow:     netFlow,
		tcpFlow:     tcpFlow,
	}
	bidi.streams[1] = &stream{
		bidi:        bidi,
		reassembled: make(chan []tcpassembly.Reassembly),
		done:        make(chan interface{}),
		op:          &RawOp{},
		netFlow:     netFlow.Reverse(),
		tcpFlow:     tcpFlow.Reverse(),
	}
	bidi.opStream = opStream
	return bidi
}

func (assembler *TcpAssembler) assemble(flow gopacket.Flow, tcp *layers.TCP, timestamp time.Time) {
	src := EndPoint{ip: flow.Src().String(), port: uint16(tcp.SrcPort)}
	dst := EndPoint{ip: flow.Dst().String(), port: uint16(tcp.DstPort)}
	dropped := false
	if assembler.filterIp != "" {
		if src.ip != assembler.filterIp && dst.ip != assembler.filterIp {
			dropped = true
		}
	}
	if assembler.filterPort != 0 {
		if src.port != assembler.filterPort && dst.port != assembler.filterPort {
			dropped = true
		}
	}
	if dropped {
		return
	}

	srcString := src.String()
	dstString := dst.String()
	var key string
	if srcString < dstString {
		key = srcString + "-" + dstString
	} else {
		key = dstString + "-" + srcString
	}

	var createNewConn = tcp.SYN && !tcp.ACK || isHttpRequestData(tcp.Payload)
	connection := assembler.retrieveConnection(src, dst, key, createNewConn)
	if connection == nil {
		return
	}

	connection.onReceive(src, dst, tcp, timestamp)

	if connection.closed() {
		assembler.deleteConnection(key)
		connection.finish()
	}
}

func (f HttpStreamFactory) New(netFlow, tcpFlow gopacket.Flow) (ret tcpassembly.Stream) {
	revkey := fmt.Sprintf("%v:%v->%v:%v",
		netFlow.Dst(),
		tcpFlow.Dst(),
		netFlow.Src(),
		tcpFlow.Src())
	streamPair, ok := (*f.uniStreams)[revkey]
	src := make(PacketSource)
	if ok {
		if streamPair.upStream == nil {
			panic("unbelievable!?")
		}
		delete(*f.uniStreams, revkey)
		s := NewHttpStream(src, "", streamPair.connSeq, f.wg, *f.eventChan, streamPair.sem, DownDirection)
		streamPair.downStream = &s
		ret = s
		go s.Process()
	} else {
		streamPair = new(HttpStreamPair)
		streamPair.connSeq = *f.seq
		streamPair.sem = make(chan byte, 1)
		key := fmt.Sprintf("%v:%v->%v:%v",
			netFlow.Src(),
			tcpFlow.Src(),
			netFlow.Dst(),
			tcpFlow.Dst())
		s := NewHttpStream(src, "", streamPair.connSeq, f.wg, *f.eventChan, streamPair.sem, UpDirection)
		streamPair.upStream = &s
		(*f.uniStreams)[key] = streamPair
		*f.seq++
		go s.Process()
		ret = s
	}
	f.wg.Add(1)
	return
}

func (s *MongoOpStream) New(a, b gopacket.Flow) tcpassembly.Stream {
	r := tcpreader.NewReaderStream()
	log.Println("starting stream", a, b)
	go s.handleStream(&r, b.String())
	return &r
}

func (http Http) New(netFlow, tcpFlow gopacket.Flow) (ret tcpassembly.Stream) {
	destPort, _ := strconv.Atoi(fmt.Sprintf("%v", tcpFlow.Dst()))
	srcPort, _ := strconv.Atoi(fmt.Sprintf("%v", tcpFlow.Src()))

	//

	//
	var key string = ""
	var isUp bool
	if isSupport(destPort) {
		isUp = true
		key = fmt.Sprintf("%v:%v-%v:%v", netFlow.Src(), tcpFlow.Src(), netFlow.Dst(), tcpFlow.Dst())
	} else if isSupport(srcPort) {
		isUp = false
		key = fmt.Sprintf("%v:%v-%v:%v", netFlow.Dst(), tcpFlow.Dst(), netFlow.Src(), tcpFlow.Src())
	}

	//
	if key != "" {
		logH.Debug("%v", netFlow)
		httpData, ok := dataMap[key]
		if !ok {
			httpData = NewHttpData()
			dataMap[key] = httpData
		}

		//請求
		if isUp {
			stream := NewHttpRequestStream(httpData.wg)
			httpData.requestStream = stream
			return &stream.reader

			//响应
		} else if httpData.requestStream != nil {
			stream := NewHttpResponseStream(httpData.requestStream, httpData.wg)
			httpData.responseStream = stream

			//
			delete(dataMap, key)
			return &stream.reader
		}
	}

	//
	return nil
}