func (t *svrTransHandler) OnRead(muxSvrConnCtx context.Context, conn net.Conn) error {
	defer t.tryRecover(muxSvrConnCtx, conn)
	connection := conn.(netpoll.Connection)
	length, _, err := parseHeader(connection.Reader())
	if err != nil {
		err = fmt.Errorf("%w: addr(%s)", err, connection.RemoteAddr())
		t.opt.Logger.Errorf("KITEX: %s", err.Error())
		connection.Close()
		return err
	}
	reader, err := connection.Reader().Slice(length)
	if err != nil {
		err = fmt.Errorf("%w: addr(%s)", err, connection.RemoteAddr())
		t.opt.Logger.Errorf("KITEX: %s", err.Error())
		connection.Close()
		return nil
	}
	gofunc.GoFunc(muxSvrConnCtx, func() {
		muxSvrConn, _ := muxSvrConnCtx.Value(ctxKeyMuxSvrConn{}).(*muxSvrConn)
		rpcInfoCtx := muxSvrConn.pool.Get().(context.Context)
		rpcInfo := rpcinfo.GetRPCInfo(rpcInfoCtx)
		ctx := t.startTracer(rpcInfoCtx, rpcInfo)
		var err error
		var recvMsg remote.Message
		var sendMsg remote.Message
		defer func() {
			panicErr := recover()
			if panicErr != nil {
				if conn != nil {
					conn.Close()
					t.opt.Logger.Errorf("KITEX: panic happened, close conn[%s], %v\n%s", conn.RemoteAddr(), panicErr, string(debug.Stack()))
				} else {
					t.opt.Logger.Errorf("KITEX: panic happened, %v\n%s", panicErr, string(debug.Stack()))
				}
			}
			t.finishTracer(ctx, rpcInfo, err, panicErr)
			remote.RecycleMessage(recvMsg)
			remote.RecycleMessage(sendMsg)
			muxSvrConn.pool.Put(rpcInfoCtx)
		}()
		recvMsg = remote.NewMessageWithNewer(t.svcInfo, rpcInfo, remote.Call, remote.Server)
		bufReader := np.NewReaderByteBuffer(reader)
		err = t.readWithByteBuffer(ctx, bufReader, recvMsg)
		if err != nil {
			t.writeErrorReplyIfNeeded(ctx, recvMsg, muxSvrConn, rpcInfo, err, false, true)
			t.OnError(ctx, err, muxSvrConn)
			return
		}
		var methodInfo serviceinfo.MethodInfo
		if methodInfo, err = trans.GetMethodInfo(rpcInfo, t.svcInfo); err != nil {
			t.writeErrorReplyIfNeeded(ctx, recvMsg, muxSvrConn, rpcInfo, err, false, true)
			t.OnError(ctx, err, muxSvrConn)
			return
		}
		if methodInfo.OneWay() {
			sendMsg = remote.NewMessage(nil, t.svcInfo, rpcInfo, remote.Reply, remote.Server)
		} else {
			sendMsg = remote.NewMessage(methodInfo.NewResult(), t.svcInfo, rpcInfo, remote.Reply, remote.Server)
		}
		ctx, err = t.transPipe.OnMessage(ctx, recvMsg, sendMsg)
		if err != nil {
			t.OnError(ctx, err, muxSvrConn)
			err = remote.NewTransError(remote.InternalError, err)
			t.writeErrorReplyIfNeeded(ctx, recvMsg, muxSvrConn, rpcInfo, err, false, false)
			return
		}
		remote.FillSendMsgFromRecvMsg(recvMsg, sendMsg)
		if err = t.transPipe.Write(ctx, muxSvrConn, sendMsg); err != nil {
			t.OnError(ctx, err, muxSvrConn)
			return
		}
	})
	return nil
}
OnRead implements the remote.ServerTransHandler interface.
Returns write err only.
package netpollmux
import (
	"context"
	"errors"
	"fmt"
	"net"
	"runtime/debug"
	"sync"
	"github.com/cloudwego/netpoll"
	stats2 "github.com/cloudwego/kitex/internal/stats"
	"github.com/cloudwego/kitex/pkg/endpoint"
	"github.com/cloudwego/kitex/pkg/gofunc"
	"github.com/cloudwego/kitex/pkg/kerrors"
	"github.com/cloudwego/kitex/pkg/remote"
	"github.com/cloudwego/kitex/pkg/remote/trans"
	np "github.com/cloudwego/kitex/pkg/remote/trans/netpoll"
	"github.com/cloudwego/kitex/pkg/rpcinfo"
	"github.com/cloudwego/kitex/pkg/serviceinfo"
	"github.com/cloudwego/kitex/pkg/stats"
)
type svrTransHandlerFactory struct{}
func NewSvrTransHandlerFactory() remote.ServerTransHandlerFactory {
	return &svrTransHandlerFactory{}
}
func (f *svrTransHandlerFactory) NewTransHandler(opt *remote.ServerOption) (remote.ServerTransHandler, error) {
	return newSvrTransHandler(opt)
}
func newSvrTransHandler(opt *remote.ServerOption) (*svrTransHandler, error) {
	return &svrTransHandler{
		opt:     opt,
		codec:   opt.Codec,
		svcInfo: opt.SvcInfo,
	}, nil
}
var _ remote.ServerTransHandler = &svrTransHandler{}
type svrTransHandler struct {
	opt        *remote.ServerOption
	svcInfo    *serviceinfo.ServiceInfo
	inkHdlFunc endpoint.Endpoint
	codec      remote.Codec
	transPipe  *remote.TransPipeline
}
func (t *svrTransHandler) Write(ctx context.Context, conn net.Conn, sendMsg remote.Message) (err error) {
	ri := rpcinfo.GetRPCInfo(ctx)
	stats2.Record(ctx, ri, stats.WriteStart, nil)
	defer func() {
		stats2.Record(ctx, ri, stats.WriteFinish, nil)
	}()
	if methodInfo, _ := trans.GetMethodInfo(ri, t.svcInfo); methodInfo != nil {
		if methodInfo.OneWay() {
			return nil
		}
	}
	bufWriter := np.NewWriterByteBuffer(netpoll.NewLinkBuffer())
	err = t.codec.Encode(ctx, sendMsg, bufWriter)
	if err != nil {
		return err
	}
	conn.(*muxSvrConn).Put(func() (buf remote.ByteBuffer, isNil bool) {
		return bufWriter, false
	})
	return nil
}
func (t *svrTransHandler) Read(ctx context.Context, conn net.Conn, msg remote.Message) (err error) {
	return nil
}
func (t *svrTransHandler) readWithByteBuffer(ctx context.Context, bufReader remote.ByteBuffer, msg remote.Message) (err error) {
	defer func() {
		if bufReader != nil {
			if err != nil {
				bufReader.Skip(bufReader.ReadableLen())
			}
			bufReader.Release(err)
		}
		stats2.Record(ctx, msg.RPCInfo(), stats.ReadFinish, err)
	}()
	stats2.Record(ctx, msg.RPCInfo(), stats.ReadStart, nil)
	err = t.codec.Decode(ctx, msg, bufReader)
	if err != nil {
		msg.Tags()[remote.ReadFailed] = true
		return err
	}
	return nil
}