mirrors/Xray-core
1
0
Fork 0
mirror of https://github.com/XTLS/Xray-core.git synced 2024-11-26 17:03:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add callClose to UDP Dispatcher

Browse Source 
Fixes https://github.com/XTLS/Xray-core/issues/1611
This commit is contained in:
RPRX 2023-02-08 14:59:14 +08:00 committed by GitHub
parent f32921df30
commit 9046eda5ce
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 28 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
transport/internet/udp/dispatcher.go
View File

@ -31,6 +31,7 @@ type Dispatcher struct {
conns map[net.Destination]*connEntry conns map[net.Destination]*connEntry
dispatcher routing.Dispatcher dispatcher routing.Dispatcher
callback ResponseCallback callback ResponseCallback
callClose func() error
} }
func NewDispatcher(dispatcher routing.Dispatcher, callback ResponseCallback) *Dispatcher { func NewDispatcher(dispatcher routing.Dispatcher, callback ResponseCallback) *Dispatcher {
@ -79,7 +80,7 @@ func (v *Dispatcher) getInboundRay(ctx context.Context, dest net.Destination) (*
cancel: removeRay, cancel: removeRay,
} }
v.conns[dest] = entry v.conns[dest] = entry
go handleInput(ctx, entry, dest, v.callback) go handleInput(ctx, entry, dest, v.callback, v.callClose)
return entry, nil return entry, nil
} }
@ -102,8 +103,13 @@ func (v *Dispatcher) Dispatch(ctx context.Context, destination net.Destination,
} }
} }
func handleInput(ctx context.Context, conn *connEntry, dest net.Destination, callback ResponseCallback) { func handleInput(ctx context.Context, conn *connEntry, dest net.Destination, callback ResponseCallback, callClose func() error) {
defer conn.cancel() defer func() {
conn.cancel()
if callClose != nil {
callClose()
}
}()
input := conn.link.Reader input := conn.link.Reader
timer := conn.timer timer := conn.timer
@ -144,7 +150,12 @@ func DialDispatcher(ctx context.Context, dispatcher routing.Dispatcher) (net.Pac
done: done.New(), done: done.New(),
} }
d := NewDispatcher(dispatcher, c.callback) d := &Dispatcher{
conns: make(map[net.Destination]*connEntry),
dispatcher: dispatcher,
callback: c.callback,
callClose: c.Close,
}
c.dispatcher = d c.dispatcher = d
return c, nil return c, nil
} }
@ -162,17 +173,23 @@ func (c *dispatcherConn) callback(ctx context.Context, packet *udp.Packet) {
} }
func (c *dispatcherConn) ReadFrom(p []byte) (int, net.Addr, error) { func (c *dispatcherConn) ReadFrom(p []byte) (int, net.Addr, error) {
var packet *udp.Packet
s:
select { select {
case <-c.done.Wait(): case <-c.done.Wait():
select {
case packet = <-c.cache:
break s
default:
return 0, nil, io.EOF return 0, nil, io.EOF
case packet := <-c.cache: }
n := copy(p, packet.Payload.Bytes()) case packet = <-c.cache:
return n, &net.UDPAddr{ }
return copy(p, packet.Payload.Bytes()), &net.UDPAddr{
IP: packet.Source.Address.IP(), IP: packet.Source.Address.IP(),
Port: int(packet.Source.Port), Port: int(packet.Source.Port),
}, nil }, nil
} }
}
func (c *dispatcherConn) WriteTo(p []byte, addr net.Addr) (int, error) { func (c *dispatcherConn) WriteTo(p []byte, addr net.Addr) (int, error) {
buffer := buf.New() buffer := buf.New()