flowchart TB
    subgraph "Remote模块"
        subgraph "传输抽象层"
            TRANSHANDLER[TransHandler]
            PIPELINE[TransPipeline]
            BOUNDHANDLER[BoundHandler]
        end
        
        subgraph "消息抽象层"
            MESSAGE[Message接口]
            MESSAGEIMPL[message实现]
            TRANSINFO[TransInfo]
        end
        
        subgraph "编解码层"
            CODEC[Codec接口]
            METACODER[MetaEncoder/Decoder]
            PAYLOADCODEC[PayloadCodec]
        end
        
        subgraph "连接管理层"
            CONNPOOL[ConnPool]
            DIALER[Dialer]
            CONNWRAPPER[ConnWrapper]
        end
        
        subgraph "客户端层"
            REMOTECLI[RemoteClient]
            CLIENTHANDLER[ClientHandler]
        end
        
        subgraph "服务端层"
            REMOTESVR[RemoteServer]
            SERVERHANDLER[ServerHandler]
        end
    end
    
    subgraph "协议实现"
        THRIFT[Thrift编解码]
        GRPC[gRPC编解码]
        PROTOBUF[Protobuf编解码]
        TTHEADER[TTHeader协议]
    end
    
    subgraph "网络实现"
        NETPOLL[Netpoll传输]
        GONET[Go net传输]
        MUX[多路复用传输]
        HTTP2[HTTP2传输]
    end
    
    TRANSHANDLER --> PIPELINE
    PIPELINE --> BOUNDHANDLER
    
    MESSAGE --> MESSAGEIMPL
    MESSAGEIMPL --> TRANSINFO
    
    CODEC --> METACODER
    CODEC --> PAYLOADCODEC
    
    REMOTECLI --> CLIENTHANDLER
    REMOTESVR --> SERVERHANDLER
    
    CLIENTHANDLER --> TRANSHANDLER
    SERVERHANDLER --> TRANSHANDLER
    
    TRANSHANDLER --> CODEC
    TRANSHANDLER --> CONNPOOL
    
    CONNPOOL --> DIALER
    CONNPOOL --> CONNWRAPPER
    
    CODEC --> THRIFT
    CODEC --> GRPC
    CODEC --> PROTOBUF
    CODEC --> TTHEADER
    
    TRANSHANDLER --> NETPOLL
    TRANSHANDLER --> GONET
    TRANSHANDLER --> MUX
    TRANSHANDLER --> HTTP2

// 客户端发送消息流程
func (c *client) Send(ctx context.Context, ri rpcinfo.RPCInfo, req remote.Message) (err error) {
    // 1. 通过传输处理器写入消息
    _, err = c.transHdlr.Write(ctx, c.conn, req)
    if err != nil {
        // 2. 发送失败时释放连接
        c.connManager.ReleaseConn(err, ri)
    }
    return err
}

// 客户端接收消息流程
func (c *client) Recv(ctx context.Context, ri rpcinfo.RPCInfo, resp remote.Message) (err error) {
    if resp != nil {
        // 1. 通过传输处理器读取消息
        ctx, err = c.transHdlr.Read(ctx, c.conn, resp)
        // 2. 处理接收到的消息
        c.transHdlr.OnMessage(ctx, nil, resp)
    } else {
        // 3. 单向调用等待数据刷新
        time.Sleep(time.Millisecond / 2)
    }
    
    // 4. 释放连接回连接池
    c.connManager.ReleaseConn(err, ri)
    return err
}

func (p *TransPipeline) Write(ctx context.Context, conn net.Conn, sendMsg Message) (nctx context.Context, err error) {
    // 1. 依次执行出站处理器
    for _, h := range p.outboundHdrls {
        ctx, err = h.Write(ctx, conn, sendMsg)
        if err != nil {
            return ctx, err
        }
    }
    // 2. 执行网络处理器
    return p.netHdlr.Write(ctx, conn, sendMsg)
}

