frame, err := sc.protocol.Decode(streamCtx, buf)
    // 如果没有足够数据，则直接返回，若是凑够了一个frame ，则handleFrame
    handleFrame(streamCtx, xframe)

// mosn.io/mosn/pkg/proxy/proxy.go
func (p *proxy) InitializeReadFilterCallbacks(cb api.ReadFilterCallbacks) {
    p.readCallbacks = cb
    ...
	p.readCallbacks.Connection().AddConnectionEventListener(p.downstreamListener)
	if p.config.DownstreamProtocol != string(protocol.Auto) {
        // 创建好的ServerStreamConnection 赋给proxy  
        // proxy.config.DownstreamProtocol = X 即 Xprotocol
		p.serverStreamConn = stream.CreateServerStreamConnection(p.context, types.ProtocolName(p.config.DownstreamProtocol), p.readCallbacks.Connection(), p)
	}
}
// mosn.io/mosn/pkg/stream/factory.go
var streamFactories map[types.ProtocolName]ProtocolStreamFactory
func init() {
	streamFactories = make(map[types.ProtocolName]ProtocolStreamFactory)
}
func Register(prot types.ProtocolName, factory ProtocolStreamFactory) {
	streamFactories[prot] = factory
}
func CreateServerStreamConnection(context context.Context, prot api.Protocol, connection api.Connection,
	callbacks types.ServerStreamConnectionEventListener) types.ServerStreamConnection {
	if ssc, ok := streamFactories[prot]; ok {
		return ssc.CreateServerStream(context, connection, callbacks)
	}
	return nil
}

// mosn.io/mosn/pkg/stream/xprotocol/factory.go
func (f *streamConnFactory) CreateServerStream(context context.Context, connection api.Connection,
	serverCallbacks types.ServerStreamConnectionEventListener) types.ServerStreamConnection {
	return newStreamConnection(context, connection, nil, serverCallbacks)
}
// mosn.io/mosn/pkg/stream/xprotocol/conn.go
func newStreamConnection(ctx context.Context, conn api.Connection, clientCallbacks types.StreamConnectionEventListener,
	serverCallbacks types.ServerStreamConnectionEventListener) types.ClientStreamConnection {
	// 1. init first context
	// 2. prepare protocols
	subProtocol := mosnctx.Get(ctx, types.ContextSubProtocol).(string)
	subProtocols := strings.Split(subProtocol, ",")
	// 2.1 exact protocol, get directly
	// 2.2 multi protocol, setup engine for further match 
	if len(subProtocols) == 1 {
		proto := xprotocol.GetProtocol(types.ProtocolName(subProtocol))
		if proto == nil {
			log.Proxy.Errorf(ctx, "[stream] [xprotocol] no such protocol: %s", subProtocol)
			return nil
		}
		sc.protocol = proto
	} else {   // mosn 支持多个subProtocol ，估计是可以根据 连接中的信息自动匹配
		engine, err := xprotocol.NewXEngine(subProtocols)
		if err != nil {
			log.Proxy.Errorf(ctx, "[stream] [xprotocol] create XEngine failed: %s", err)
			return nil
		}
		sc.engine = engine
	}
	// client
	if sc.clientCallbacks != nil {
		// default client concurrency capacity: 8
		sc.clientStreams = make(map[uint64]*xStream, 8) // Stream 抽象了一般rpc 框架维护的 <requestId,Request>
	}
	// set support transfer connection
	return sc
}
// mosn.io/mosn/pkg/protocol/xprotocol/factory.go
var (
	protocolMap = make(map[types.ProtocolName]XProtocol)
	matcherMap  = make(map[types.ProtocolName]types.ProtocolMatch)
)
// RegisterProtocol register the protocol to factory
func RegisterProtocol(name types.ProtocolName, protocol XProtocol) error {
	// check name conflict
	_, ok := protocolMap[name]
	if ok {
		return errors.New("duplicate protocol register:" + string(name))
	}
	protocolMap[name] = protocol
	return nil
}
func GetProtocol(name types.ProtocolName) XProtocol {
	return protocolMap[name]
}

// mosn.io/mosn/pkg/stream/xprotocol/conn.go
type streamConn struct {
    ctx        context.Context
    ctxManager *stream.ContextManager

	netConn    api.Connection               // 底层连接
	engine   *xprotocol.XEngine             // xprotocol fields
	protocol xprotocol.XProtocol
    // server side fields
	serverCallbacks types.ServerStreamConnectionEventListener 
    // client side fields
	clientMutex        sync.RWMutex                           
	clientStreamIDBase uint64
	clientStreams      map[uint64]*xStream
	clientCallbacks    types.StreamConnectionEventListener
}

