feat: use binary codec for SUDP packets (#5461)

client/control.go

@@ -103,7 +103,7 @@ func NewControl(ctx context.Context, sessionCtx *SessionContext) (*Control, erro
 		sessionCtx.UDPPacketCodec,
 	)
 	ctl.vm = visitor.NewManager(ctl.ctx, sessionCtx.RunID, sessionCtx.Common,
-		ctl.connectServer, ctl.msgTransporter, sessionCtx.VnetController)
+		ctl.connectServer, ctl.msgTransporter, sessionCtx.VnetController, sessionCtx.UDPPacketCodec)
 	return ctl, nil
 }
 

client/proxy/sudp.go

@@ -87,7 +87,13 @@ func (pxy *SUDPProxy) InWorkConn(conn net.Conn, _ *msg.StartWorkConn) {
 	}
 
 	workConn := netpkg.WrapReadWriteCloserToConn(remote, conn)
-	payloadConn := msg.NewConn(workConn, msg.NewReadWriter(workConn, pxy.clientCfg.Transport.WireProtocol))
+	payloadRW, err := msg.NewUDPPacketReadWriter(workConn, pxy.clientCfg.Transport.WireProtocol, pxy.udpPacketCodec)
+	if err != nil {
+		xl.Errorf("create SUDP packet read writer: %v", err)
+		_ = workConn.Close()
+		return
+	}
+	payloadConn := msg.NewConn(workConn, payloadRW)
 	readCh := make(chan *msg.UDPPacket, 1024)
 	sendCh := make(chan msg.Message, 1024)
 	isClose := false

client/service_shutdown_test.go

@@ -35,7 +35,7 @@ func newGracefulCloseTestService() *Service {
 		doneCh: make(chan struct{}),
 	}
 	ctl.pm = proxy.NewManager(ctx, common, nil, nil, nil, "")
-	ctl.vm = visitor.NewManager(ctx, "graceful-close-race", common, nil, nil, nil)
+	ctl.vm = visitor.NewManager(ctx, "graceful-close-race", common, nil, nil, nil, "")
 	return &Service{ctl: ctl, cancel: context.CancelCauseFunc(func(error) {})}
 }
 

client/visitor/sudp.go

@@ -113,7 +113,13 @@ func (sv *SUDPVisitor) dispatcher() {
 func (sv *SUDPVisitor) worker(workConn net.Conn, firstPacket *msg.UDPPacket) {
 	xl := xlog.FromContextSafe(sv.ctx)
 	xl.Debugf("starting sudp proxy worker")
-	payloadConn := msg.NewConn(workConn, msg.NewReadWriter(workConn, sv.clientCfg.Transport.WireProtocol))
+	payloadRW, err := msg.NewUDPPacketReadWriter(workConn, sv.clientCfg.Transport.WireProtocol, udpPacketCodecFromHelper(sv.helper))
+	if err != nil {
+		xl.Errorf("create SUDP packet read writer: %v", err)
+		_ = workConn.Close()
+		return
+	}
+	payloadConn := msg.NewConn(workConn, payloadRW)
 
 	wg := &sync.WaitGroup{}
 	wg.Add(2)

client/visitor/visitor.go

@@ -50,6 +50,17 @@ type Helper interface {
 	RunID() string
 }
 
+type udpPacketCodecProvider interface {
+	UDPPacketCodec() string
+}
+
+func udpPacketCodecFromHelper(helper Helper) string {
+	if provider, ok := helper.(udpPacketCodecProvider); ok {
+		return provider.UDPPacketCodec()
+	}
+	return ""
+}
+
 // Visitor is used for forward traffics from local port tot remote service.
 type Visitor interface {
 	Run() error

client/visitor/visitor_manager.go

@@ -53,7 +53,12 @@ func NewManager(
 	connectServer func() (*msg.Conn, error),
 	msgTransporter transport.MessageTransporter,
 	vnetController *vnet.Controller,
+	udpPacketCodecs ...string,
 ) *Manager {
+	udpPacketCodec := ""
+	if len(udpPacketCodecs) > 0 {
+		udpPacketCodec = udpPacketCodecs[0]
+	}
 	m := &Manager{
 		clientCfg:     clientCfg,
 		cfgs:          make(map[string]v1.VisitorConfigurer),
@@ -68,6 +73,7 @@ func NewManager(
 		vnetController:  vnetController,
 		transferConnFn:  m.TransferConn,
 		runID:           runID,
+		udpPacketCodec:  udpPacketCodec,
 	}
 	return m
 }
@@ -205,6 +211,7 @@ type visitorHelperImpl struct {
 	vnetController  *vnet.Controller
 	transferConnFn  func(name string, conn net.Conn) error
 	runID           string
+	udpPacketCodec  string
 }
 
 func (v *visitorHelperImpl) ConnectServer() (*msg.Conn, error) {
@@ -226,3 +233,7 @@ func (v *visitorHelperImpl) VNetController() *vnet.Controller {
 func (v *visitorHelperImpl) RunID() string {
 	return v.runID
 }
+
+func (v *visitorHelperImpl) UDPPacketCodec() string {
+	return v.udpPacketCodec
+}