func (p *TransPipeline) OnMessage(ctx context.Context, args, result Message) (context.Context, error) {
    // 1. 依次执行入站处理器
    for _, h := range p.inboundHdrls {
        ctx, err = h.OnMessage(ctx, args, result)
        if err != nil {
            return ctx, err
        }
    }
    // 2. 执行网络处理器
    return p.netHdlr.OnMessage(ctx, args, result)
}

// 编码流程
func (c *defaultCodec) Encode(ctx context.Context, msg Message, out ByteBuffer) error {
    // 1. 编码元信息和负载
    if me, ok := c.payloadCodec.(MetaEncoder); ok {
        return me.EncodeMetaAndPayload(ctx, msg, out, me)
    }
    
    // 2. 分别编码元信息和负载
    if err := c.encodeMeta(ctx, msg, out); err != nil {
        return err
    }
    return c.payloadCodec.Marshal(ctx, msg, out)
}

// 解码流程
func (c *defaultCodec) Decode(ctx context.Context, msg Message, in ByteBuffer) error {
    // 1. 解码元信息
    if md, ok := c.payloadCodec.(MetaDecoder); ok {
        if err := md.DecodeMeta(ctx, msg, in); err != nil {
            return err
        }
    }
    
    // 2. 解码负载数据
    return c.payloadCodec.Unmarshal(ctx, msg, in)
}

// 获取连接
func (p *LongPool) Get(ctx context.Context, network, address string, opt *ConnOption) (net.Conn, error) {
    // 1. 从连接池获取可用连接
    if conn := p.tryGetConn(address); conn != nil {
        return conn, nil
    }
    
    // 2. 连接池无可用连接，创建新连接
    conn, err := opt.Dialer.DialContext(ctx, network, address)
    if err != nil {
        return nil, err
    }
    
    // 3. 设置连接参数
    if err := p.setConnOptions(conn, opt); err != nil {
        conn.Close()
        return nil, err
    }
    
    return conn, nil
}

// 释放连接
func (p *LongPool) Put(conn net.Conn, err error) {
    if err != nil {
        // 1. 有错误时直接关闭连接
        conn.Close()
        return
    }
    
    // 2. 检查连接是否可复用
    if !p.isReusable(conn) {
        conn.Close()
        return
    }
    
    // 3. 放回连接池
    p.putConn(conn)
}

// 实现自定义的传输处理器
type CustomTransHandler struct {
    // 自定义字段
}

func (h *CustomTransHandler) Write(ctx context.Context, conn net.Conn, send Message) (context.Context, error) {
    // 自定义写入逻辑
}

func (h *CustomTransHandler) Read(ctx context.Context, conn net.Conn, msg Message) (context.Context, error) {
    // 自定义读取逻辑
}

// 实现自定义的编解码器
type CustomCodec struct {
    // 自定义字段
}

func (c *CustomCodec) Encode(ctx context.Context, msg Message, out ByteBuffer) error {
    // 自定义编码逻辑
}

func (c *CustomCodec) Decode(ctx context.Context, msg Message, in ByteBuffer) error {
    // 自定义解码逻辑
}

// 实现自定义的边界处理器
type CustomBoundHandler struct {
    // 自定义字段
}

func (h *CustomBoundHandler) Write(ctx context.Context, conn net.Conn, send Message) (context.Context, error) {
    // 自定义出站处理逻辑
}

func (h *CustomBoundHandler) OnMessage(ctx context.Context, args, result Message) (context.Context, error) {
    // 自定义入站处理逻辑
}

// Message是Kitex消息的核心抽象
type Message interface {
    RPCInfo() rpcinfo.RPCInfo                    // 获取RPC调用信息
    Data() interface{}                           // 获取消息数据
    NewData(method string) (ok bool)             // 创建新的数据对象
    MessageType() MessageType                    // 获取消息类型
    SetMessageType(MessageType)                  // 设置消息类型
    RPCRole() RPCRole                           // 获取RPC角色
    PayloadLen() int                            // 获取负载长度
    SetPayloadLen(size int)                     // 设置负载长度
    TransInfo() TransInfo                       // 获取传输信息
    Tags() map[string]interface{}               // 获取标签信息
    PayloadCodec() PayloadCodec                 // 获取负载编解码器
    SetPayloadCodec(pc PayloadCodec)            // 设置负载编解码器
    Recycle()                                   // 回收消息对象
}