// mosn.io/mosn/pkg/types/buffer.go
type BufferPoolCtx interface {
	// Index returns the bufferpool's Index
	Index() int
	New() interface{}           // new 一个元素
	Reset(interface{})          // reset 一个元素  放回到buffer 中的元素要 reset 到一种clean 的状态
}
// mosn.io/mosn/pkg/buffer/buffer.go
// bufferPool is buffer pool
type bufferPool struct {
	ctx types.BufferPoolCtx
	sync.Pool
}
// 取出一个元素，有则取出，无则new 一个
func (p *bufferPool) take() (value interface{}) {
	value = p.Get()
	if value == nil {
		value = p.ctx.New()
	}
	return
}
// 放回元素 到buffer 中
func (p *bufferPool) give(value interface{}) {
	p.ctx.Reset(value)
	p.Put(value)
}

type bufferValue struct {
	value    [maxBufferPool]interface{}
	transmit [maxBufferPool]interface{}
}
func (bv *bufferValue) Take(poolCtx types.BufferPoolCtx) (value interface{}) {
	i := poolCtx.Index()
	value = bPool[i].take()
	bv.value[i] = value
	return
}
// 返回poolCtx 对应的上一次take 的对象，如果没有，则新生成一个
func (bv *bufferValue) Find(poolCtx types.BufferPoolCtx, ...) interface{} {
	i := poolCtx.Index()
	if i <= 0 || i > int(index) {
		panic("buffer should call buffer.RegisterBuffer()")
	}
	if bv.value[i] != nil {
		return bv.value[i]
	}
	return bv.Take(poolCtx)
}

const maxBufferPool = 16
var (
	index int32
	bPool = bufferPoolArray[:]
    bufferPoolArray [maxBufferPool]bufferPool
    
    vPool = new(valuePool)
)
type valuePool struct {
	sync.Pool
}

// mosn.io/pkg/buffer/types.go
type IoBuffer interface {
	Read(p []byte) (n int, err error)
	ReadOnce(r io.Reader) (n int64, err error)
	Write(p []byte) (n int, err error)
	WriteString(s string) (n int, err error)
	WriteByte(p byte) error
	WriteUint32(p uint32) error
	WriteUint64(p uint64) error
	WriteTo(w io.Writer) (n int64, err error)
	Len() int
	Cap() int
	Reset()
	// String returns the contents of the unread portion of the buffer as a string. If the Buffer is a nil pointer, it returns "<nil>".
	String() string
	// Alloc alloc bytes from BytePoolBuffer
	Alloc(int)
	// Free free bytes to BytePoolBuffer
	Free()
	Append(data []byte) error
	CloseWithError(err error)
}
// mosn.io/pkg/buffer/iobuffer.go
type ioBuffer struct {
	buf     []byte // contents: buf[off : len(buf)]
	off     int    // read from &buf[off], write to &buf[len(buf)]
	offMark int
	count   int32
	eof     bool
	b *[]byte
}
func newIoBuffer(capacity int) IoBuffer {
	buffer := &ioBuffer{
		offMark: ResetOffMark,
		count:   1,
	}
	if capacity <= 0 {
		capacity = DefaultSize
	}
	buffer.b = GetBytes(capacity)
	buffer.buf = (*buffer.b)[:0]
	return buffer
}

// mosn.io/pkg/buffer/iobuffer_pool.go
var ibPool IoBufferPool
// IoBufferPool is Iobuffer Pool
type IoBufferPool struct {
	pool sync.Pool
}
var ibPool IoBufferPool
func GetIoBuffer(size int) IoBuffer {
	return ibPool.take(size)
}
func PutIoBuffer(buf IoBuffer) error {
	count := buf.Count(-1)
	if count > 0 {
		return nil
	} else if count < 0 {
		return errors.New("PutIoBuffer duplicate")
	}
	if p, _ := buf.(*pipe); p != nil {
		buf = p.IoBuffer
	}
	ibPool.give(buf)
	return nil
}

