func (t *svrTransHandler) SetInvokeHandleFunc(inkHdlFunc endpoint.Endpoint) {
	t.inkHdlFunc = inkHdlFunc
}
SetInvokeHandleFunc implements the remote.InvokeHandleFuncSetter interface.
package trans
import (
	"context"
	"net"
	"runtime/debug"
	stats2 "github.com/cloudwego/kitex/internal/stats"
	"github.com/cloudwego/kitex/pkg/endpoint"
	"github.com/cloudwego/kitex/pkg/kerrors"
	"github.com/cloudwego/kitex/pkg/remote"
	"github.com/cloudwego/kitex/pkg/rpcinfo"
	"github.com/cloudwego/kitex/pkg/serviceinfo"
	"github.com/cloudwego/kitex/pkg/stats"
)
func NewDefaultSvrTransHandler(opt *remote.ServerOption, ext Extension) (remote.ServerTransHandler, error) {
	return &svrTransHandler{
		opt:     opt,
		codec:   opt.Codec,
		svcInfo: opt.SvcInfo,
		ext:     ext,
	}, nil
}
type svrTransHandler struct {
	opt        *remote.ServerOption
	svcInfo    *serviceinfo.ServiceInfo
	inkHdlFunc endpoint.Endpoint
	codec      remote.Codec
	transPipe  *remote.TransPipeline
	ext        Extension
}
func (t *svrTransHandler) Write(ctx context.Context, conn net.Conn, sendMsg remote.Message) (err error) {
	var bufWriter remote.ByteBuffer
	ri := sendMsg.RPCInfo()
	stats2.Record(ctx, ri, stats.WriteStart, nil)
	defer func() {
		t.ext.ReleaseBuffer(bufWriter, err)
		stats2.Record(ctx, ri, stats.WriteFinish, err)
	}()
	if methodInfo, _ := GetMethodInfo(ri, t.svcInfo); methodInfo != nil {
		if methodInfo.OneWay() {
			return nil
		}
	}
	bufWriter = t.ext.NewWriteByteBuffer(ctx, conn, sendMsg)
	err = t.codec.Encode(ctx, sendMsg, bufWriter)
	if err != nil {
		return err
	}
	return bufWriter.Flush()
}
func (t *svrTransHandler) Read(ctx context.Context, conn net.Conn, recvMsg remote.Message) (err error) {
	var bufReader remote.ByteBuffer
	defer func() {
		t.ext.ReleaseBuffer(bufReader, err)
		stats2.Record(ctx, recvMsg.RPCInfo(), stats.ReadFinish, err)
	}()
	stats2.Record(ctx, recvMsg.RPCInfo(), stats.ReadStart, nil)
	bufReader = t.ext.NewReadByteBuffer(ctx, conn, recvMsg)
	err = t.codec.Decode(ctx, recvMsg, bufReader)
	if err != nil {
		recvMsg.Tags()[remote.ReadFailed] = true
		return err
	}
	return nil
}
func (t *svrTransHandler) OnRead(ctx context.Context, conn net.Conn) error {
	ri := rpcinfo.GetRPCInfo(ctx)
	t.ext.SetReadTimeout(ctx, conn, ri.Config(), remote.Server)
	var err error
	var closeConn bool
	var recvMsg remote.Message
	var sendMsg remote.Message
	defer func() {
		panicErr := recover()
		if panicErr != nil {
			closeConn = true
			if conn != nil {
				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()))
			}
		}
		if closeConn && conn != nil {
			conn.Close()
		}
		t.finishTracer(ctx, ri, err, panicErr)
		remote.RecycleMessage(recvMsg)
		remote.RecycleMessage(sendMsg)
	}()
	ctx = t.startTracer(ctx, ri)
	recvMsg = remote.NewMessageWithNewer(t.svcInfo, ri, remote.Call, remote.Server)
	recvMsg.SetPayloadCodec(t.opt.PayloadCodec)
	err = t.Read(ctx, conn, recvMsg)
	if err != nil {
		closeConn = true
		t.writeErrorReplyIfNeeded(ctx, recvMsg, conn, err, ri, true)
		t.OnError(ctx, err, conn)
		return nil
	}
	var methodInfo serviceinfo.MethodInfo
	if methodInfo, err = GetMethodInfo(ri, t.svcInfo); err != nil {
		closeConn = true
		t.writeErrorReplyIfNeeded(ctx, recvMsg, conn, err, ri, true)
		t.OnError(ctx, err, conn)
		return nil
	}
	if methodInfo.OneWay() {
		sendMsg = remote.NewMessage(nil, t.svcInfo, ri, remote.Reply, remote.Server)
	} else {
		sendMsg = remote.NewMessage(methodInfo.NewResult(), t.svcInfo, ri, remote.Reply, remote.Server)
	}
	ctx, err = t.transPipe.OnMessage(ctx, recvMsg, sendMsg)
	if err != nil {
		t.OnError(ctx, err, conn)
		err = remote.NewTransError(remote.InternalError, err)
		t.writeErrorReplyIfNeeded(ctx, recvMsg, conn, err, ri, false)
		return nil
	}
	remote.FillSendMsgFromRecvMsg(recvMsg, sendMsg)
	if err = t.transPipe.Write(ctx, conn, sendMsg); err != nil {
		t.OnError(ctx, err, conn)
		return nil
	}
	return nil
}
func (t *svrTransHandler) OnMessage(ctx context.Context, args, result remote.Message) (context.Context, error) {
	err := t.inkHdlFunc(ctx, args.Data(), result.Data())
	return ctx, err
}
func (t *svrTransHandler) OnActive(ctx context.Context, conn net.Conn) (context.Context, error) {
	_, ctx = t.opt.InitRPCInfoFunc(ctx, conn.RemoteAddr())
	return ctx, nil
}
func (t *svrTransHandler) OnInactive(ctx context.Context, conn net.Conn) {
	rpcinfo.PutRPCInfo(rpcinfo.GetRPCInfo(ctx))
}
func (t *svrTransHandler) OnError(ctx context.Context, err error, conn net.Conn) {
	rAddr := conn.RemoteAddr()
	if rAddr.Network() == "unix" {
		ri := rpcinfo.GetRPCInfo(ctx)
		if ri.From().Address() != nil {
			rAddr = ri.From().Address()
		}
	}
	if t.ext.IsRemoteClosedErr(err) {
		ri := rpcinfo.GetRPCInfo(ctx)
		if ri == nil {
			return
		}
		remote := rpcinfo.AsMutableEndpointInfo(ri.From())
		remote.SetTag(rpcinfo.RemoteClosedTag, "1")
	} else if pe, ok := err.(*kerrors.DetailedError); ok {
		t.opt.Logger.Errorf("KITEX: processing request error, remote=%v, err=%s\n%s", rAddr, err.Error(), pe.Stack())
	} else {
		t.opt.Logger.Errorf("KITEX: processing request error, remote=%v, err=%s", rAddr, err.Error())
	}
}