// 创建新消息
func NewMessage(data interface{}, ri rpcinfo.RPCInfo, msgType MessageType, rpcRole RPCRole) Message {
    // 1. 从对象池获取消息实例
    msg := messagePool.Get().(*message)
    
    // 2. 初始化消息字段
    msg.data = data
    msg.rpcInfo = ri
    msg.msgType = msgType
    msg.rpcRole = rpcRole
    
    // 3. 获取传输信息对象
    msg.transInfo = transInfoPool.Get().(*transInfo)
    
    return msg
}

// 回收消息对象
func RecycleMessage(msg Message) {
    if msg != nil {
        msg.Recycle()
    }
}

// 在客户端发送消息时的使用
func (c *client) Send(ctx context.Context, ri rpcinfo.RPCInfo, req remote.Message) (err error) {
    // 通过传输处理器写入消息
    _, err = c.transHdlr.Write(ctx, c.conn, req)
    return err
}

// 在服务端处理消息时的使用
func (s *serverHandler) OnMessage(ctx context.Context, args, result Message) (context.Context, error) {
    // 处理接收到的消息
    return s.eps.Call(ctx, args, result)
}

sequenceDiagram
    autonumber
    participant CLIENT as 客户端代码
    participant POOL as 对象池
    participant MSG as Message实例
    participant TRANSINFO as TransInfo
    participant HANDLER as 传输处理器
    
    CLIENT->>POOL: NewMessage(data, ri, msgType, role)
    POOL->>MSG: messagePool.Get()
    MSG->>MSG: 初始化字段
    MSG->>POOL: transInfoPool.Get()
    POOL->>TRANSINFO: 返回TransInfo实例
    TRANSINFO-->>MSG: 设置transInfo
    MSG-->>CLIENT: 返回Message实例
    
    CLIENT->>HANDLER: Write(ctx, conn, msg)
    HANDLER->>HANDLER: 编码和发送消息
    HANDLER-->>CLIENT: 返回发送结果
    
    CLIENT->>MSG: Recycle()
    MSG->>MSG: zero()清理字段
    MSG->>POOL: messagePool.Put(msg)
    MSG->>TRANSINFO: transInfo.Recycle()
    TRANSINFO->>POOL: transInfoPool.Put(transInfo)

// Codec是Kitex编解码层的抽象
type Codec interface {
    Encode(ctx context.Context, msg Message, out ByteBuffer) error  // 编码消息
    Decode(ctx context.Context, msg Message, in ByteBuffer) error   // 解码消息
    Name() string                                                   // 编解码器名称
}

// MetaEncoder是具有元信息和负载阶段的编码层抽象
type MetaEncoder interface {
    EncodeMetaAndPayload(ctx context.Context, msg Message, out ByteBuffer, me MetaEncoder) error
    EncodePayload(ctx context.Context, msg Message, out ByteBuffer) error
}

// MetaDecoder是具有元信息和负载阶段的解码层抽象
type MetaDecoder interface {
    DecodeMeta(ctx context.Context, msg Message, in ByteBuffer) error
    DecodePayload(ctx context.Context, msg Message, in ByteBuffer) error
}

// 默认编解码器实现
type defaultCodec struct {
    payloadCodec PayloadCodec
}

func (c *defaultCodec) Encode(ctx context.Context, msg Message, out ByteBuffer) error {
    // 1. 检查是否支持元信息编码
    if me, ok := c.payloadCodec.(MetaEncoder); ok {
        return me.EncodeMetaAndPayload(ctx, msg, out, me)
    }
    
    // 2. 分别编码元信息和负载
    if err := c.encodeMeta(ctx, msg, out); err != nil {
        return err
    }
    return c.payloadCodec.Marshal(ctx, msg, out)
}

