Add Stream based Packet Addr for UDP Connections.

(This change is machine assisted.)

app/tun/tunsorter/tunsorter.go

@@ -56,6 +56,10 @@ func (t *TunSorter) onNewConnection(connection *trackedUDPConnection) {
 		ctx := context.WithValue(t.ctx, udp.DispatcherConnectionTerminationSignalReceiverMark, connection) // nolint:staticcheck
 		packetAddrDispatcherFactory := udp.NewPacketAddrDispatcherCreator(ctx)
 		udpDispatcherConstructor = packetAddrDispatcherFactory.NewPacketAddrDispatcher
+	case packetaddr.PacketAddrType_Stream:
+		ctx := context.WithValue(t.ctx, udp.DispatcherConnectionTerminationSignalReceiverMark, connection) // nolint:staticcheck
+		packetAddrDispatcherFactory := udp.NewStreamPacketAddrDispatcherCreator(ctx)
+		udpDispatcherConstructor = packetAddrDispatcherFactory.NewPacketAddrDispatcher
 	}
 	udpDispatcher := udpDispatcherConstructor(t.dispatcher, func(ctx context.Context, packet *vudp.Packet) {
 		connection.onWritePacket(packet.Source, packet.Payload.Bytes())

app/webrtc/connect_via_net.go

@@ -38,8 +38,8 @@ func newConnectViaNet(
 	connectVia string,
 	packetEncoding packetaddr.PacketAddrType,
 ) (*connectViaNet, error) {
-	if packetEncoding != packetaddr.PacketAddrType_Packet {
-		return nil, newError("active listener connect_via currently requires packet_encoding=Packet")
+	if packetEncoding != packetaddr.PacketAddrType_Packet && packetEncoding != packetaddr.PacketAddrType_Stream {
+		return nil, newError("active listener connect_via currently requires packet_encoding=Packet or Stream")
 	}
 
 	base, err := stdnet.NewNet()
@@ -205,7 +205,7 @@ func (n *connectViaNet) newPacketConn(network string) (v2net.PacketConn, *net.UD
 	ctx := n.ctx
 	ctx = session.SetForcedOutboundTagToContext(ctx, n.connectVia)
 
-	conn, err := packetaddr.CreatePacketAddrConn(ctx, n.dispatcher, false)
+	conn, err := packetaddr.CreatePacketAddrConn(ctx, n.dispatcher, n.packetEncoding == packetaddr.PacketAddrType_Stream)
 	if err != nil {
 		return nil, nil, newError("failed to create packetaddr connection for active listener").Base(err)
 	}

common/net/packetaddr/config.pb.go

@@ -20,6 +20,7 @@ type PacketAddrType int32
 const (
 	PacketAddrType_None   PacketAddrType = 0
 	PacketAddrType_Packet PacketAddrType = 1
+	PacketAddrType_Stream PacketAddrType = 2
 )
 
 // Enum value maps for PacketAddrType.
@@ -27,10 +28,12 @@ var (
 	PacketAddrType_name = map[int32]string{
 		0: "None",
 		1: "Packet",
+		2: "Stream",
 	}
 	PacketAddrType_value = map[string]int32{
 		"None":   0,
 		"Packet": 1,
+		"Stream": 2,
 	}
 )
 
@@ -65,11 +68,13 @@ var File_common_net_packetaddr_config_proto protoreflect.FileDescriptor
 
 const file_common_net_packetaddr_config_proto_rawDesc = "" +
 	"\n" +
-	"\"common/net/packetaddr/config.proto\x12\x19v2ray.core.net.packetaddr*&\n" +
+	"\"common/net/packetaddr/config.proto\x12\x19v2ray.core.net.packetaddr*2\n" +
 	"\x0ePacketAddrType\x12\b\n" +
 	"\x04None\x10\x00\x12\n" +
 	"\n" +
-	"\x06Packet\x10\x01B\x81\x01\n" +
+	"\x06Packet\x10\x01\x12\n" +
+	"\n" +
+	"\x06Stream\x10\x02B\x81\x01\n" +
 	"$com.v2ray.core.common.net.packetaddrP\x01Z4github.com/v2fly/v2ray-core/v5/common/net/packetaddr\xaa\x02 V2Ray.Core.Common.Net.Packetaddrb\x06proto3"
 
 var (

common/net/packetaddr/config.proto

@@ -9,4 +9,5 @@ option java_multiple_files = true;
 enum PacketAddrType {
   None = 0;
   Packet = 1;
-}
+  Stream = 2;
+}

common/net/packetaddr/connection_adaptor.go

@@ -16,7 +16,6 @@ import (
 
 var (
 	errNotPacketConn = errors.New("not a packet connection")
-	errUnsupported   = errors.New("unsupported action")
 )
 
 func ToPacketAddrConn(link *transport.Link, dest net.Destination) (net.PacketConn, error) {
@@ -27,7 +26,17 @@ func ToPacketAddrConn(link *transport.Link, dest net.Destination) (net.PacketCon
 	case seqPacketMagicAddress:
 		return &packetConnectionAdaptor{
 			readerAccess: &sync.Mutex{},
+			writerAccess: &sync.Mutex{},
 			readerBuffer: nil,
+			isStream:     false,
+			link:         link,
+		}, nil
+	case streamPacketMagicAddress:
+		return &packetConnectionAdaptor{
+			readerAccess: &sync.Mutex{},
+			writerAccess: &sync.Mutex{},
+			streamReader: &buf.BufferedReader{Reader: link.Reader},
+			isStream:     true,
 			link:         link,
 		}, nil
 	default:
@@ -36,34 +45,58 @@ func ToPacketAddrConn(link *transport.Link, dest net.Destination) (net.PacketCon
 }
 
 func CreatePacketAddrConn(ctx context.Context, dispatcher routing.Dispatcher, isStream bool) (net.PacketConn, error) {
-	if isStream {
-		return nil, errUnsupported
-	}
 	packetDest := net.Destination{
 		Address: net.DomainAddress(seqPacketMagicAddress),
 		Port:    0,
 		Network: net.Network_UDP,
 	}
+	if isStream {
+		packetDest.Address = net.DomainAddress(streamPacketMagicAddress)
+		packetDest.Network = net.Network_TCP
+	}
 	link, err := dispatcher.Dispatch(ctx, packetDest)
 	if err != nil {
 		return nil, err
 	}
-	return &packetConnectionAdaptor{
+	conn := &packetConnectionAdaptor{
 		readerAccess: &sync.Mutex{},
+		writerAccess: &sync.Mutex{},
 		readerBuffer: nil,
+		isStream:     isStream,
 		link:         link,
-	}, nil
+	}
+	if isStream {
+		conn.streamReader = &buf.BufferedReader{Reader: link.Reader}
+	}
+	return conn, nil
 }
 
 type packetConnectionAdaptor struct {
 	readerAccess *sync.Mutex
+	writerAccess *sync.Mutex
 	readerBuffer buf.MultiBuffer
+	streamReader *buf.BufferedReader
+	isStream     bool
 	link         *transport.Link
 }
 
 func (c *packetConnectionAdaptor) ReadFrom(p []byte) (n int, addr gonet.Addr, err error) {
 	c.readerAccess.Lock()
 	defer c.readerAccess.Unlock()
+	if c.isStream {
+		packet, err := ExtractPacketFromStream(c.streamReader)
+		if err != nil {
+			return 0, nil, err
+		}
+		data, addr, err := ExtractAddressFromPacket(packet)
+		if err != nil {
+			packet.Release()
+			return 0, nil, err
+		}
+		n = copy(p, data.Bytes())
+		data.Release()
+		return n, addr, nil
+	}
 	if c.readerBuffer.IsEmpty() {
 		c.readerBuffer, err = c.link.Reader.ReadMultiBuffer()
 		if err != nil {
@@ -90,6 +123,14 @@ func (c *packetConnectionAdaptor) WriteTo(p []byte, addr gonet.Addr) (n int, err
 	if err != nil {
 		return 0, err
 	}
+	if c.isStream {
+		buffer, err = AttachLengthToPacket(buffer)
+		if err != nil {
+			return 0, err
+		}
+	}
+	c.writerAccess.Lock()
+	defer c.writerAccess.Unlock()
 	mb := buf.MultiBuffer{buffer}
 	err = c.link.Writer.WriteMultiBuffer(mb)
 	if err != nil {
@@ -122,6 +163,9 @@ func (c packetConnectionAdaptor) SetWriteDeadline(t time.Time) error {
 }
 
 func ToPacketAddrConnWrapper(conn net.PacketConn, isStream bool) FusedConnection {
+	if isStream {
+		return &streamPacketConnWrapper{PacketConn: conn}
+	}
 	return &packetConnWrapper{conn}
 }
 
@@ -168,6 +212,82 @@ func (pc *packetConnWrapper) Write(p []byte) (n int, err error) {
 	return len(p), nil
 }
 
+type streamPacketConnWrapper struct {
+	net.PacketConn
+
+	readAccess  sync.Mutex
+	readBuffer  []byte
+	packetRead  []byte
+	writeAccess sync.Mutex
+	writeBuffer []byte
+}
+
+func (pc *streamPacketConnWrapper) RemoteAddr() gonet.Addr {
+	return nil
+}
+
+func (pc *streamPacketConnWrapper) Read(p []byte) (n int, err error) {
+	pc.readAccess.Lock()
+	defer pc.readAccess.Unlock()
+	for len(pc.readBuffer) == 0 {
+		if pc.packetRead == nil {
+			pc.packetRead = make([]byte, maxStreamPacketAddrSegmentLen)
+		}
+		var addr gonet.Addr
+		n, addr, err = pc.PacketConn.ReadFrom(pc.packetRead)
+		if err != nil {
+			return 0, err
+		}
+		packet, err := AttachAddressToPacket(buf.FromBytes(pc.packetRead[:n]), addr)
+		if err != nil {
+			return 0, err
+		}
+		frame, err := AttachLengthToPacket(packet)
+		if err != nil {
+			return 0, err
+		}
+		pc.readBuffer = append(pc.readBuffer[:0], frame.Bytes()...)
+		frame.Release()
+	}
+	n = copy(p, pc.readBuffer)
+	pc.readBuffer = pc.readBuffer[n:]
+	return n, nil
+}
+
+func (pc *streamPacketConnWrapper) Write(p []byte) (n int, err error) {
+	pc.writeAccess.Lock()
+	defer pc.writeAccess.Unlock()
+	pc.writeBuffer = append(pc.writeBuffer, p...)
+	for len(pc.writeBuffer) >= streamPacketLengthSize {
+		length := int(pc.writeBuffer[0])<<8 | int(pc.writeBuffer[1])
+		if length == 0 {
+			return 0, errors.New("invalid empty packetaddr segment")
+		}
+		frameEnd := streamPacketLengthSize + length
+		if len(pc.writeBuffer) < frameEnd {
+			break
+		}
+		data, addr, err := ExtractAddressFromPacket(buf.FromBytes(pc.writeBuffer[streamPacketLengthSize:frameEnd]))
+		if err != nil {
+			return 0, err
+		}
+		if _, err = pc.PacketConn.WriteTo(data.Bytes(), addr); err != nil {
+			data.Release()
+			return 0, err
+		}
+		data.Release()
+		pc.writeBuffer = pc.writeBuffer[frameEnd:]
+	}
+	if len(pc.writeBuffer) == 0 && cap(pc.writeBuffer) > buf.Size {
+		pc.writeBuffer = nil
+	}
+	return len(p), nil
+}
+
+func (pc *streamPacketConnWrapper) Close() error {
+	return pc.PacketConn.Close()
+}
+
 func (pc *packetConnWrapper) Close() error {
 	return pc.PacketConn.Close()
 }
@@ -179,6 +299,8 @@ func GetDestinationSubsetOf(dest net.Destination) (bool, error) {
 	switch dest.Address.Domain() {
 	case seqPacketMagicAddress:
 		return false, nil
+	case streamPacketMagicAddress:
+		return true, nil
 	default:
 		return false, errNotPacketConn
 	}

common/net/packetaddr/packetaddr.go

@@ -2,6 +2,8 @@ package packetaddr
 
 import (
 	"bytes"
+	"encoding/binary"
+	"io"
 	gonet "net"
 
 	"github.com/v2fly/v2ray-core/v5/common/buf"
@@ -15,28 +17,83 @@ var addrParser = protocol.NewAddressParser(
 	protocol.AddressFamilyByte(0x02, net.AddressFamilyIPv6),
 )
 
+const (
+	streamPacketLengthSize        = 2
+	maxPacketAddrHeaderSize       = 1 + 16 + 2
+	maxStreamPacketAddrSegmentLen = 0xffff
+)
+
 // AttachAddressToPacket
 // relinquish ownership of data
 // gain ownership of the returning value
 func AttachAddressToPacket(data *buf.Buffer, address gonet.Addr) (*buf.Buffer, error) {
-	packetBuf := buf.New()
 	udpaddr := address.(*gonet.UDPAddr)
+	packetBuf := buf.NewWithSize(data.Len() + maxPacketAddrHeaderSize)
 	port, err := net.PortFromInt(uint32(udpaddr.Port))
 	if err != nil {
+		packetBuf.Release()
+		data.Release()
 		return nil, err
 	}
 	err = addrParser.WriteAddressPort(packetBuf, net.IPAddress(udpaddr.IP), port)
 	if err != nil {
+		packetBuf.Release()
+		data.Release()
 		return nil, err
 	}
-	_, err = packetBuf.Write(data.Bytes())
-	if err != nil {
+	if n, err := packetBuf.Write(data.Bytes()); err != nil {
+		packetBuf.Release()
+		data.Release()
 		return nil, err
+	} else if n != int(data.Len()) {
+		packetBuf.Release()
+		data.Release()
+		return nil, errors.New("failed to write full packet payload")
 	}
 	data.Release()
 	return packetBuf, nil
 }
 
+// AttachLengthToPacket prefixes a packetaddr segment with its big-endian uint16 length.
+// relinquish ownership of data
+// gain ownership of the returning value
+func AttachLengthToPacket(data *buf.Buffer) (*buf.Buffer, error) {
+	if data.Len() > maxStreamPacketAddrSegmentLen {
+		data.Release()
+		return nil, errors.New("packetaddr segment too large")
+	}
+	packetBuf := buf.NewWithSize(data.Len() + streamPacketLengthSize)
+	binary.BigEndian.PutUint16(packetBuf.Extend(streamPacketLengthSize), uint16(data.Len()))
+	if n, err := packetBuf.Write(data.Bytes()); err != nil {
+		packetBuf.Release()
+		data.Release()
+		return nil, err
+	} else if n != int(data.Len()) {
+		packetBuf.Release()
+		data.Release()
+		return nil, errors.New("failed to write full packetaddr segment")
+	}
+	data.Release()
+	return packetBuf, nil
+}
+
+func ExtractPacketFromStream(reader io.Reader) (*buf.Buffer, error) {
+	var lengthBuf [streamPacketLengthSize]byte
+	if _, err := io.ReadFull(reader, lengthBuf[:]); err != nil {
+		return nil, err
+	}
+	length := binary.BigEndian.Uint16(lengthBuf[:])
+	if length == 0 {
+		return nil, errors.New("invalid empty packetaddr segment")
+	}
+	packetBuf := buf.NewWithSize(int32(length))
+	if _, err := packetBuf.ReadFullFrom(reader, int32(length)); err != nil {
+		packetBuf.Release()
+		return nil, err
+	}
+	return packetBuf, nil
+}
+
 // ExtractAddressFromPacket
 // relinquish ownership of data
 // gain ownership of the returning value

common/net/packetaddr/packetaddr_test.go

@@ -1,12 +1,17 @@
 package packetaddr
 
 import (
+	"bytes"
+	"encoding/binary"
+	"io"
 	sysnet "net"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 
 	"github.com/v2fly/v2ray-core/v5/common/buf"
+	v2net "github.com/v2fly/v2ray-core/v5/common/net"
 )
 
 func TestPacketEncodingIPv4(t *testing.T) {
@@ -40,3 +45,156 @@ func TestPacketEncodingIPv6(t *testing.T) {
 	assert.Equal(t, packetPayload.Bytes(), packetData[:])
 	assert.Equal(t, packetAddress, decodedAddress)
 }
+
+func TestStreamPacketEncodingFragmented(t *testing.T) {
+	packetAddress := &sysnet.UDPAddr{
+		IP:   sysnet.IPv4(1, 2, 3, 4).To4(),
+		Port: 1234,
+	}
+	packetData := []byte("hello over streamaddr")
+	wrapped, err := AttachAddressToPacket(buf.FromBytes(packetData), packetAddress)
+	assert.NoError(t, err)
+	framed, err := AttachLengthToPacket(wrapped)
+	assert.NoError(t, err)
+	frameBytes := append([]byte(nil), framed.Bytes()...)
+	framed.Release()
+
+	reader := io.MultiReader(
+		bytes.NewReader(frameBytes[:1]),
+		bytes.NewReader(frameBytes[1:3]),
+		bytes.NewReader(frameBytes[3:]),
+	)
+	packetPayload, err := ExtractPacketFromStream(reader)
+	assert.NoError(t, err)
+
+	decodedPayload, decodedAddress, err := ExtractAddressFromPacket(packetPayload)
+	assert.NoError(t, err)
+
+	assert.Equal(t, packetData, decodedPayload.Bytes())
+	assert.Equal(t, packetAddress, decodedAddress)
+}
+
+func TestStreamPacketEncodingRejectsEmptyFrame(t *testing.T) {
+	_, err := ExtractPacketFromStream(bytes.NewReader([]byte{0, 0}))
+	assert.Error(t, err)
+}
+
+func TestStreamPacketConnWrapperWriteFragmented(t *testing.T) {
+	conn := newTestPacketConn()
+	wrapper := &streamPacketConnWrapper{PacketConn: conn}
+	packetAddress := &sysnet.UDPAddr{
+		IP:   sysnet.IPv4(1, 2, 3, 4).To4(),
+		Port: 1234,
+	}
+	packetData := []byte("fragmented write")
+	wrapped, err := AttachAddressToPacket(buf.FromBytes(packetData), packetAddress)
+	assert.NoError(t, err)
+	framed, err := AttachLengthToPacket(wrapped)
+	assert.NoError(t, err)
+	frameBytes := append([]byte(nil), framed.Bytes()...)
+	framed.Release()
+
+	n, err := wrapper.Write(frameBytes[:3])
+	assert.NoError(t, err)
+	assert.Equal(t, 3, n)
+	select {
+	case <-conn.writeCh:
+		t.Fatal("unexpected UDP write before a complete stream frame")
+	default:
+	}
+
+	n, err = wrapper.Write(frameBytes[3:])
+	assert.NoError(t, err)
+	assert.Equal(t, len(frameBytes)-3, n)
+	written := <-conn.writeCh
+	assert.Equal(t, packetData, written.payload)
+	assert.Equal(t, packetAddress, written.addr)
+}
+
+func TestStreamPacketConnWrapperReadPartial(t *testing.T) {
+	conn := newTestPacketConn()
+	wrapper := &streamPacketConnWrapper{PacketConn: conn}
+	packetAddress := &sysnet.UDPAddr{
+		IP:   sysnet.IPv4(1, 2, 3, 4).To4(),
+		Port: 1234,
+	}
+	packetData := []byte("partial read")
+	conn.readCh <- testPacket{payload: packetData, addr: packetAddress}
+
+	var streamData []byte
+	readBuf := make([]byte, 3)
+	for len(streamData) < streamPacketLengthSize || len(streamData) < streamPacketLengthSize+int(binary.BigEndian.Uint16(streamData[:streamPacketLengthSize])) {
+		n, err := wrapper.Read(readBuf)
+		assert.NoError(t, err)
+		streamData = append(streamData, readBuf[:n]...)
+	}
+
+	packetPayload, err := ExtractPacketFromStream(bytes.NewReader(streamData))
+	assert.NoError(t, err)
+	decodedPayload, decodedAddress, err := ExtractAddressFromPacket(packetPayload)
+	assert.NoError(t, err)
+	assert.Equal(t, packetData, decodedPayload.Bytes())
+	assert.Equal(t, packetAddress, decodedAddress)
+}
+
+func TestGetDestinationSubsetOfStream(t *testing.T) {
+	isStream, err := GetDestinationSubsetOf(v2net.Destination{
+		Network: v2net.Network_TCP,
+		Address: v2net.DomainAddress(streamPacketMagicAddress),
+		Port:    0,
+	})
+	assert.NoError(t, err)
+	assert.True(t, isStream)
+}
+
+type testPacket struct {
+	payload []byte
+	addr    sysnet.Addr
+}
+
+type testPacketConn struct {
+	readCh  chan testPacket
+	writeCh chan testPacket
+}
+
+func newTestPacketConn() *testPacketConn {
+	return &testPacketConn{
+		readCh:  make(chan testPacket, 1),
+		writeCh: make(chan testPacket, 1),
+	}
+}
+
+func (c *testPacketConn) ReadFrom(p []byte) (int, sysnet.Addr, error) {
+	packet, ok := <-c.readCh
+	if !ok {
+		return 0, nil, io.EOF
+	}
+	return copy(p, packet.payload), packet.addr, nil
+}
+
+func (c *testPacketConn) WriteTo(p []byte, addr sysnet.Addr) (int, error) {
+	copied := append([]byte(nil), p...)
+	c.writeCh <- testPacket{payload: copied, addr: addr}
+	return len(p), nil
+}
+
+func (c *testPacketConn) Close() error {
+	close(c.readCh)
+	return nil
+}
+
+func (c *testPacketConn) LocalAddr() sysnet.Addr {
+	return &sysnet.UDPAddr{}
+}
+
+func (c *testPacketConn) SetDeadline(time.Time) error {
+	return nil
+}
+
+func (c *testPacketConn) SetReadDeadline(time.Time) error {
+	return nil
+}
+
+func (c *testPacketConn) SetWriteDeadline(time.Time) error {
+	return nil
+}

common/net/packetaddr/special_address.go

@@ -1,3 +1,6 @@
 package packetaddr
 
-const seqPacketMagicAddress = "sp.packet-addr.v2fly.arpa"
+const (
+	seqPacketMagicAddress    = "sp.packet-addr.v2fly.arpa"
+	streamPacketMagicAddress = "st.packet-addr.v2fly.arpa"
+)

infra/conf/v4/hysteria2.go

@@ -71,6 +71,8 @@ func (c *Hysteria2ServerConfig) Build() (proto.Message, error) {
 	switch c.PacketEncoding {
 	case "Packet":
 		config.PacketEncoding = packetaddr.PacketAddrType_Packet
+	case "Stream":
+		config.PacketEncoding = packetaddr.PacketAddrType_Stream
 	case "", "None":
 		config.PacketEncoding = packetaddr.PacketAddrType_None
 	}

infra/conf/v4/shadowsocks.go

@@ -47,6 +47,8 @@ func (v *ShadowsocksServerConfig) Build() (proto.Message, error) {
 	switch v.PacketEncoding {
 	case "Packet":
 		config.PacketEncoding = packetaddr.PacketAddrType_Packet
+	case "Stream":
+		config.PacketEncoding = packetaddr.PacketAddrType_Stream
 	case "", "None":
 		config.PacketEncoding = packetaddr.PacketAddrType_None
 	}

infra/conf/v4/socks.go

@@ -70,6 +70,8 @@ func (v *SocksServerConfig) Build() (proto.Message, error) {
 	switch v.PacketEncoding {
 	case "Packet":
 		config.PacketEncoding = packetaddr.PacketAddrType_Packet
+	case "Stream":
+		config.PacketEncoding = packetaddr.PacketAddrType_Stream
 	case "", "None":
 		config.PacketEncoding = packetaddr.PacketAddrType_None
 	}

infra/conf/v4/trojan.go

@@ -172,6 +172,8 @@ func (c *TrojanServerConfig) Build() (proto.Message, error) {
 	switch c.PacketEncoding {
 	case "Packet":
 		config.PacketEncoding = packetaddr.PacketAddrType_Packet
+	case "Stream":
+		config.PacketEncoding = packetaddr.PacketAddrType_Stream
 	case "", "None":
 		config.PacketEncoding = packetaddr.PacketAddrType_None
 	}

proxy/hysteria2/server.go

@@ -177,6 +177,9 @@ func (s *Server) handleUDPPayload(ctx context.Context, clientReader *PacketReade
 	case packetaddr.PacketAddrType_Packet:
 		packetAddrDispatcherFactory := udp.NewPacketAddrDispatcherCreator(ctx)
 		udpDispatcherConstructor = packetAddrDispatcherFactory.NewPacketAddrDispatcher
+	case packetaddr.PacketAddrType_Stream:
+		packetAddrDispatcherFactory := udp.NewStreamPacketAddrDispatcherCreator(ctx)
+		udpDispatcherConstructor = packetAddrDispatcherFactory.NewPacketAddrDispatcher
 	}
 
 	udpServer := udpDispatcherConstructor(dispatcher, func(ctx context.Context, packet *udp_proto.Packet) {

proxy/shadowsocks/server.go

@@ -78,6 +78,9 @@ func (s *Server) handlerUDPPayload(ctx context.Context, conn internet.Connection
 	case packetaddr.PacketAddrType_Packet:
 		packetAddrDispatcherFactory := udp.NewPacketAddrDispatcherCreator(ctx)
 		udpDispatcherConstructor = packetAddrDispatcherFactory.NewPacketAddrDispatcher
+	case packetaddr.PacketAddrType_Stream:
+		packetAddrDispatcherFactory := udp.NewStreamPacketAddrDispatcherCreator(ctx)
+		udpDispatcherConstructor = packetAddrDispatcherFactory.NewPacketAddrDispatcher
 	}
 
 	udpServer := udpDispatcherConstructor(dispatcher, func(ctx context.Context, packet *udp_proto.Packet) {

proxy/socks/server.go

@@ -228,6 +228,9 @@ func (s *Server) handleUDPPayload(ctx context.Context, conn internet.Connection,
 	case packetaddr.PacketAddrType_Packet:
 		packetAddrDispatcherFactory := udp.NewPacketAddrDispatcherCreator(ctx)
 		udpDispatcherConstructor = packetAddrDispatcherFactory.NewPacketAddrDispatcher
+	case packetaddr.PacketAddrType_Stream:
+		packetAddrDispatcherFactory := udp.NewStreamPacketAddrDispatcherCreator(ctx)
+		udpDispatcherConstructor = packetAddrDispatcherFactory.NewPacketAddrDispatcher
 	}
 	udpServer := udpDispatcherConstructor(dispatcher, func(ctx context.Context, packet *udp_proto.Packet) {
 		payload := packet.Payload

proxy/trojan/server.go

@@ -213,6 +213,9 @@ func (s *Server) handleUDPPayload(ctx context.Context, clientReader *PacketReade
 	case packetaddr.PacketAddrType_Packet:
 		packetAddrDispatcherFactory := udp.NewPacketAddrDispatcherCreator(ctx)
 		udpDispatcherConstructor = packetAddrDispatcherFactory.NewPacketAddrDispatcher
+	case packetaddr.PacketAddrType_Stream:
+		packetAddrDispatcherFactory := udp.NewStreamPacketAddrDispatcherCreator(ctx)
+		udpDispatcherConstructor = packetAddrDispatcherFactory.NewPacketAddrDispatcher
 	}
 
 	udpServer := udpDispatcherConstructor(dispatcher, func(ctx context.Context, packet *udp_proto.Packet) {

transport/internet/udp/dispatcher_packetaddr.go

@@ -49,17 +49,22 @@ func (p PacketAddrDispatcher) readWorker() {
 }
 
 type PacketAddrDispatcherCreator struct {
-	ctx context.Context
+	ctx      context.Context
+	isStream bool
 }
 
 func NewPacketAddrDispatcherCreator(ctx context.Context) PacketAddrDispatcherCreator {
 	return PacketAddrDispatcherCreator{ctx: ctx}
 }
 
+func NewStreamPacketAddrDispatcherCreator(ctx context.Context) PacketAddrDispatcherCreator {
+	return PacketAddrDispatcherCreator{ctx: ctx, isStream: true}
+}
+
 func (pdc *PacketAddrDispatcherCreator) NewPacketAddrDispatcher(
 	dispatcher routing.Dispatcher, callback ResponseCallback,
 ) DispatcherI {
-	packetConn, _ := packetaddr.CreatePacketAddrConn(pdc.ctx, dispatcher, false)
+	packetConn, _ := packetaddr.CreatePacketAddrConn(pdc.ctx, dispatcher, pdc.isStream)
 	pd := &PacketAddrDispatcher{conn: packetConn, callback: callback, ctx: pdc.ctx}
 	go pd.readWorker()
 	return pd