var bbPool *byteBufferPool
func init() {
	bbPool = newByteBufferPool()
}
func GetBytes(size int) *[]byte {
	return bbPool.take(size)
}
// mosn.io/pkg/buffer/bytebuffer_pool.go
// byteBufferPool is []byte pools
type byteBufferPool struct {
	minShift int
	minSize  int
	maxSize  int
	pool []*bufferSlot
}
type bufferSlot struct {
	defaultSize int
	pool        sync.Pool
}
func (p *byteBufferPool) take(size int) *[]byte {
	slot := p.slot(size)
	if slot == errSlot {
		b := newBytes(size)
		return &b
	}
	v := p.pool[slot].pool.Get()
	if v == nil {
		b := newBytes(p.pool[slot].defaultSize)
		b = b[0:size]
		return &b
	}
	b := v.(*[]byte)
	*b = (*b)[0:size]
	return b
}
func (p *byteBufferPool) give(buf *[]byte) {
	if buf == nil {
		return
	}
	size := cap(*buf)
	slot := p.slot(size)
	if slot == errSlot {
		return
	}
	if size != int(p.pool[slot].defaultSize) {
		return
	}
	p.pool[slot].pool.Put(buf)
}

// mosn.io/mosn/pkg/protocol/buffer.go
type ProtocolBuffers struct {
	reqData     buffer.IoBuffer
	reqHeader   buffer.IoBuffer
	reqHeaders  map[string]string
	reqTrailers map[string]string

	rspData     buffer.IoBuffer
	rspHeader   buffer.IoBuffer
	rspHeaders  map[string]string
	rspTrailers map[string]string
}
// GetReqData returns IoBuffer for request data
func (p *ProtocolBuffers) GetReqData(size int) buffer.IoBuffer {
	if size <= 0 {
		size = defaultDataSize
	}
	p.reqData = buffer.GetIoBuffer(size)
	return p.reqData
}

func (adsClient *ADSClient) Start() {
    adsClient.StreamClient = adsClient.AdsConfig.GetStreamClient()
    utils.GoWithRecover(func() {
        adsClient.sendThread()
    }, nil)
    utils.GoWithRecover(func() {
        adsClient.receiveThread()
    }, nil)
}
```go

定时发送请求
```go
func (adsClient *ADSClient) sendThread() {
    refreshDelay := adsClient.AdsConfig.RefreshDelay
    t1 := time.NewTimer(*refreshDelay)
    for {
        select {
        ...
        case <-t1.C:
            err := adsClient.reqClusters(adsClient.StreamClient)
            if err != nil {
                log.DefaultLogger.Infof("[xds] [ads client] send thread request cds fail!auto retry next period")
                adsClient.reconnect()
            }
            t1.Reset(*refreshDelay)
        }
    }
}

func (adsClient *ADSClient) receiveThread() {
    for {
        select {
    
        default:
            adsClient.StreamClientMutex.RLock()
            sc := adsClient.StreamClient
            adsClient.StreamClientMutex.RUnlock()
            ...
            resp, err := sc.Recv()
            ...
            typeURL := resp.TypeUrl
            HandleTypeURL(typeURL, adsClient, resp)
        }
    }
}

func HandleTypeURL(url string, client *ADSClient, resp *envoy_api_v2.DiscoveryResponse) {
    if f, ok := typeURLHandleFuncs[url]; ok {
        f(client, resp)
    }
}
func init() {
    RegisterTypeURLHandleFunc(EnvoyListener, HandleEnvoyListener)
    RegisterTypeURLHandleFunc(EnvoyCluster, HandleEnvoyCluster)
    RegisterTypeURLHandleFunc(EnvoyClusterLoadAssignment, HandleEnvoyClusterLoadAssignment)
    RegisterTypeURLHandleFunc(EnvoyRouteConfiguration, HandleEnvoyRouteConfiguration)
}

func HandleEnvoyCluster(client *ADSClient, resp *envoy_api_v2.DiscoveryResponse) {
    clusters := client.handleClustersResp(resp)
    ...
    conv.ConvertUpdateClusters(clusters)
    clusterNames := make([]string, 0)
    ...
    for _, cluster := range clusters {
        if cluster.GetType() == envoy_api_v2.Cluster_EDS {
            clusterNames = append(clusterNames, cluster.Name)
        }
    }
    ...
}

func ConvertUpdateEndpoints(loadAssignments []*envoy_api_v2.ClusterLoadAssignment) error {
    for _, loadAssignment := range loadAssignments {
        clusterName := loadAssignment.ClusterName
        for _, endpoints := range loadAssignment.Endpoints {
            hosts := ConvertEndpointsConfig(&endpoints)
            clusterMngAdapter := clusterAdapter.GetClusterMngAdapterInstance()
            ...
            clusterAdapter.GetClusterMngAdapterInstance().TriggerClusterHostUpdate(clusterName, hosts); 
            ...
            
        }
    }
    return errGlobal
}