func (c *defaultCodec) Decode(ctx context.Context, msg Message, in ByteBuffer) error {
    // 1. 检查是否支持元信息解码
    if md, ok := c.payloadCodec.(MetaDecoder); ok {
        if err := md.DecodeMeta(ctx, msg, in); err != nil {
            return err
        }
    }
    
    // 2. 解码负载数据
    return c.payloadCodec.Unmarshal(ctx, msg, in)
}

func (c *defaultCodec) Name() string {
    return c.payloadCodec.Name()
}

sequenceDiagram
    autonumber
    participant CLIENT as 客户端
    participant CODEC as Codec
    participant PAYLOADCODEC as PayloadCodec
    participant BUFFER as ByteBuffer
    participant SERVER as 服务端
    
    CLIENT->>CODEC: Encode(ctx, msg, out)
    CODEC->>CODEC: 检查MetaEncoder支持
    
    alt 支持MetaEncoder
        CODEC->>PAYLOADCODEC: EncodeMetaAndPayload(ctx, msg, out, me)
        PAYLOADCODEC->>BUFFER: 写入元信息和负载
    else 不支持MetaEncoder
        CODEC->>CODEC: encodeMeta(ctx, msg, out)
        CODEC->>PAYLOADCODEC: Marshal(ctx, msg, out)
        PAYLOADCODEC->>BUFFER: 写入负载数据
    end
    
    BUFFER-->>CLIENT: 编码完成
    CLIENT->>SERVER: 发送编码数据
    
    SERVER->>CODEC: Decode(ctx, msg, in)
    CODEC->>CODEC: 检查MetaDecoder支持
    
    alt 支持MetaDecoder
        CODEC->>PAYLOADCODEC: DecodeMeta(ctx, msg, in)
        PAYLOADCODEC->>BUFFER: 读取元信息
    end
    
    CODEC->>PAYLOADCODEC: Unmarshal(ctx, msg, in)
    PAYLOADCODEC->>BUFFER: 读取负载数据
    PAYLOADCODEC-->>CODEC: 解码完成
    CODEC-->>SERVER: 返回解码结果

// TransReadWriter定义读写操作
type TransReadWriter interface {
    Write(ctx context.Context, conn net.Conn, send Message) (nctx context.Context, err error)
    Read(ctx context.Context, conn net.Conn, msg Message) (nctx context.Context, err error)
}

// TransHandler类似Netty中的handler角色
type TransHandler interface {
    TransReadWriter
    OnInactive(ctx context.Context, conn net.Conn)                           // 连接关闭回调
    OnError(ctx context.Context, err error, conn net.Conn)                  // 错误处理回调
    OnMessage(ctx context.Context, args, result Message) (context.Context, error)  // 消息处理回调
    SetPipeline(pipeline *TransPipeline)                                    // 设置传输管道
}

// ClientTransHandler客户端传输处理器
type ClientTransHandler interface {
    TransHandler
}

// ServerTransHandler服务端传输处理器
type ServerTransHandler interface {
    TransHandler
    OnActive(ctx context.Context, conn net.Conn) (context.Context, error)   // 连接激活回调
    OnRead(ctx context.Context, conn net.Conn) (context.Context, error)     // 读取数据回调
}

// 默认服务端传输处理器
type svrTransHandler struct {
    opt      *ServerOption
    codec    Codec
    pipeline *TransPipeline
}

func (t *svrTransHandler) Write(ctx context.Context, conn net.Conn, send Message) (context.Context, error) {
    // 1. 获取写缓冲区
    buf := netpoll.NewLinkBuffer()
    defer buf.Release()
    
    // 2. 编码消息到缓冲区
    if err := t.codec.Encode(ctx, send, buf); err != nil {
        return ctx, err
    }
    
    // 3. 写入网络连接
    _, err := conn.Write(buf.Bytes())
    return ctx, err
}

