Improve UDP batch support

This commit is contained in:
世界
2026-08-30 17:41:40 +08:00
parent 9ab4c0f5cb
commit 2bbff23640
28 changed files with 304 additions and 205 deletions
+6 -32
View File
@@ -5,57 +5,31 @@ import (
"net" "net"
"github.com/sagernet/sing/common/buf" "github.com/sagernet/sing/common/buf"
E "github.com/sagernet/sing/common/exceptions"
M "github.com/sagernet/sing/common/metadata" M "github.com/sagernet/sing/common/metadata"
N "github.com/sagernet/sing/common/network" N "github.com/sagernet/sing/common/network"
) )
// Deprecated
type ConnectionHandler interface { type ConnectionHandler interface {
NewConnection(ctx context.Context, conn net.Conn, metadata InboundContext) error NewConnection(ctx context.Context, conn net.Conn, metadata InboundContext, onClose N.CloseHandlerFunc)
} }
type ConnectionHandlerEx interface {
NewConnectionEx(ctx context.Context, conn net.Conn, metadata InboundContext, onClose N.CloseHandlerFunc)
}
// Deprecated: use PacketHandlerEx instead
type PacketHandler interface { type PacketHandler interface {
NewPacket(ctx context.Context, conn N.PacketConn, buffer *buf.Buffer, metadata InboundContext) error NewPacket(buffer *buf.Buffer, source M.Socksaddr)
} }
type PacketHandlerEx interface { type PacketBatchHandler interface {
NewPacketEx(buffer *buf.Buffer, source M.Socksaddr) NewPacketBatch(buffers []*buf.Buffer, sources []M.Socksaddr)
} }
// Deprecated: use OOBPacketHandlerEx instead
type OOBPacketHandler interface { type OOBPacketHandler interface {
NewPacket(ctx context.Context, conn N.PacketConn, buffer *buf.Buffer, oob []byte, metadata InboundContext) error NewPacket(buffer *buf.Buffer, oob []byte, source M.Socksaddr)
} }
type OOBPacketHandlerEx interface {
NewPacketEx(buffer *buf.Buffer, oob []byte, source M.Socksaddr)
}
// Deprecated
type PacketConnectionHandler interface { type PacketConnectionHandler interface {
NewPacketConnection(ctx context.Context, conn N.PacketConn, metadata InboundContext) error NewPacketConnection(ctx context.Context, conn N.PacketConn, metadata InboundContext, onClose N.CloseHandlerFunc)
} }
type PacketConnectionHandlerEx interface {
NewPacketConnectionEx(ctx context.Context, conn N.PacketConn, metadata InboundContext, onClose N.CloseHandlerFunc)
}
// Deprecated: use TCPConnectionHandlerEx instead
//
//nolint:staticcheck
type UpstreamHandlerAdapter interface { type UpstreamHandlerAdapter interface {
N.TCPConnectionHandler
N.UDPConnectionHandler
E.Handler
}
type UpstreamHandlerAdapterEx interface {
N.TCPConnectionHandlerEx N.TCPConnectionHandlerEx
N.UDPConnectionHandlerEx N.UDPConnectionHandlerEx
} }
+2 -2
View File
@@ -22,12 +22,12 @@ type Inbound interface {
type TCPInjectableInbound interface { type TCPInjectableInbound interface {
Inbound Inbound
ConnectionHandlerEx ConnectionHandler
} }
type UDPInjectableInbound interface { type UDPInjectableInbound interface {
Inbound Inbound
PacketConnectionHandlerEx PacketConnectionHandler
} }
type InboundRegistry interface { type InboundRegistry interface {
+38 -38
View File
@@ -9,31 +9,31 @@ import (
) )
type ( type (
ConnectionHandlerFuncEx = func(ctx context.Context, conn net.Conn, metadata InboundContext, onClose N.CloseHandlerFunc) ConnectionHandlerFunc = func(ctx context.Context, conn net.Conn, metadata InboundContext, onClose N.CloseHandlerFunc)
PacketConnectionHandlerFuncEx = func(ctx context.Context, conn N.PacketConn, metadata InboundContext, onClose N.CloseHandlerFunc) PacketConnectionHandlerFunc = func(ctx context.Context, conn N.PacketConn, metadata InboundContext, onClose N.CloseHandlerFunc)
) )
func NewUpstreamHandlerEx( func NewUpstreamHandler(
metadata InboundContext, metadata InboundContext,
connectionHandler ConnectionHandlerFuncEx, connectionHandler ConnectionHandlerFunc,
packetHandler PacketConnectionHandlerFuncEx, packetHandler PacketConnectionHandlerFunc,
) UpstreamHandlerAdapterEx { ) UpstreamHandlerAdapter {
return &myUpstreamHandlerWrapperEx{ return &myUpstreamHandlerWrapper{
metadata: metadata, metadata: metadata,
connectionHandler: connectionHandler, connectionHandler: connectionHandler,
packetHandler: packetHandler, packetHandler: packetHandler,
} }
} }
var _ UpstreamHandlerAdapterEx = (*myUpstreamHandlerWrapperEx)(nil) var _ UpstreamHandlerAdapter = (*myUpstreamHandlerWrapper)(nil)
type myUpstreamHandlerWrapperEx struct { type myUpstreamHandlerWrapper struct {
metadata InboundContext metadata InboundContext
connectionHandler ConnectionHandlerFuncEx connectionHandler ConnectionHandlerFunc
packetHandler PacketConnectionHandlerFuncEx packetHandler PacketConnectionHandlerFunc
} }
func (w *myUpstreamHandlerWrapperEx) NewConnectionEx(ctx context.Context, conn net.Conn, source M.Socksaddr, destination M.Socksaddr, onClose N.CloseHandlerFunc) { func (w *myUpstreamHandlerWrapper) NewConnectionEx(ctx context.Context, conn net.Conn, source M.Socksaddr, destination M.Socksaddr, onClose N.CloseHandlerFunc) {
myMetadata := w.metadata myMetadata := w.metadata
if source.IsValid() { if source.IsValid() {
myMetadata.Source = source myMetadata.Source = source
@@ -44,7 +44,7 @@ func (w *myUpstreamHandlerWrapperEx) NewConnectionEx(ctx context.Context, conn n
w.connectionHandler(ctx, conn, myMetadata, onClose) w.connectionHandler(ctx, conn, myMetadata, onClose)
} }
func (w *myUpstreamHandlerWrapperEx) NewPacketConnectionEx(ctx context.Context, conn N.PacketConn, source M.Socksaddr, destination M.Socksaddr, onClose N.CloseHandlerFunc) { func (w *myUpstreamHandlerWrapper) NewPacketConnectionEx(ctx context.Context, conn N.PacketConn, source M.Socksaddr, destination M.Socksaddr, onClose N.CloseHandlerFunc) {
myMetadata := w.metadata myMetadata := w.metadata
if source.IsValid() { if source.IsValid() {
myMetadata.Source = source myMetadata.Source = source
@@ -55,24 +55,24 @@ func (w *myUpstreamHandlerWrapperEx) NewPacketConnectionEx(ctx context.Context,
w.packetHandler(ctx, conn, myMetadata, onClose) w.packetHandler(ctx, conn, myMetadata, onClose)
} }
var _ UpstreamHandlerAdapterEx = (*myUpstreamContextHandlerWrapperEx)(nil) var _ UpstreamHandlerAdapter = (*myUpstreamContextHandlerWrapper)(nil)
type myUpstreamContextHandlerWrapperEx struct { type myUpstreamContextHandlerWrapper struct {
connectionHandler ConnectionHandlerFuncEx connectionHandler ConnectionHandlerFunc
packetHandler PacketConnectionHandlerFuncEx packetHandler PacketConnectionHandlerFunc
} }
func NewUpstreamContextHandlerEx( func NewUpstreamContextHandler(
connectionHandler ConnectionHandlerFuncEx, connectionHandler ConnectionHandlerFunc,
packetHandler PacketConnectionHandlerFuncEx, packetHandler PacketConnectionHandlerFunc,
) UpstreamHandlerAdapterEx { ) UpstreamHandlerAdapter {
return &myUpstreamContextHandlerWrapperEx{ return &myUpstreamContextHandlerWrapper{
connectionHandler: connectionHandler, connectionHandler: connectionHandler,
packetHandler: packetHandler, packetHandler: packetHandler,
} }
} }
func (w *myUpstreamContextHandlerWrapperEx) NewConnectionEx(ctx context.Context, conn net.Conn, source M.Socksaddr, destination M.Socksaddr, onClose N.CloseHandlerFunc) { func (w *myUpstreamContextHandlerWrapper) NewConnectionEx(ctx context.Context, conn net.Conn, source M.Socksaddr, destination M.Socksaddr, onClose N.CloseHandlerFunc) {
_, myMetadata := ExtendContext(ctx) _, myMetadata := ExtendContext(ctx)
if source.IsValid() { if source.IsValid() {
myMetadata.Source = source myMetadata.Source = source
@@ -83,7 +83,7 @@ func (w *myUpstreamContextHandlerWrapperEx) NewConnectionEx(ctx context.Context,
w.connectionHandler(ctx, conn, *myMetadata, onClose) w.connectionHandler(ctx, conn, *myMetadata, onClose)
} }
func (w *myUpstreamContextHandlerWrapperEx) NewPacketConnectionEx(ctx context.Context, conn N.PacketConn, source M.Socksaddr, destination M.Socksaddr, onClose N.CloseHandlerFunc) { func (w *myUpstreamContextHandlerWrapper) NewPacketConnectionEx(ctx context.Context, conn N.PacketConn, source M.Socksaddr, destination M.Socksaddr, onClose N.CloseHandlerFunc) {
_, myMetadata := ExtendContext(ctx) _, myMetadata := ExtendContext(ctx)
if source.IsValid() { if source.IsValid() {
myMetadata.Source = source myMetadata.Source = source
@@ -94,24 +94,24 @@ func (w *myUpstreamContextHandlerWrapperEx) NewPacketConnectionEx(ctx context.Co
w.packetHandler(ctx, conn, *myMetadata, onClose) w.packetHandler(ctx, conn, *myMetadata, onClose)
} }
func NewRouteHandlerEx( func NewRouteHandler(
metadata InboundContext, metadata InboundContext,
router ConnectionRouterEx, router ConnectionRouterEx,
) UpstreamHandlerAdapterEx { ) UpstreamHandlerAdapter {
return &routeHandlerWrapperEx{ return &routeHandlerWrapper{
metadata: metadata, metadata: metadata,
router: router, router: router,
} }
} }
var _ UpstreamHandlerAdapterEx = (*routeHandlerWrapperEx)(nil) var _ UpstreamHandlerAdapter = (*routeHandlerWrapper)(nil)
type routeHandlerWrapperEx struct { type routeHandlerWrapper struct {
metadata InboundContext metadata InboundContext
router ConnectionRouterEx router ConnectionRouterEx
} }
func (r *routeHandlerWrapperEx) NewConnectionEx(ctx context.Context, conn net.Conn, source M.Socksaddr, destination M.Socksaddr, onClose N.CloseHandlerFunc) { func (r *routeHandlerWrapper) NewConnectionEx(ctx context.Context, conn net.Conn, source M.Socksaddr, destination M.Socksaddr, onClose N.CloseHandlerFunc) {
if source.IsValid() { if source.IsValid() {
r.metadata.Source = source r.metadata.Source = source
} }
@@ -121,7 +121,7 @@ func (r *routeHandlerWrapperEx) NewConnectionEx(ctx context.Context, conn net.Co
r.router.RouteConnectionEx(ctx, conn, r.metadata, onClose) r.router.RouteConnectionEx(ctx, conn, r.metadata, onClose)
} }
func (r *routeHandlerWrapperEx) NewPacketConnectionEx(ctx context.Context, conn N.PacketConn, source M.Socksaddr, destination M.Socksaddr, onClose N.CloseHandlerFunc) { func (r *routeHandlerWrapper) NewPacketConnectionEx(ctx context.Context, conn N.PacketConn, source M.Socksaddr, destination M.Socksaddr, onClose N.CloseHandlerFunc) {
if source.IsValid() { if source.IsValid() {
r.metadata.Source = source r.metadata.Source = source
} }
@@ -131,21 +131,21 @@ func (r *routeHandlerWrapperEx) NewPacketConnectionEx(ctx context.Context, conn
r.router.RoutePacketConnectionEx(ctx, conn, r.metadata, onClose) r.router.RoutePacketConnectionEx(ctx, conn, r.metadata, onClose)
} }
func NewRouteContextHandlerEx( func NewRouteContextHandler(
router ConnectionRouterEx, router ConnectionRouterEx,
) UpstreamHandlerAdapterEx { ) UpstreamHandlerAdapter {
return &routeContextHandlerWrapperEx{ return &routeContextHandlerWrapper{
router: router, router: router,
} }
} }
var _ UpstreamHandlerAdapterEx = (*routeContextHandlerWrapperEx)(nil) var _ UpstreamHandlerAdapter = (*routeContextHandlerWrapper)(nil)
type routeContextHandlerWrapperEx struct { type routeContextHandlerWrapper struct {
router ConnectionRouterEx router ConnectionRouterEx
} }
func (r *routeContextHandlerWrapperEx) NewConnectionEx(ctx context.Context, conn net.Conn, source M.Socksaddr, destination M.Socksaddr, onClose N.CloseHandlerFunc) { func (r *routeContextHandlerWrapper) NewConnectionEx(ctx context.Context, conn net.Conn, source M.Socksaddr, destination M.Socksaddr, onClose N.CloseHandlerFunc) {
_, metadata := ExtendContext(ctx) _, metadata := ExtendContext(ctx)
if source.IsValid() { if source.IsValid() {
metadata.Source = source metadata.Source = source
@@ -156,7 +156,7 @@ func (r *routeContextHandlerWrapperEx) NewConnectionEx(ctx context.Context, conn
r.router.RouteConnectionEx(ctx, conn, *metadata, onClose) r.router.RouteConnectionEx(ctx, conn, *metadata, onClose)
} }
func (r *routeContextHandlerWrapperEx) NewPacketConnectionEx(ctx context.Context, conn N.PacketConn, source M.Socksaddr, destination M.Socksaddr, onClose N.CloseHandlerFunc) { func (r *routeContextHandlerWrapper) NewPacketConnectionEx(ctx context.Context, conn N.PacketConn, source M.Socksaddr, destination M.Socksaddr, onClose N.CloseHandlerFunc) {
_, metadata := ExtendContext(ctx) _, metadata := ExtendContext(ctx)
if source.IsValid() { if source.IsValid() {
metadata.Source = source metadata.Source = source
+59 -50
View File
@@ -12,21 +12,30 @@ import (
type ( type (
// Deprecated // Deprecated
ConnectionHandlerFunc = func(ctx context.Context, conn net.Conn, metadata InboundContext) error LegacyConnectionHandlerFunc = func(ctx context.Context, conn net.Conn, metadata InboundContext) error
// Deprecated // Deprecated
PacketConnectionHandlerFunc = func(ctx context.Context, conn N.PacketConn, metadata InboundContext) error LegacyPacketConnectionHandlerFunc = func(ctx context.Context, conn N.PacketConn, metadata InboundContext) error
) )
// Deprecated // Deprecated
// //
//nolint:staticcheck //nolint:staticcheck
func NewUpstreamHandler( type LegacyUpstreamHandlerAdapter interface {
N.TCPConnectionHandler
N.UDPConnectionHandler
E.Handler
}
// Deprecated
//
//nolint:staticcheck
func NewLegacyUpstreamHandler(
metadata InboundContext, metadata InboundContext,
connectionHandler ConnectionHandlerFunc, connectionHandler LegacyConnectionHandlerFunc,
packetHandler PacketConnectionHandlerFunc, packetHandler LegacyPacketConnectionHandlerFunc,
errorHandler E.Handler, errorHandler E.Handler,
) UpstreamHandlerAdapter { ) LegacyUpstreamHandlerAdapter {
return &myUpstreamHandlerWrapper{ return &legacyUpstreamHandlerWrapper{
metadata: metadata, metadata: metadata,
connectionHandler: connectionHandler, connectionHandler: connectionHandler,
packetHandler: packetHandler, packetHandler: packetHandler,
@@ -34,20 +43,20 @@ func NewUpstreamHandler(
} }
} }
var _ UpstreamHandlerAdapter = (*myUpstreamHandlerWrapper)(nil) var _ LegacyUpstreamHandlerAdapter = (*legacyUpstreamHandlerWrapper)(nil)
// Deprecated: use myUpstreamHandlerWrapperEx instead. // Deprecated: use NewUpstreamHandler instead.
// //
//nolint:staticcheck //nolint:staticcheck
type myUpstreamHandlerWrapper struct { type legacyUpstreamHandlerWrapper struct {
metadata InboundContext metadata InboundContext
connectionHandler ConnectionHandlerFunc connectionHandler LegacyConnectionHandlerFunc
packetHandler PacketConnectionHandlerFunc packetHandler LegacyPacketConnectionHandlerFunc
errorHandler E.Handler errorHandler E.Handler
} }
// Deprecated: use myUpstreamHandlerWrapperEx instead. // Deprecated: use NewUpstreamHandler instead.
func (w *myUpstreamHandlerWrapper) NewConnection(ctx context.Context, conn net.Conn, metadata M.Metadata) error { func (w *legacyUpstreamHandlerWrapper) NewConnection(ctx context.Context, conn net.Conn, metadata M.Metadata) error {
myMetadata := w.metadata myMetadata := w.metadata
if metadata.Source.IsValid() { if metadata.Source.IsValid() {
myMetadata.Source = metadata.Source myMetadata.Source = metadata.Source
@@ -58,8 +67,8 @@ func (w *myUpstreamHandlerWrapper) NewConnection(ctx context.Context, conn net.C
return w.connectionHandler(ctx, conn, myMetadata) return w.connectionHandler(ctx, conn, myMetadata)
} }
// Deprecated: use myUpstreamHandlerWrapperEx instead. // Deprecated: use NewUpstreamHandler instead.
func (w *myUpstreamHandlerWrapper) NewPacketConnection(ctx context.Context, conn N.PacketConn, metadata M.Metadata) error { func (w *legacyUpstreamHandlerWrapper) NewPacketConnection(ctx context.Context, conn N.PacketConn, metadata M.Metadata) error {
myMetadata := w.metadata myMetadata := w.metadata
if metadata.Source.IsValid() { if metadata.Source.IsValid() {
myMetadata.Source = metadata.Source myMetadata.Source = metadata.Source
@@ -70,8 +79,8 @@ func (w *myUpstreamHandlerWrapper) NewPacketConnection(ctx context.Context, conn
return w.packetHandler(ctx, conn, myMetadata) return w.packetHandler(ctx, conn, myMetadata)
} }
// Deprecated: use myUpstreamHandlerWrapperEx instead. // Deprecated: use NewUpstreamHandler instead.
func (w *myUpstreamHandlerWrapper) NewError(ctx context.Context, err error) { func (w *legacyUpstreamHandlerWrapper) NewError(ctx context.Context, err error) {
w.errorHandler.NewError(ctx, err) w.errorHandler.NewError(ctx, err)
} }
@@ -83,28 +92,28 @@ func UpstreamMetadata(metadata InboundContext) M.Metadata {
} }
} }
// Deprecated: Use NewUpstreamContextHandlerEx instead. // Deprecated: Use NewUpstreamContextHandler instead.
type myUpstreamContextHandlerWrapper struct { type legacyUpstreamContextHandlerWrapper struct {
connectionHandler ConnectionHandlerFunc connectionHandler LegacyConnectionHandlerFunc
packetHandler PacketConnectionHandlerFunc packetHandler LegacyPacketConnectionHandlerFunc
errorHandler E.Handler errorHandler E.Handler
} }
// Deprecated: Use NewUpstreamContextHandlerEx instead. // Deprecated: Use NewUpstreamContextHandler instead.
func NewUpstreamContextHandler( func NewLegacyUpstreamContextHandler(
connectionHandler ConnectionHandlerFunc, connectionHandler LegacyConnectionHandlerFunc,
packetHandler PacketConnectionHandlerFunc, packetHandler LegacyPacketConnectionHandlerFunc,
errorHandler E.Handler, errorHandler E.Handler,
) UpstreamHandlerAdapter { ) LegacyUpstreamHandlerAdapter {
return &myUpstreamContextHandlerWrapper{ return &legacyUpstreamContextHandlerWrapper{
connectionHandler: connectionHandler, connectionHandler: connectionHandler,
packetHandler: packetHandler, packetHandler: packetHandler,
errorHandler: errorHandler, errorHandler: errorHandler,
} }
} }
// Deprecated: Use NewUpstreamContextHandlerEx instead. // Deprecated: Use NewUpstreamContextHandler instead.
func (w *myUpstreamContextHandlerWrapper) NewConnection(ctx context.Context, conn net.Conn, metadata M.Metadata) error { func (w *legacyUpstreamContextHandlerWrapper) NewConnection(ctx context.Context, conn net.Conn, metadata M.Metadata) error {
myMetadata := ContextFrom(ctx) myMetadata := ContextFrom(ctx)
if metadata.Source.IsValid() { if metadata.Source.IsValid() {
myMetadata.Source = metadata.Source myMetadata.Source = metadata.Source
@@ -115,8 +124,8 @@ func (w *myUpstreamContextHandlerWrapper) NewConnection(ctx context.Context, con
return w.connectionHandler(ctx, conn, *myMetadata) return w.connectionHandler(ctx, conn, *myMetadata)
} }
// Deprecated: Use NewUpstreamContextHandlerEx instead. // Deprecated: Use NewUpstreamContextHandler instead.
func (w *myUpstreamContextHandlerWrapper) NewPacketConnection(ctx context.Context, conn N.PacketConn, metadata M.Metadata) error { func (w *legacyUpstreamContextHandlerWrapper) NewPacketConnection(ctx context.Context, conn N.PacketConn, metadata M.Metadata) error {
myMetadata := ContextFrom(ctx) myMetadata := ContextFrom(ctx)
if metadata.Source.IsValid() { if metadata.Source.IsValid() {
myMetadata.Source = metadata.Source myMetadata.Source = metadata.Source
@@ -127,18 +136,18 @@ func (w *myUpstreamContextHandlerWrapper) NewPacketConnection(ctx context.Contex
return w.packetHandler(ctx, conn, *myMetadata) return w.packetHandler(ctx, conn, *myMetadata)
} }
// Deprecated: Use NewUpstreamContextHandlerEx instead. // Deprecated: Use NewUpstreamContextHandler instead.
func (w *myUpstreamContextHandlerWrapper) NewError(ctx context.Context, err error) { func (w *legacyUpstreamContextHandlerWrapper) NewError(ctx context.Context, err error) {
w.errorHandler.NewError(ctx, err) w.errorHandler.NewError(ctx, err)
} }
// Deprecated: Use ConnectionRouterEx instead. // Deprecated: Use ConnectionRouterEx instead.
func NewRouteHandler( func NewLegacyRouteHandler(
metadata InboundContext, metadata InboundContext,
router ConnectionRouter, router ConnectionRouter,
logger logger.ContextLogger, logger logger.ContextLogger,
) UpstreamHandlerAdapter { ) LegacyUpstreamHandlerAdapter {
return &routeHandlerWrapper{ return &legacyRouteHandlerWrapper{
metadata: metadata, metadata: metadata,
router: router, router: router,
logger: logger, logger: logger,
@@ -146,29 +155,29 @@ func NewRouteHandler(
} }
// Deprecated: Use ConnectionRouterEx instead. // Deprecated: Use ConnectionRouterEx instead.
func NewRouteContextHandler( func NewLegacyRouteContextHandler(
router ConnectionRouter, router ConnectionRouter,
logger logger.ContextLogger, logger logger.ContextLogger,
) UpstreamHandlerAdapter { ) LegacyUpstreamHandlerAdapter {
return &routeContextHandlerWrapper{ return &legacyRouteContextHandlerWrapper{
router: router, router: router,
logger: logger, logger: logger,
} }
} }
var _ UpstreamHandlerAdapter = (*routeHandlerWrapper)(nil) var _ LegacyUpstreamHandlerAdapter = (*legacyRouteHandlerWrapper)(nil)
// Deprecated: Use ConnectionRouterEx instead. // Deprecated: Use ConnectionRouterEx instead.
// //
//nolint:staticcheck //nolint:staticcheck
type routeHandlerWrapper struct { type legacyRouteHandlerWrapper struct {
metadata InboundContext metadata InboundContext
router ConnectionRouter router ConnectionRouter
logger logger.ContextLogger logger logger.ContextLogger
} }
// Deprecated: Use ConnectionRouterEx instead. // Deprecated: Use ConnectionRouterEx instead.
func (w *routeHandlerWrapper) NewConnection(ctx context.Context, conn net.Conn, metadata M.Metadata) error { func (w *legacyRouteHandlerWrapper) NewConnection(ctx context.Context, conn net.Conn, metadata M.Metadata) error {
myMetadata := w.metadata myMetadata := w.metadata
if metadata.Source.IsValid() { if metadata.Source.IsValid() {
myMetadata.Source = metadata.Source myMetadata.Source = metadata.Source
@@ -180,7 +189,7 @@ func (w *routeHandlerWrapper) NewConnection(ctx context.Context, conn net.Conn,
} }
// Deprecated: Use ConnectionRouterEx instead. // Deprecated: Use ConnectionRouterEx instead.
func (w *routeHandlerWrapper) NewPacketConnection(ctx context.Context, conn N.PacketConn, metadata M.Metadata) error { func (w *legacyRouteHandlerWrapper) NewPacketConnection(ctx context.Context, conn N.PacketConn, metadata M.Metadata) error {
myMetadata := w.metadata myMetadata := w.metadata
if metadata.Source.IsValid() { if metadata.Source.IsValid() {
myMetadata.Source = metadata.Source myMetadata.Source = metadata.Source
@@ -192,20 +201,20 @@ func (w *routeHandlerWrapper) NewPacketConnection(ctx context.Context, conn N.Pa
} }
// Deprecated: Use ConnectionRouterEx instead. // Deprecated: Use ConnectionRouterEx instead.
func (w *routeHandlerWrapper) NewError(ctx context.Context, err error) { func (w *legacyRouteHandlerWrapper) NewError(ctx context.Context, err error) {
w.logger.ErrorContext(ctx, err) w.logger.ErrorContext(ctx, err)
} }
var _ UpstreamHandlerAdapter = (*routeContextHandlerWrapper)(nil) var _ LegacyUpstreamHandlerAdapter = (*legacyRouteContextHandlerWrapper)(nil)
// Deprecated: Use ConnectionRouterEx instead. // Deprecated: Use ConnectionRouterEx instead.
type routeContextHandlerWrapper struct { type legacyRouteContextHandlerWrapper struct {
router ConnectionRouter router ConnectionRouter
logger logger.ContextLogger logger logger.ContextLogger
} }
// Deprecated: Use ConnectionRouterEx instead. // Deprecated: Use ConnectionRouterEx instead.
func (w *routeContextHandlerWrapper) NewConnection(ctx context.Context, conn net.Conn, metadata M.Metadata) error { func (w *legacyRouteContextHandlerWrapper) NewConnection(ctx context.Context, conn net.Conn, metadata M.Metadata) error {
myMetadata := ContextFrom(ctx) myMetadata := ContextFrom(ctx)
if metadata.Source.IsValid() { if metadata.Source.IsValid() {
myMetadata.Source = metadata.Source myMetadata.Source = metadata.Source
@@ -217,7 +226,7 @@ func (w *routeContextHandlerWrapper) NewConnection(ctx context.Context, conn net
} }
// Deprecated: Use ConnectionRouterEx instead. // Deprecated: Use ConnectionRouterEx instead.
func (w *routeContextHandlerWrapper) NewPacketConnection(ctx context.Context, conn N.PacketConn, metadata M.Metadata) error { func (w *legacyRouteContextHandlerWrapper) NewPacketConnection(ctx context.Context, conn N.PacketConn, metadata M.Metadata) error {
myMetadata := ContextFrom(ctx) myMetadata := ContextFrom(ctx)
if metadata.Source.IsValid() { if metadata.Source.IsValid() {
myMetadata.Source = metadata.Source myMetadata.Source = metadata.Source
@@ -229,6 +238,6 @@ func (w *routeContextHandlerWrapper) NewPacketConnection(ctx context.Context, co
} }
// Deprecated: Use ConnectionRouterEx instead. // Deprecated: Use ConnectionRouterEx instead.
func (w *routeContextHandlerWrapper) NewError(ctx context.Context, err error) { func (w *legacyRouteContextHandlerWrapper) NewError(ctx context.Context, err error) {
w.logger.ErrorContext(ctx, err) w.logger.ErrorContext(ctx, err)
} }
+2 -1
View File
@@ -3,6 +3,7 @@ package interrupt
import ( import (
"net" "net"
"github.com/sagernet/sing/common/bufio"
"github.com/sagernet/sing/common/x/list" "github.com/sagernet/sing/common/x/list"
) )
@@ -71,5 +72,5 @@ func (c *PacketConn) WriterReplaceable() bool {
} }
func (c *PacketConn) Upstream() any { func (c *PacketConn) Upstream() any {
return c.PacketConn return bufio.NewPacketConn(c.PacketConn)
} }
+6 -6
View File
@@ -25,9 +25,9 @@ type Listener struct {
logger logger.ContextLogger logger logger.ContextLogger
network []string network []string
listenOptions option.ListenOptions listenOptions option.ListenOptions
connHandler adapter.ConnectionHandlerEx connHandler adapter.ConnectionHandler
packetHandler adapter.PacketHandlerEx packetHandler adapter.PacketHandler
oobPacketHandler adapter.OOBPacketHandlerEx oobPacketHandler adapter.OOBPacketHandler
threadUnsafePacketWriter bool threadUnsafePacketWriter bool
disablePacketOutput bool disablePacketOutput bool
setSystemProxy bool setSystemProxy bool
@@ -48,9 +48,9 @@ type Options struct {
Logger logger.ContextLogger Logger logger.ContextLogger
Network []string Network []string
Listen option.ListenOptions Listen option.ListenOptions
ConnectionHandler adapter.ConnectionHandlerEx ConnectionHandler adapter.ConnectionHandler
PacketHandler adapter.PacketHandlerEx PacketHandler adapter.PacketHandler
OOBPacketHandler adapter.OOBPacketHandlerEx OOBPacketHandler adapter.OOBPacketHandler
ThreadUnsafePacketWriter bool ThreadUnsafePacketWriter bool
DisablePacketOutput bool DisablePacketOutput bool
SetSystemProxy bool SetSystemProxy bool
+1 -1
View File
@@ -106,6 +106,6 @@ func (l *Listener) loopTCPIn() {
metadata.OriginDestination = M.SocksaddrFromNet(conn.LocalAddr()).Unwrap() metadata.OriginDestination = M.SocksaddrFromNet(conn.LocalAddr()).Unwrap()
ctx := log.ContextWithNewID(l.ctx) ctx := log.ContextWithNewID(l.ctx)
l.logger.InfoContext(ctx, "inbound connection from ", metadata.Source) l.logger.InfoContext(ctx, "inbound connection from ", metadata.Source)
go l.connHandler.NewConnectionEx(ctx, conn, metadata, nil) go l.connHandler.NewConnection(ctx, conn, metadata, nil)
} }
} }
+99 -17
View File
@@ -11,6 +11,7 @@ import (
"github.com/sagernet/sing-box/adapter" "github.com/sagernet/sing-box/adapter"
"github.com/sagernet/sing-box/common/redir" "github.com/sagernet/sing-box/common/redir"
"github.com/sagernet/sing/common/buf" "github.com/sagernet/sing/common/buf"
sBufio "github.com/sagernet/sing/common/bufio"
"github.com/sagernet/sing/common/control" "github.com/sagernet/sing/common/control"
E "github.com/sagernet/sing/common/exceptions" E "github.com/sagernet/sing/common/exceptions"
M "github.com/sagernet/sing/common/metadata" M "github.com/sagernet/sing/common/metadata"
@@ -18,6 +19,8 @@ import (
"github.com/sagernet/sing/service" "github.com/sagernet/sing/service"
) )
const udpOutputBatchSize = 128
func (l *Listener) ListenUDP() (net.PacketConn, error) { func (l *Listener) ListenUDP() (net.PacketConn, error) {
bindAddr := M.SocksaddrFrom(l.listenOptions.Listen.Build(netip.AddrFrom4([4]byte{127, 0, 0, 1})), l.listenOptions.ListenPort) bindAddr := M.SocksaddrFrom(l.listenOptions.Listen.Build(netip.AddrFrom4([4]byte{127, 0, 0, 1})), l.listenOptions.ListenPort)
var listenConfig net.ListenConfig var listenConfig net.ListenConfig
@@ -98,6 +101,15 @@ func (l *Listener) PacketWriter() N.PacketWriter {
func (l *Listener) loopUDPIn() { func (l *Listener) loopUDPIn() {
defer close(l.packetOutboundClosed) defer close(l.packetOutboundClosed)
if l.oobPacketHandler == nil {
if batchHandler, isBatchHandler := l.packetHandler.(adapter.PacketBatchHandler); isBatchHandler {
packetConn := sBufio.NewPacketConn(l.udpConn)
if readWaiter, created := sBufio.CreatePacketBatchReadWaiter(packetConn); created {
l.loopUDPInBatch(batchHandler, readWaiter)
return
}
}
}
var buffer *buf.Buffer var buffer *buf.Buffer
if !l.threadUnsafePacketWriter { if !l.threadUnsafePacketWriter {
buffer = buf.NewPacket() buffer = buf.NewPacket()
@@ -126,7 +138,7 @@ func (l *Listener) loopUDPIn() {
return return
} }
buffer.Truncate(n) buffer.Truncate(n)
l.oobPacketHandler.NewPacketEx(buffer, oob[:oobN], M.SocksaddrFromNetIP(addr).Unwrap()) l.oobPacketHandler.NewPacket(buffer, oob[:oobN], M.SocksaddrFromNetIP(addr).Unwrap())
} }
} else { } else {
for { for {
@@ -148,37 +160,82 @@ func (l *Listener) loopUDPIn() {
return return
} }
buffer.Truncate(n) buffer.Truncate(n)
l.packetHandler.NewPacketEx(buffer, M.SocksaddrFromNetIP(addr).Unwrap()) l.packetHandler.NewPacket(buffer, M.SocksaddrFromNetIP(addr).Unwrap())
} }
} }
} }
func (l *Listener) loopUDPInBatch(handler adapter.PacketBatchHandler, readWaiter N.PacketBatchReadWaiter) {
readWaitOptions := N.ReadWaitOptions{
BatchSize: sBufio.DefaultPacketReadBatchSize,
}
readWaiter.InitializeReadWaiter(readWaitOptions)
for {
buffers, sources, err := readWaiter.WaitReadPackets()
if err != nil {
buf.ReleaseMulti(buffers)
if l.shutdown.Load() && E.IsClosed(err) {
return
}
l.udpConn.Close()
l.logger.Error("udp listener closed: ", err)
return
}
handler.NewPacketBatch(buffers, sources)
}
}
func (l *Listener) loopUDPOut() { func (l *Listener) loopUDPOut() {
packetConn := sBufio.NewPacketConn(l.udpConn)
batchWriter := sBufio.NewPacketBatchWriter(packetConn)
packets := make([]*N.PacketBuffer, 0, udpOutputBatchSize)
buffers := make([]*buf.Buffer, 0, udpOutputBatchSize)
destinations := make([]M.Socksaddr, 0, udpOutputBatchSize)
for { for {
select { select {
case packet := <-l.packetOutbound: case packet := <-l.packetOutbound:
destination := packet.Destination.AddrPort() packets = append(packets, packet)
_, err := l.udpConn.WriteToUDPAddrPort(packet.Buffer.Bytes(), destination)
packet.Buffer.Release()
N.PutPacketBuffer(packet)
if err != nil {
if l.shutdown.Load() && E.IsClosed(err) {
return
}
l.logger.Error("udp listener write back: ", destination, ": ", err)
continue
}
continue
case <-l.packetOutboundClosed: case <-l.packetOutboundClosed:
l.releasePacketOutbound()
return
} }
for { drain:
for len(packets) < udpOutputBatchSize {
select { select {
case packet := <-l.packetOutbound: case packet := <-l.packetOutbound:
packet.Buffer.Release() packets = append(packets, packet)
N.PutPacketBuffer(packet)
default: default:
break drain
}
}
for _, packet := range packets {
buffers = append(buffers, packet.Buffer)
destinations = append(destinations, packet.Destination)
}
err := batchWriter.WritePacketBatch(buffers, destinations)
for _, packet := range packets {
N.PutPacketBuffer(packet)
}
packets = packets[:0]
buffers = buffers[:0]
destinations = destinations[:0]
if err != nil {
if l.shutdown.Load() && E.IsClosed(err) {
return return
} }
l.logger.Error("udp listener write back: ", err)
}
}
}
func (l *Listener) releasePacketOutbound() {
for {
select {
case packet := <-l.packetOutbound:
packet.Buffer.Release()
N.PutPacketBuffer(packet)
default:
return
} }
} }
} }
@@ -203,5 +260,30 @@ func (w *packetWriter) WritePacket(buffer *buf.Buffer, destination M.Socksaddr)
} }
} }
func (w *packetWriter) WritePacketBatch(buffers []*buf.Buffer, destinations []M.Socksaddr) error {
if len(buffers) == 0 || len(buffers) != len(destinations) {
buf.ReleaseMulti(buffers)
return os.ErrInvalid
}
for index, buffer := range buffers {
packet := N.NewPacketBuffer()
packet.Buffer = buffer
packet.Destination = destinations[index]
select {
case w.packetOutbound <- packet:
default:
buffer.Release()
N.PutPacketBuffer(packet)
buf.ReleaseMulti(buffers[index+1:])
if w.shutdown.Load() {
return os.ErrClosed
}
w.logger.Trace("dropped packet batch to ", destinations[index])
return nil
}
}
return nil
}
func (w *packetWriter) WriteIsThreadUnsafe() { func (w *packetWriter) WriteIsThreadUnsafe() {
} }
+1 -1
View File
@@ -42,7 +42,7 @@ func NewRouterWithOptions(router adapter.ConnectionRouterEx, logger logger.Conte
return log.ContextWithNewID(ctx) return log.ContextWithNewID(ctx)
}, },
Logger: logger, Logger: logger,
HandlerEx: adapter.NewRouteContextHandlerEx(router), HandlerEx: adapter.NewRouteContextHandler(router),
Padding: options.Padding, Padding: options.Padding,
Brutal: brutalOptions, Brutal: brutalOptions,
}) })
+1 -1
View File
@@ -96,7 +96,7 @@ func (h *Inbound) Close() error {
return common.Close(h.listener, h.tlsConfig) return common.Close(h.listener, h.tlsConfig)
} }
func (h *Inbound) NewConnectionEx(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) { func (h *Inbound) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
if h.tlsConfig != nil { if h.tlsConfig != nil {
tlsConn, err := tls.ServerHandshake(ctx, conn, h.tlsConfig) tlsConn, err := tls.ServerHandshake(ctx, conn, h.tlsConfig)
if err != nil { if err != nil {
+35 -2
View File
@@ -3,6 +3,7 @@ package direct
import ( import (
"context" "context"
"net" "net"
"os"
"time" "time"
"github.com/sagernet/sing-box/adapter" "github.com/sagernet/sing-box/adapter"
@@ -80,11 +81,15 @@ func (i *Inbound) Close() error {
return i.listener.Close() return i.listener.Close()
} }
func (i *Inbound) NewPacketEx(buffer *buf.Buffer, source M.Socksaddr) { func (i *Inbound) NewPacket(buffer *buf.Buffer, source M.Socksaddr) {
i.udpNat.NewPacket([][]byte{buffer.Bytes()}, source, i.listener.UDPAddr(), nil) i.udpNat.NewPacket([][]byte{buffer.Bytes()}, source, i.listener.UDPAddr(), nil)
} }
func (i *Inbound) NewConnectionEx(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) { func (i *Inbound) NewPacketBatch(buffers []*buf.Buffer, sources []M.Socksaddr) {
i.udpNat.NewPacketBatch(buffers, sources, i.listener.UDPAddr(), nil)
}
func (i *Inbound) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
metadata.Inbound = i.Tag() metadata.Inbound = i.Tag()
metadata.InboundType = i.Type() metadata.InboundType = i.Type()
destination := metadata.OriginDestination destination := metadata.OriginDestination
@@ -142,3 +147,31 @@ type directPacketWriter struct {
func (w *directPacketWriter) WritePacket(buffer *buf.Buffer, addr M.Socksaddr) error { func (w *directPacketWriter) WritePacket(buffer *buf.Buffer, addr M.Socksaddr) error {
return w.writer.WritePacket(buffer, w.source) return w.writer.WritePacket(buffer, w.source)
} }
func (w *directPacketWriter) CreatePacketBatchWriter() (N.PacketBatchWriter, bool) {
writer, created := bufio.CreatePacketBatchWriter(w.writer)
if !created {
return nil, false
}
return &directPacketBatchWriter{
writer: writer,
source: w.source,
}, true
}
type directPacketBatchWriter struct {
writer N.PacketBatchWriter
source M.Socksaddr
}
func (w *directPacketBatchWriter) WritePacketBatch(buffers []*buf.Buffer, destinations []M.Socksaddr) error {
if len(buffers) == 0 || len(buffers) != len(destinations) {
buf.ReleaseMulti(buffers)
return os.ErrInvalid
}
sources := make([]M.Socksaddr, len(destinations))
for index := range sources {
sources[index] = w.source
}
return w.writer.WritePacketBatch(buffers, sources)
}
+2 -2
View File
@@ -43,7 +43,7 @@ func (d *Outbound) ListenPacket(ctx context.Context, destination M.Socksaddr) (n
return nil, os.ErrInvalid return nil, os.ErrInvalid
} }
func (d *Outbound) NewConnectionEx(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) { func (d *Outbound) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
metadata.Destination = M.Socksaddr{} metadata.Destination = M.Socksaddr{}
for { for {
conn.SetReadDeadline(time.Now().Add(C.DNSTimeout)) conn.SetReadDeadline(time.Now().Add(C.DNSTimeout))
@@ -58,6 +58,6 @@ func (d *Outbound) NewConnectionEx(ctx context.Context, conn net.Conn, metadata
} }
} }
func (d *Outbound) NewPacketConnectionEx(ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) { func (d *Outbound) NewPacketConnection(ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
NewDNSPacketConnection(ctx, d.router, conn, nil, metadata) NewDNSPacketConnection(ctx, d.router, conn, nil, metadata)
} }
+9 -9
View File
@@ -25,9 +25,9 @@ func RegisterSelector(registry *outbound.Registry) {
} }
var ( var (
_ adapter.OutboundGroup = (*Selector)(nil) _ adapter.OutboundGroup = (*Selector)(nil)
_ adapter.ConnectionHandlerEx = (*Selector)(nil) _ adapter.ConnectionHandler = (*Selector)(nil)
_ adapter.PacketConnectionHandlerEx = (*Selector)(nil) _ adapter.PacketConnectionHandler = (*Selector)(nil)
) )
type Selector struct { type Selector struct {
@@ -156,21 +156,21 @@ func (s *Selector) ListenPacket(ctx context.Context, destination M.Socksaddr) (n
return s.interruptGroup.NewPacketConn(conn, interrupt.IsExternalConnectionFromContext(ctx)), nil return s.interruptGroup.NewPacketConn(conn, interrupt.IsExternalConnectionFromContext(ctx)), nil
} }
func (s *Selector) NewConnectionEx(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) { func (s *Selector) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
ctx = interrupt.ContextWithIsExternalConnection(ctx) ctx = interrupt.ContextWithIsExternalConnection(ctx)
selected := s.selected.Load() selected := s.selected.Load()
if outboundHandler, isHandler := selected.(adapter.ConnectionHandlerEx); isHandler { if outboundHandler, isHandler := selected.(adapter.ConnectionHandler); isHandler {
outboundHandler.NewConnectionEx(ctx, conn, metadata, onClose) outboundHandler.NewConnection(ctx, conn, metadata, onClose)
} else { } else {
s.connection.NewConnection(ctx, selected, conn, metadata, onClose) s.connection.NewConnection(ctx, selected, conn, metadata, onClose)
} }
} }
func (s *Selector) NewPacketConnectionEx(ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) { func (s *Selector) NewPacketConnection(ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
ctx = interrupt.ContextWithIsExternalConnection(ctx) ctx = interrupt.ContextWithIsExternalConnection(ctx)
selected := s.selected.Load() selected := s.selected.Load()
if outboundHandler, isHandler := selected.(adapter.PacketConnectionHandlerEx); isHandler { if outboundHandler, isHandler := selected.(adapter.PacketConnectionHandler); isHandler {
outboundHandler.NewPacketConnectionEx(ctx, conn, metadata, onClose) outboundHandler.NewPacketConnection(ctx, conn, metadata, onClose)
} else { } else {
s.connection.NewPacketConnection(ctx, selected, conn, metadata, onClose) s.connection.NewPacketConnection(ctx, selected, conn, metadata, onClose)
} }
+2 -2
View File
@@ -163,12 +163,12 @@ func (s *URLTest) ListenPacket(ctx context.Context, destination M.Socksaddr) (ne
return nil, err return nil, err
} }
func (s *URLTest) NewConnectionEx(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) { func (s *URLTest) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
ctx = interrupt.ContextWithIsExternalConnection(ctx) ctx = interrupt.ContextWithIsExternalConnection(ctx)
s.connection.NewConnection(ctx, s, conn, metadata, onClose) s.connection.NewConnection(ctx, s, conn, metadata, onClose)
} }
func (s *URLTest) NewPacketConnectionEx(ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) { func (s *URLTest) NewPacketConnection(ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
ctx = interrupt.ContextWithIsExternalConnection(ctx) ctx = interrupt.ContextWithIsExternalConnection(ctx)
s.connection.NewPacketConnection(ctx, s, conn, metadata, onClose) s.connection.NewPacketConnection(ctx, s, conn, metadata, onClose)
} }
+2 -2
View File
@@ -86,7 +86,7 @@ func (h *Inbound) Close() error {
) )
} }
func (h *Inbound) NewConnectionEx(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) { func (h *Inbound) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
if h.tlsConfig != nil { if h.tlsConfig != nil {
tlsConn, err := tls.ServerHandshake(ctx, conn, h.tlsConfig) tlsConn, err := tls.ServerHandshake(ctx, conn, h.tlsConfig)
if err != nil { if err != nil {
@@ -96,7 +96,7 @@ func (h *Inbound) NewConnectionEx(ctx context.Context, conn net.Conn, metadata a
} }
conn = tlsConn conn = tlsConn
} }
err := http.HandleConnectionEx(ctx, conn, std_bufio.NewReader(conn), h.authenticator, adapter.NewUpstreamHandlerEx(metadata, h.newUserConnection, h.streamUserPacketConnection), metadata.Source, onClose) err := http.HandleConnectionEx(ctx, conn, std_bufio.NewReader(conn), h.authenticator, adapter.NewUpstreamHandler(metadata, h.newUserConnection, h.streamUserPacketConnection), metadata.Source, onClose)
if err != nil { if err != nil {
N.CloseOnHandshakeFailure(conn, onClose, err) N.CloseOnHandshakeFailure(conn, onClose, err)
h.logger.ErrorContext(ctx, E.Cause(err, "process connection from ", metadata.Source)) h.logger.ErrorContext(ctx, E.Cause(err, "process connection from ", metadata.Source))
+3 -3
View File
@@ -98,7 +98,7 @@ func (h *Inbound) Close() error {
) )
} }
func (h *Inbound) NewConnectionEx(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) { func (h *Inbound) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
err := h.newConnection(ctx, conn, metadata, onClose) err := h.newConnection(ctx, conn, metadata, onClose)
N.CloseOnHandshakeFailure(conn, onClose, err) N.CloseOnHandshakeFailure(conn, onClose, err)
if err != nil { if err != nil {
@@ -125,9 +125,9 @@ func (h *Inbound) newConnection(ctx context.Context, conn net.Conn, metadata ada
} }
switch headerBytes[0] { switch headerBytes[0] {
case socks4.Version, socks5.Version: case socks4.Version, socks5.Version:
return socks.HandleConnectionEx(ctx, conn, reader, h.authenticator, adapter.NewUpstreamHandlerEx(metadata, h.newUserConnection, h.streamUserPacketConnection), h.listener, h.udpTimeout, metadata.Source, onClose) return socks.HandleConnectionEx(ctx, conn, reader, h.authenticator, adapter.NewUpstreamHandler(metadata, h.newUserConnection, h.streamUserPacketConnection), h.listener, h.udpTimeout, metadata.Source, onClose)
default: default:
return http.HandleConnectionEx(ctx, conn, reader, h.authenticator, adapter.NewUpstreamHandlerEx(metadata, h.newUserConnection, h.streamUserPacketConnection), metadata.Source, onClose) return http.HandleConnectionEx(ctx, conn, reader, h.authenticator, adapter.NewUpstreamHandler(metadata, h.newUserConnection, h.streamUserPacketConnection), metadata.Source, onClose)
} }
} }
+1 -1
View File
@@ -53,7 +53,7 @@ func (h *Redirect) Close() error {
return h.listener.Close() return h.listener.Close()
} }
func (h *Redirect) NewConnectionEx(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) { func (h *Redirect) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
destination, err := redir.GetOriginalDestination(conn) destination, err := redir.GetOriginalDestination(conn)
if err != nil { if err != nil {
conn.Close() conn.Close()
+2 -2
View File
@@ -71,7 +71,7 @@ func (t *TProxy) Close() error {
return t.listener.Close() return t.listener.Close()
} }
func (t *TProxy) NewConnectionEx(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) { func (t *TProxy) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
metadata.Inbound = t.Tag() metadata.Inbound = t.Tag()
metadata.InboundType = t.Type() metadata.InboundType = t.Type()
metadata.Destination = M.SocksaddrFromNet(conn.LocalAddr()).Unwrap() metadata.Destination = M.SocksaddrFromNet(conn.LocalAddr()).Unwrap()
@@ -91,7 +91,7 @@ func (t *TProxy) NewPacketConnectionEx(ctx context.Context, conn N.PacketConn, s
t.router.RoutePacketConnectionEx(ctx, conn, metadata, onClose) t.router.RoutePacketConnectionEx(ctx, conn, metadata, onClose)
} }
func (t *TProxy) NewPacketEx(buffer *buf.Buffer, oob []byte, source M.Socksaddr) { func (t *TProxy) NewPacket(buffer *buf.Buffer, oob []byte, source M.Socksaddr) {
destination, err := redir.GetOriginalDestinationFromOOB(oob) destination, err := redir.GetOriginalDestinationFromOOB(oob)
if err != nil { if err != nil {
t.logger.Warn("process packet from ", source, ": get tproxy destination: ", err) t.logger.Warn("process packet from ", source, ": get tproxy destination: ", err)
+5 -5
View File
@@ -75,11 +75,11 @@ func newInbound(ctx context.Context, router adapter.Router, logger log.ContextLo
} }
switch { switch {
case options.Method == shadowsocks.MethodNone: case options.Method == shadowsocks.MethodNone:
inbound.service = shadowsocks.NewNoneService(int64(udpTimeout.Seconds()), adapter.NewUpstreamHandler(adapter.InboundContext{}, inbound.newConnection, inbound.newPacketConnection, inbound)) inbound.service = shadowsocks.NewNoneService(int64(udpTimeout.Seconds()), adapter.NewLegacyUpstreamHandler(adapter.InboundContext{}, inbound.newConnection, inbound.newPacketConnection, inbound))
case common.Contains(shadowaead.List, options.Method): case common.Contains(shadowaead.List, options.Method):
inbound.service, err = shadowaead.NewService(options.Method, nil, options.Password, int64(udpTimeout.Seconds()), adapter.NewUpstreamHandler(adapter.InboundContext{}, inbound.newConnection, inbound.newPacketConnection, inbound)) inbound.service, err = shadowaead.NewService(options.Method, nil, options.Password, int64(udpTimeout.Seconds()), adapter.NewLegacyUpstreamHandler(adapter.InboundContext{}, inbound.newConnection, inbound.newPacketConnection, inbound))
case common.Contains(shadowaead_2022.List, options.Method): case common.Contains(shadowaead_2022.List, options.Method):
inbound.service, err = shadowaead_2022.NewServiceWithPassword(options.Method, options.Password, int64(udpTimeout.Seconds()), adapter.NewUpstreamHandler(adapter.InboundContext{}, inbound.newConnection, inbound.newPacketConnection, inbound), ntp.TimeFuncFromContext(ctx)) inbound.service, err = shadowaead_2022.NewServiceWithPassword(options.Method, options.Password, int64(udpTimeout.Seconds()), adapter.NewLegacyUpstreamHandler(adapter.InboundContext{}, inbound.newConnection, inbound.newPacketConnection, inbound), ntp.TimeFuncFromContext(ctx))
default: default:
err = E.New("unsupported method: ", options.Method) err = E.New("unsupported method: ", options.Method)
} }
@@ -107,7 +107,7 @@ func (h *Inbound) Close() error {
} }
//nolint:staticcheck //nolint:staticcheck
func (h *Inbound) NewConnectionEx(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) { func (h *Inbound) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
err := h.service.NewConnection(ctx, conn, adapter.UpstreamMetadata(metadata)) err := h.service.NewConnection(ctx, conn, adapter.UpstreamMetadata(metadata))
N.CloseOnHandshakeFailure(conn, onClose, err) N.CloseOnHandshakeFailure(conn, onClose, err)
if err != nil { if err != nil {
@@ -120,7 +120,7 @@ func (h *Inbound) NewConnectionEx(ctx context.Context, conn net.Conn, metadata a
} }
//nolint:staticcheck //nolint:staticcheck
func (h *Inbound) NewPacketEx(buffer *buf.Buffer, source M.Socksaddr) { func (h *Inbound) NewPacket(buffer *buf.Buffer, source M.Socksaddr) {
err := h.service.NewPacket(h.ctx, &stubPacketConn{h.listener.PacketWriter()}, buffer, M.Metadata{Source: source}) err := h.service.NewPacket(h.ctx, &stubPacketConn{h.listener.PacketWriter()}, buffer, M.Metadata{Source: source})
if err != nil { if err != nil {
h.logger.Error(E.Cause(err, "process packet from ", source)) h.logger.Error(E.Cause(err, "process packet from ", source))
+4 -4
View File
@@ -68,14 +68,14 @@ func newMultiInbound(ctx context.Context, router adapter.Router, logger log.Cont
options.Method, options.Method,
options.Password, options.Password,
int64(udpTimeout.Seconds()), int64(udpTimeout.Seconds()),
adapter.NewUpstreamHandler(adapter.InboundContext{}, inbound.newConnection, inbound.newPacketConnection, inbound), adapter.NewLegacyUpstreamHandler(adapter.InboundContext{}, inbound.newConnection, inbound.newPacketConnection, inbound),
ntp.TimeFuncFromContext(ctx), ntp.TimeFuncFromContext(ctx),
) )
} else if common.Contains(shadowaead.List, options.Method) { } else if common.Contains(shadowaead.List, options.Method) {
service, err = shadowaead.NewMultiService[int]( service, err = shadowaead.NewMultiService[int](
options.Method, options.Method,
int64(udpTimeout.Seconds()), int64(udpTimeout.Seconds()),
adapter.NewUpstreamHandler(adapter.InboundContext{}, inbound.newConnection, inbound.newPacketConnection, inbound), adapter.NewLegacyUpstreamHandler(adapter.InboundContext{}, inbound.newConnection, inbound.newPacketConnection, inbound),
) )
} else { } else {
return nil, E.New("unsupported method: " + options.Method) return nil, E.New("unsupported method: " + options.Method)
@@ -138,7 +138,7 @@ func (h *MultiInbound) UpdateUsers(users []string, uPSKs []string) error {
} }
//nolint:staticcheck //nolint:staticcheck
func (h *MultiInbound) NewConnectionEx(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) { func (h *MultiInbound) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
err := h.service.NewConnection(ctx, conn, adapter.UpstreamMetadata(metadata)) err := h.service.NewConnection(ctx, conn, adapter.UpstreamMetadata(metadata))
N.CloseOnHandshakeFailure(conn, onClose, err) N.CloseOnHandshakeFailure(conn, onClose, err)
if err != nil { if err != nil {
@@ -151,7 +151,7 @@ func (h *MultiInbound) NewConnectionEx(ctx context.Context, conn net.Conn, metad
} }
//nolint:staticcheck //nolint:staticcheck
func (h *MultiInbound) NewPacketEx(buffer *buf.Buffer, source M.Socksaddr) { func (h *MultiInbound) NewPacket(buffer *buf.Buffer, source M.Socksaddr) {
err := h.service.NewPacket(h.ctx, &stubPacketConn{h.listener.PacketWriter()}, buffer, M.Metadata{Source: source}) err := h.service.NewPacket(h.ctx, &stubPacketConn{h.listener.PacketWriter()}, buffer, M.Metadata{Source: source})
if err != nil { if err != nil {
h.logger.Error(E.Cause(err, "process packet from ", source)) h.logger.Error(E.Cause(err, "process packet from ", source))
+3 -3
View File
@@ -60,7 +60,7 @@ func newRelayInbound(ctx context.Context, router adapter.Router, logger log.Cont
options.Method, options.Method,
options.Password, options.Password,
int64(udpTimeout.Seconds()), int64(udpTimeout.Seconds()),
adapter.NewUpstreamHandler(adapter.InboundContext{}, inbound.newConnection, inbound.newPacketConnection, inbound), adapter.NewLegacyUpstreamHandler(adapter.InboundContext{}, inbound.newConnection, inbound.newPacketConnection, inbound),
) )
if err != nil { if err != nil {
return nil, err return nil, err
@@ -98,7 +98,7 @@ func (h *RelayInbound) Close() error {
} }
//nolint:staticcheck //nolint:staticcheck
func (h *RelayInbound) NewConnectionEx(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) { func (h *RelayInbound) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
err := h.service.NewConnection(ctx, conn, adapter.UpstreamMetadata(metadata)) err := h.service.NewConnection(ctx, conn, adapter.UpstreamMetadata(metadata))
N.CloseOnHandshakeFailure(conn, onClose, err) N.CloseOnHandshakeFailure(conn, onClose, err)
if err != nil { if err != nil {
@@ -111,7 +111,7 @@ func (h *RelayInbound) NewConnectionEx(ctx context.Context, conn net.Conn, metad
} }
//nolint:staticcheck //nolint:staticcheck
func (h *RelayInbound) NewPacketEx(buffer *buf.Buffer, source M.Socksaddr) { func (h *RelayInbound) NewPacket(buffer *buf.Buffer, source M.Socksaddr) {
err := h.service.NewPacket(h.ctx, &stubPacketConn{h.listener.PacketWriter()}, buffer, M.Metadata{Source: source}) err := h.service.NewPacket(h.ctx, &stubPacketConn{h.listener.PacketWriter()}, buffer, M.Metadata{Source: source})
if err != nil { if err != nil {
h.logger.Error(E.Cause(err, "process packet from ", source)) h.logger.Error(E.Cause(err, "process packet from ", source))
+1 -1
View File
@@ -108,7 +108,7 @@ func (h *Inbound) Close() error {
return h.listener.Close() return h.listener.Close()
} }
func (h *Inbound) NewConnectionEx(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) { func (h *Inbound) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
err := h.service.NewConnection(adapter.WithContext(log.ContextWithNewID(ctx), &metadata), conn, metadata.Source, metadata.Destination, onClose) err := h.service.NewConnection(adapter.WithContext(log.ContextWithNewID(ctx), &metadata), conn, metadata.Source, metadata.Destination, onClose)
N.CloseOnHandshakeFailure(conn, onClose, err) N.CloseOnHandshakeFailure(conn, onClose, err)
if err != nil { if err != nil {
+2 -2
View File
@@ -70,8 +70,8 @@ func (h *Inbound) Close() error {
return h.listener.Close() return h.listener.Close()
} }
func (h *Inbound) NewConnectionEx(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) { func (h *Inbound) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
err := socks.HandleConnectionEx(ctx, conn, std_bufio.NewReader(conn), h.authenticator, adapter.NewUpstreamHandlerEx(metadata, h.newUserConnection, h.streamUserPacketConnection), h.listener, h.udpTimeout, metadata.Source, onClose) err := socks.HandleConnectionEx(ctx, conn, std_bufio.NewReader(conn), h.authenticator, adapter.NewUpstreamHandler(metadata, h.newUserConnection, h.streamUserPacketConnection), h.listener, h.udpTimeout, metadata.Source, onClose)
N.CloseOnHandshakeFailure(conn, onClose, err) N.CloseOnHandshakeFailure(conn, onClose, err)
if err != nil { if err != nil {
if E.IsClosedOrCanceled(err) { if E.IsClosedOrCanceled(err) {
+4 -4
View File
@@ -84,9 +84,9 @@ func NewInbound(ctx context.Context, router adapter.Router, logger log.ContextLo
} }
inbound.fallbackAddrTLSNextProto = fallbackAddrNextProto inbound.fallbackAddrTLSNextProto = fallbackAddrNextProto
} }
fallbackHandler = adapter.NewUpstreamContextHandlerEx(inbound.fallbackConnection, nil) fallbackHandler = adapter.NewUpstreamContextHandler(inbound.fallbackConnection, nil)
} }
service := trojan.NewService[int](adapter.NewUpstreamContextHandlerEx(inbound.newConnection, inbound.newPacketConnection), fallbackHandler, logger) service := trojan.NewService[int](adapter.NewUpstreamContextHandler(inbound.newConnection, inbound.newPacketConnection), fallbackHandler, logger)
err := service.UpdateUsers(common.MapIndexed(options.Users, func(index int, it option.TrojanUser) int { err := service.UpdateUsers(common.MapIndexed(options.Users, func(index int, it option.TrojanUser) int {
return index return index
}), common.Map(options.Users, func(it option.TrojanUser) string { }), common.Map(options.Users, func(it option.TrojanUser) string {
@@ -164,7 +164,7 @@ func (h *Inbound) Close() error {
) )
} }
func (h *Inbound) NewConnectionEx(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) { func (h *Inbound) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
if h.tlsConfig != nil && h.transport == nil { if h.tlsConfig != nil && h.transport == nil {
tlsConn, err := tls.ServerHandshake(ctx, conn, h.tlsConfig) tlsConn, err := tls.ServerHandshake(ctx, conn, h.tlsConfig)
if err != nil { if err != nil {
@@ -258,5 +258,5 @@ func (h *inboundTransportHandler) NewConnectionEx(ctx context.Context, conn net.
metadata.InboundDetour = h.listener.ListenOptions().Detour metadata.InboundDetour = h.listener.ListenOptions().Detour
//nolint:staticcheck //nolint:staticcheck
h.logger.InfoContext(ctx, "inbound connection from ", metadata.Source) h.logger.InfoContext(ctx, "inbound connection from ", metadata.Source)
(*Inbound)(h).NewConnectionEx(ctx, conn, metadata, onClose) (*Inbound)(h).NewConnection(ctx, conn, metadata, onClose)
} }
+3 -3
View File
@@ -58,7 +58,7 @@ func NewInbound(ctx context.Context, router adapter.Router, logger log.ContextLo
if err != nil { if err != nil {
return nil, err return nil, err
} }
service := vless.NewService[int](logger, adapter.NewUpstreamContextHandlerEx(inbound.newConnectionEx, inbound.newPacketConnectionEx)) service := vless.NewService[int](logger, adapter.NewUpstreamContextHandler(inbound.newConnectionEx, inbound.newPacketConnectionEx))
service.UpdateUsers(common.MapIndexed(inbound.users, func(index int, _ option.VLESSUser) int { service.UpdateUsers(common.MapIndexed(inbound.users, func(index int, _ option.VLESSUser) int {
return index return index
}), common.Map(inbound.users, func(it option.VLESSUser) string { }), common.Map(inbound.users, func(it option.VLESSUser) string {
@@ -147,7 +147,7 @@ func (h *Inbound) Close() error {
) )
} }
func (h *Inbound) NewConnectionEx(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) { func (h *Inbound) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
if h.tlsConfig != nil && h.transport == nil { if h.tlsConfig != nil && h.transport == nil {
tlsConn, err := tls.ServerHandshake(ctx, conn, h.tlsConfig) tlsConn, err := tls.ServerHandshake(ctx, conn, h.tlsConfig)
if err != nil { if err != nil {
@@ -218,5 +218,5 @@ func (h *inboundTransportHandler) NewConnectionEx(ctx context.Context, conn net.
metadata.InboundDetour = h.listener.ListenOptions().Detour metadata.InboundDetour = h.listener.ListenOptions().Detour
//nolint:staticcheck //nolint:staticcheck
h.logger.InfoContext(ctx, "inbound connection from ", metadata.Source) h.logger.InfoContext(ctx, "inbound connection from ", metadata.Source)
(*Inbound)(h).NewConnectionEx(ctx, conn, metadata, onClose) (*Inbound)(h).NewConnection(ctx, conn, metadata, onClose)
} }
+3 -3
View File
@@ -66,7 +66,7 @@ func NewInbound(ctx context.Context, router adapter.Router, logger log.ContextLo
if options.Transport != nil && options.Transport.Type != "" { if options.Transport != nil && options.Transport.Type != "" {
serviceOptions = append(serviceOptions, vmess.ServiceWithDisableHeaderProtection()) serviceOptions = append(serviceOptions, vmess.ServiceWithDisableHeaderProtection())
} }
service := vmess.NewService[int](adapter.NewUpstreamContextHandlerEx(inbound.newConnectionEx, inbound.newPacketConnectionEx), serviceOptions...) service := vmess.NewService[int](adapter.NewUpstreamContextHandler(inbound.newConnectionEx, inbound.newPacketConnectionEx), serviceOptions...)
inbound.service = service inbound.service = service
err = service.UpdateUsers(common.MapIndexed(options.Users, func(index int, it option.VMessUser) int { err = service.UpdateUsers(common.MapIndexed(options.Users, func(index int, it option.VMessUser) int {
return index return index
@@ -153,7 +153,7 @@ func (h *Inbound) Close() error {
) )
} }
func (h *Inbound) NewConnectionEx(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) { func (h *Inbound) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
if h.tlsConfig != nil && h.transport == nil { if h.tlsConfig != nil && h.transport == nil {
tlsConn, err := tls.ServerHandshake(ctx, conn, h.tlsConfig) tlsConn, err := tls.ServerHandshake(ctx, conn, h.tlsConfig)
if err != nil { if err != nil {
@@ -224,5 +224,5 @@ func (h *inboundTransportHandler) NewConnectionEx(ctx context.Context, conn net.
metadata.InboundDetour = h.listener.ListenOptions().Detour metadata.InboundDetour = h.listener.ListenOptions().Detour
//nolint:staticcheck //nolint:staticcheck
h.logger.InfoContext(ctx, "inbound connection from ", metadata.Source) h.logger.InfoContext(ctx, "inbound connection from ", metadata.Source)
(*Inbound)(h).NewConnectionEx(ctx, conn, metadata, onClose) (*Inbound)(h).NewConnection(ctx, conn, metadata, onClose)
} }
+6 -6
View File
@@ -74,7 +74,7 @@ func (r *Router) routeConnection(ctx context.Context, conn net.Conn, metadata ad
metadata.LastInbound = metadata.Inbound metadata.LastInbound = metadata.Inbound
metadata.Inbound = metadata.InboundDetour metadata.Inbound = metadata.InboundDetour
metadata.InboundDetour = "" metadata.InboundDetour = ""
injectable.NewConnectionEx(ctx, conn, metadata, onClose) injectable.NewConnection(ctx, conn, metadata, onClose)
return nil return nil
} }
metadata.Network = N.NetworkTCP metadata.Network = N.NetworkTCP
@@ -152,8 +152,8 @@ func (r *Router) routeConnection(ctx context.Context, conn net.Conn, metadata ad
for _, tracker := range r.trackers { for _, tracker := range r.trackers {
conn = tracker.RoutedConnection(ctx, conn, metadata, selectedRule, selectedOutbound) conn = tracker.RoutedConnection(ctx, conn, metadata, selectedRule, selectedOutbound)
} }
if outboundHandler, isHandler := selectedOutbound.(adapter.ConnectionHandlerEx); isHandler { if outboundHandler, isHandler := selectedOutbound.(adapter.ConnectionHandler); isHandler {
outboundHandler.NewConnectionEx(ctx, conn, metadata, onClose) outboundHandler.NewConnection(ctx, conn, metadata, onClose)
} else { } else {
r.connection.NewConnection(ctx, selectedOutbound, conn, metadata, onClose) r.connection.NewConnection(ctx, selectedOutbound, conn, metadata, onClose)
} }
@@ -209,7 +209,7 @@ func (r *Router) routePacketConnection(ctx context.Context, conn N.PacketConn, m
metadata.LastInbound = metadata.Inbound metadata.LastInbound = metadata.Inbound
metadata.Inbound = metadata.InboundDetour metadata.Inbound = metadata.InboundDetour
metadata.InboundDetour = "" metadata.InboundDetour = ""
injectable.NewPacketConnectionEx(ctx, conn, metadata, onClose) injectable.NewPacketConnection(ctx, conn, metadata, onClose)
return nil return nil
} }
// TODO: move to UoT // TODO: move to UoT
@@ -281,8 +281,8 @@ func (r *Router) routePacketConnection(ctx context.Context, conn N.PacketConn, m
if metadata.FakeIP { if metadata.FakeIP {
conn = bufio.NewNATPacketConn(bufio.NewNetPacketConn(conn), metadata.OriginDestination, metadata.Destination) conn = bufio.NewNATPacketConn(bufio.NewNetPacketConn(conn), metadata.OriginDestination, metadata.Destination)
} }
if outboundHandler, isHandler := selectedOutbound.(adapter.PacketConnectionHandlerEx); isHandler { if outboundHandler, isHandler := selectedOutbound.(adapter.PacketConnectionHandler); isHandler {
outboundHandler.NewPacketConnectionEx(ctx, conn, metadata, onClose) outboundHandler.NewPacketConnection(ctx, conn, metadata, onClose)
} else { } else {
r.connection.NewPacketConnection(ctx, selectedOutbound, conn, metadata, onClose) r.connection.NewPacketConnection(ctx, selectedOutbound, conn, metadata, onClose)
} }
+2 -2
View File
@@ -132,7 +132,7 @@ func (i *Service) Close() error {
return i.listener.Close() return i.listener.Close()
} }
func (i *Service) NewConnectionEx(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) { func (i *Service) NewConnection(ctx context.Context, conn net.Conn, metadata adapter.InboundContext, onClose N.CloseHandlerFunc) {
metadata.Inbound = i.Tag() metadata.Inbound = i.Tag()
metadata.InboundType = i.Type() metadata.InboundType = i.Type()
metadata.Destination = M.Socksaddr{} metadata.Destination = M.Socksaddr{}
@@ -146,7 +146,7 @@ func (i *Service) NewConnectionEx(ctx context.Context, conn net.Conn, metadata a
} }
} }
func (i *Service) NewPacketEx(buffer *buf.Buffer, oob []byte, source M.Socksaddr) { func (i *Service) NewPacket(buffer *buf.Buffer, oob []byte, source M.Socksaddr) {
go i.exchangePacket(buffer, oob, source) go i.exchangePacket(buffer, oob, source)
} }