func (t *svrTransHandler) Read(ctx context.Context, conn net.Conn, msg Message) (context.Context, error) {
    // 1. 获取读缓冲区
    buf := netpoll.NewLinkBuffer()
    defer buf.Release()
    
    // 2. 从连接读取数据
    if _, err := buf.ReadFrom(conn); err != nil {
        return ctx, err
    }
    
    // 3. 解码消息
    if err := t.codec.Decode(ctx, msg, buf); err != nil {
        return ctx, err
    }
    
    return ctx, nil
}

func (t *svrTransHandler) OnMessage(ctx context.Context, args, result Message) (context.Context, error) {
    // 业务逻辑处理
    return t.opt.SvcSearcher.Call(ctx, args, result)
}

sequenceDiagram
    autonumber
    participant CLIENT as 客户端
    participant CONN as 网络连接
    participant HANDLER as TransHandler
    participant CODEC as Codec
    participant BUSINESS as 业务逻辑
    
    CLIENT->>CONN: 发送请求数据
    CONN->>HANDLER: OnRead(ctx, conn)
    HANDLER->>HANDLER: Read(ctx, conn, msg)
    HANDLER->>CODEC: Decode(ctx, msg, buf)
    CODEC-->>HANDLER: 解码完成
    
    HANDLER->>HANDLER: OnMessage(ctx, args, result)
    HANDLER->>BUSINESS: 调用业务逻辑
    BUSINESS-->>HANDLER: 返回处理结果
    
    HANDLER->>HANDLER: Write(ctx, conn, result)
    HANDLER->>CODEC: Encode(ctx, result, buf)
    CODEC-->>HANDLER: 编码完成
    HANDLER->>CONN: 写入响应数据
    CONN->>CLIENT: 返回响应
    
    alt 连接异常
        CONN->>HANDLER: OnError(ctx, err, conn)
        HANDLER->>HANDLER: 错误处理逻辑
    end
    
    alt 连接关闭
        CONN->>HANDLER: OnInactive(ctx, conn)
        HANDLER->>HANDLER: 清理连接资源
    end

// TransPipeline包含传输处理器链
type TransPipeline struct {
    netHdlr       TransHandler      // 网络处理器
    inboundHdrls  []InboundHandler  // 入站处理器列表
    outboundHdrls []OutboundHandler // 出站处理器列表
}

// BoundHandler是边界处理器的抽象
type BoundHandler interface{}

// OutboundHandler处理写事件
type OutboundHandler interface {
    BoundHandler
    Write(ctx context.Context, conn net.Conn, send Message) (context.Context, error)
}

// InboundHandler处理读事件
type InboundHandler interface {
    BoundHandler
    OnActive(ctx context.Context, conn net.Conn) (context.Context, error)
    OnInactive(ctx context.Context, conn net.Conn) context.Context
    OnRead(ctx context.Context, conn net.Conn) (context.Context, error)
    OnMessage(ctx context.Context, args, result Message) (context.Context, error)
}

// 创建传输管道
func NewTransPipeline(netHdlr TransHandler) *TransPipeline {
    transPl := newTransPipeline()
    transPl.netHdlr = netHdlr
    netHdlr.SetPipeline(transPl)
    return transPl
}

// 添加入站处理器
func (p *TransPipeline) AddInboundHandler(hdlr InboundHandler) *TransPipeline {
    p.inboundHdrls = append(p.inboundHdrls, hdlr)
    return p
}

// 添加出站处理器
func (p *TransPipeline) AddOutboundHandler(hdlr OutboundHandler) *TransPipeline {
    p.outboundHdrls = append(p.outboundHdrls, hdlr)
    return p
}

// 写入消息（出站处理）
func (p *TransPipeline) Write(ctx context.Context, conn net.Conn, sendMsg Message) (nctx context.Context, err error) {
    // 1. 依次执行出站处理器
    for _, h := range p.outboundHdrls {
        ctx, err = h.Write(ctx, conn, sendMsg)
        if err != nil {
            return ctx, err
        }
    }
    // 2. 执行网络处理器
    return p.netHdlr.Write(ctx, conn, sendMsg)
}

// 消息处理（入站处理）
func (p *TransPipeline) OnMessage(ctx context.Context, args, result Message) (context.Context, error) {
    var err error
    // 1. 依次执行入站处理器
    for _, h := range p.inboundHdrls {
        ctx, err = h.OnMessage(ctx, args, result)
        if err != nil {
            return ctx, err
        }
    }
    // 2. 执行网络处理器
    return p.netHdlr.OnMessage(ctx, args, result)
}

sequenceDiagram
    autonumber
    participant CLIENT as 客户端
    participant PIPELINE as TransPipeline
    participant OUT1 as 出站处理器1
    participant OUT2 as 出站处理器2
    participant NET as 网络处理器
    participant IN1 as 入站处理器1
    participant IN2 as 入站处理器2
    participant BUSINESS as 业务逻辑
    
    Note over CLIENT,BUSINESS: 出站处理（发送消息）
    CLIENT->>PIPELINE: Write(ctx, conn, msg)
    PIPELINE->>OUT1: Write(ctx, conn, msg)
    OUT1->>OUT2: Write(ctx, conn, msg)
    OUT2->>NET: Write(ctx, conn, msg)
    NET-->>PIPELINE: 写入完成
    PIPELINE-->>CLIENT: 返回结果
    
    Note over CLIENT,BUSINESS: 入站处理（接收消息）
    NET->>PIPELINE: OnMessage(ctx, args, result)
    PIPELINE->>IN1: OnMessage(ctx, args, result)
    IN1->>IN2: OnMessage(ctx, args, result)
    IN2->>BUSINESS: 调用业务逻辑
    BUSINESS-->>IN2: 返回处理结果
    IN2-->>IN1: 传递结果
    IN1-->>PIPELINE: 传递结果
    PIPELINE-->>NET: 返回最终结果

// ConnPool是连接池的抽象接口
type ConnPool interface {
    Get(ctx context.Context, network, address string, opt *ConnOption) (net.Conn, error)
    Put(net.Conn, error)
    Discard(net.Conn)
    Clean(network, address string)
    Reporter() ConnPoolReporter
}

// ConnOption连接选项
type ConnOption struct {
    Dialer    Dialer
    ConnectTimeout time.Duration
    ReadTimeout    time.Duration
    WriteTimeout   time.Duration
}

// 长连接池实现
type LongPool struct {
    pools    sync.Map  // 地址到连接池的映射
    maxIdle  int       // 最大空闲连接数
    reporter ConnPoolReporter
}

func (p *LongPool) Get(ctx context.Context, network, address string, opt *ConnOption) (net.Conn, error) {
    // 1. 获取地址对应的连接池
    pool := p.getPool(address)
    
    // 2. 尝试从池中获取连接
    if conn := pool.tryGet(); conn != nil {
        return conn, nil
    }
    
    // 3. 池中无可用连接，创建新连接
    conn, err := opt.Dialer.DialContext(ctx, network, address)
    if err != nil {
        return nil, err
    }
    
    // 4. 设置连接选项
    if err := setConnOption(conn, opt); err != nil {
        conn.Close()
        return nil, err
    }
    
    return conn, nil
}

func (p *LongPool) Put(conn net.Conn, err error) {
    if err != nil {
        // 有错误时直接关闭连接
        conn.Close()
        return
    }
    
    // 检查连接是否可复用
    if !isConnReusable(conn) {
        conn.Close()
        return
    }
    
    // 放回连接池
    address := conn.RemoteAddr().String()
    pool := p.getPool(address)
    pool.put(conn)
}

sequenceDiagram
    autonumber
    participant CLIENT as 客户端
    participant POOL as ConnPool
    participant CONNMAP as 连接映射
    participant DIALER as Dialer
    participant CONN as 网络连接
    
    CLIENT->>POOL: Get(ctx, network, address, opt)
    POOL->>CONNMAP: getPool(address)
    CONNMAP->>CONNMAP: tryGet()从池中获取
    
    alt 池中有可用连接
        CONNMAP-->>POOL: 返回现有连接
        POOL-->>CLIENT: 返回连接
    else 池中无可用连接
        POOL->>DIALER: DialContext(ctx, network, address)
        DIALER->>CONN: 建立新连接
        CONN-->>DIALER: 连接建立成功
        DIALER-->>POOL: 返回新连接
        POOL->>POOL: setConnOption(conn, opt)
        POOL-->>CLIENT: 返回连接
    end
    
    CLIENT->>CLIENT: 使用连接进行RPC调用
    
    CLIENT->>POOL: Put(conn, err)
    
    alt 调用有错误
        POOL->>CONN: Close()关闭连接
    else 调用成功
        POOL->>POOL: isConnReusable(conn)检查可复用性
        alt 连接可复用
            POOL->>CONNMAP: put(conn)放回池中
        else 连接不可复用
            POOL->>CONN: Close()关闭连接
        end
    end

// 创建消息时指定正确的类型和角色
msg := remote.NewMessage(req, ri, remote.Call, remote.Client)
defer remote.RecycleMessage(msg)

// 设置必要的传输信息
msg.TransInfo().PutTransStrInfo(map[string]string{
    "custom-header": "value",
})

// 实现自定义编解码器
type CustomCodec struct {
    name string
}

func (c *CustomCodec) Encode(ctx context.Context, msg remote.Message, out remote.ByteBuffer) error {
    // 自定义编码逻辑
    return nil
}

func (c *CustomCodec) Decode(ctx context.Context, msg remote.Message, in remote.ByteBuffer) error {
    // 自定义解码逻辑
    return nil
}

func (c *CustomCodec) Name() string {
    return c.name
}

// 添加自定义处理器
type LoggingHandler struct{}

func (h *LoggingHandler) Write(ctx context.Context, conn net.Conn, send remote.Message) (context.Context, error) {
    // 记录发送日志
    log.Infof("Sending message: %+v", send)
    return ctx, nil
}

// 在创建TransPipeline时添加处理器
pipeline := remote.NewTransPipeline(netHandler)
pipeline.AddOutboundHandler(&LoggingHandler{})

classDiagram
    class Message {
        <<interface>>
        +RPCInfo() rpcinfo.RPCInfo
        +Data() interface{}
        +NewData(method string) bool
        +MessageType() MessageType
        +SetMessageType(MessageType)
        +RPCRole() RPCRole
        +PayloadLen() int
        +SetPayloadLen(size int)
        +TransInfo() TransInfo
        +Tags() map[string]interface{}
        +PayloadCodec() PayloadCodec
        +SetPayloadCodec(pc PayloadCodec)
        +Recycle()
    }
    
    class message {
        -msgType MessageType
        -data interface{}
        -rpcInfo rpcinfo.RPCInfo
        -rpcRole RPCRole
        -compressType CompressType
        -payloadSize int
        -transInfo TransInfo
        -tags map[string]interface{}
        -payloadCodec PayloadCodec
        +zero()
        +Recycle()
    }
    
    class TransInfo {
        <<interface>>
        +TransStrInfo() map[string]string
        +TransIntInfo() map[uint16]string
        +PutTransIntInfo(map[uint16]string)
        +PutTransStrInfo(kvInfo map[string]string)
        +Recycle()
    }
    
    class transInfo {
        -strInfo map[string]string
        -intInfo map[uint16]string
        +zero()
        +Recycle()
    }
    
    class ServerOption {
        +SvcSearcher ServiceSearcher
        +TransServerFactory TransServerFactory
        +SvrHandlerFactory ServerTransHandlerFactory
        +Codec Codec
        +PayloadCodec PayloadCodec
        +Listener net.Listener
        +Address net.Addr
        +ReusePort bool
        +ExitWaitTime time.Duration
        +AcceptFailedDelayTime time.Duration
        +MaxConnectionIdleTime time.Duration
        +ReadWriteTimeout time.Duration
        +TracerCtl *rpcinfo.TraceController
        +GRPCCfg *grpc.ServerConfig
        +TTHeaderStreamingOptions TTHeaderStreamingOptions
        +CompatibleMiddlewareForUnary bool
    }
    
    class ClientOption {
        +SvcInfo *serviceinfo.ServiceInfo
        +CliHandlerFactory ClientTransHandlerFactory
        +Codec Codec
        +PayloadCodec PayloadCodec
        +ConnPool ConnPool
        +Dialer Dialer
        +EnableConnPoolReporter bool
        +GRPCStreamingCliHandlerFactory ClientTransHandlerFactory
        +GRPCStreamingConnPool ConnPool
        +TTHeaderStreamingCliHandlerFactory ClientStreamFactory
    }
    
    class Option {
        +Outbounds []OutboundHandler
        +Inbounds []InboundHandler
        +StreamingMetaHandlers []StreamingMetaHandler
        +PrependBoundHandler(h BoundHandler)
        +AppendBoundHandler(h BoundHandler)
    }
    
    class TransPipeline {
        -netHdlr TransHandler
        -inboundHdrls []InboundHandler
        -outboundHdrls []OutboundHandler
        +AddInboundHandler(hdlr InboundHandler) *TransPipeline
        +AddOutboundHandler(hdlr OutboundHandler) *TransPipeline
        +Write(ctx context.Context, conn net.Conn, sendMsg Message) (context.Context, error)
        +OnMessage(ctx context.Context, args, result Message) (context.Context, error)
    }
    
    class MessageType {
        <<enumeration>>
        InvalidMessageType = 0
        Call = 1
        Reply = 2
        Exception = 3
        Oneway = 4
        Stream = 5
        Heartbeat = 6
    }
    
    class RPCRole {
        <<enumeration>>
        Client
        Server
    }
    
    class CompressType {
        <<enumeration>>
        NoCompress
        Gzip
        Snappy
    }
    
    Message <|-- message
    TransInfo <|-- transInfo
    message --> MessageType
    message --> RPCRole
    message --> CompressType
    message --> TransInfo
    ServerOption --> Option
    ClientOption --> Option
    TransPipeline --> TransHandler
    TransPipeline --> InboundHandler
    TransPipeline --> OutboundHandler

// Message接口定义
type Message interface {
    RPCInfo() rpcinfo.RPCInfo                    // RPC调用信息
    Data() interface{}                           // 消息数据
    NewData(method string) (ok bool)             // 创建新数据
    MessageType() MessageType                    // 消息类型
    SetMessageType(MessageType)                  // 设置消息类型
    RPCRole() RPCRole                           // RPC角色
    PayloadLen() int                            // 负载长度
    SetPayloadLen(size int)                     // 设置负载长度
    TransInfo() TransInfo                       // 传输信息
    Tags() map[string]interface{}               // 标签信息
    PayloadCodec() PayloadCodec                 // 负载编解码器
    SetPayloadCodec(pc PayloadCodec)            // 设置编解码器
    Recycle()                                   // 回收消息
}

// message实现结构
type message struct {
    msgType      MessageType                    // 消息类型
    data         interface{}                    // 消息数据
    rpcInfo      rpcinfo.RPCInfo               // RPC信息
    rpcRole      RPCRole                       // RPC角色
    compressType CompressType                  // 压缩类型
    payloadSize  int                           // 负载大小
    transInfo    TransInfo                     // 传输信息
    tags         map[string]interface{}        // 标签映射
    payloadCodec PayloadCodec                  // 负载编解码器
}