hesoyam/Xray-core
1
0
Fork 0
mirror of https://github.com/XTLS/Xray-core.git synced 2026-11-03 17:06:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

SS2022: Correct interrupt behavior and UDP conn change

Browse Source
This commit is contained in:
Fangliding
2026-10-14 16:24:11 +08:00
parent 701af60772
commit 79551872bd
3 changed files with 18 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
proxy/shadowsocks_2022/outbound.go
+4 -4
View File
@@ -169,8 +169,8 @@ func (o *Outbound) Process(ctx context.Context, link *transport.Link, dialer int
return buf.Copy(responseReader, link.Writer, buf.UpdateActivity(timer)) return buf.Copy(responseReader, link.Writer, buf.UpdateActivity(timer))
} }
responseDoneAndCloseWriter := task.OnSuccess(responseDone, task.Close(link.Writer)) responseDonePost := task.OnSuccess(responseDone, task.Close(link.Writer))
if err := task.Run(ctx, requestDone, responseDoneAndCloseWriter); err != nil { if err := task.Run(ctx, requestDone, responseDonePost); err != nil {
return errors.New("connection ends").Base(err) return errors.New("connection ends").Base(err)
} }
@@ -212,8 +212,8 @@ func (o *Outbound) Process(ctx context.Context, link *transport.Link, dialer int
return nil return nil
} }
responseDoneAndCloseWriter := task.OnSuccess(responseDone, task.Close(link.Writer)) responseDonePost := task.OnSuccess(responseDone, task.Close(link.Writer))
if err := task.Run(ctx, requestDone, responseDoneAndCloseWriter); err != nil { if err := task.Run(ctx, requestDone, responseDonePost); err != nil {
return errors.New("connection ends").Base(err) return errors.New("connection ends").Base(err)
} }
proxy/shadowsocks_2022/shadowsocks_2022.go
+4 -4
View File
@@ -18,9 +18,7 @@ import (
) )
func (s *ServerUDPSession) UpdateConn(conn stat.Connection) { func (s *ServerUDPSession) UpdateConn(conn stat.Connection) {
if s.currentConn.Load() == nil { s.currentConn.Store(conn)
s.currentConn.Store(conn)
}
if s.timer != nil { if s.timer != nil {
s.timer.Update() s.timer.Update()
} }
@@ -70,6 +68,8 @@ func (s *ServerUDPSession) EnsureLink(
return link, nil return link, nil
} }
s.currentConn.Store(conn)
sessCtx, cancel := context.WithCancel(ctx) sessCtx, cancel := context.WithCancel(ctx)
inbound := session.InboundFromContext(sessCtx) inbound := session.InboundFromContext(sessCtx)
if inbound != nil && s.User != nil { if inbound != nil && s.User != nil {
@@ -94,7 +94,6 @@ func (s *ServerUDPSession) EnsureLink(
return nil, err return nil, err
} }
s.link.Store(link)
sessionPolicy := policyManager.ForLevel(level) sessionPolicy := policyManager.ForLevel(level)
s.timer = signal.CancelAfterInactivity(sessCtx, func() { s.timer = signal.CancelAfterInactivity(sessCtx, func() {
if s.manager != nil { if s.manager != nil {
@@ -103,6 +102,7 @@ func (s *ServerUDPSession) EnsureLink(
s.Close() s.Close()
cancel() cancel()
}, sessionPolicy.Timeouts.ConnectionIdle) }, sessionPolicy.Timeouts.ConnectionIdle)
s.link.Store(link)
go handleUDPResponse(s, link, dest, responseEncoder) go handleUDPResponse(s, link, dest, responseEncoder)
return link, nil return link, nil
proxy/shadowsocks_2022/stream.go
+10 -2
View File
@@ -11,6 +11,7 @@ import (
"sync" "sync"
"time" "time"
"github.com/xtls/xray-core/common"
"github.com/xtls/xray-core/common/antireplay" "github.com/xtls/xray-core/common/antireplay"
"github.com/xtls/xray-core/common/buf" "github.com/xtls/xray-core/common/buf"
"github.com/xtls/xray-core/common/errors" "github.com/xtls/xray-core/common/errors"
@@ -629,6 +630,7 @@ func InitServerStream(conn net.Conn, method *CipherMethod, psk, saltSlice []byte
func TransportTCP(ctx context.Context, sessionPolicy policy.Session, reader buf.Reader, writer buf.Writer, link *transport.Link) error { func TransportTCP(ctx context.Context, sessionPolicy policy.Session, reader buf.Reader, writer buf.Writer, link *transport.Link) error {
ctx, cancel := context.WithCancel(ctx) ctx, cancel := context.WithCancel(ctx)
timer := signal.CancelAfterInactivity(ctx, cancel, sessionPolicy.Timeouts.ConnectionIdle) timer := signal.CancelAfterInactivity(ctx, cancel, sessionPolicy.Timeouts.ConnectionIdle)
defer timer.SetTimeout(0)
ctx = policy.ContextWithBufferPolicy(ctx, sessionPolicy.Buffer) ctx = policy.ContextWithBufferPolicy(ctx, sessionPolicy.Buffer)
requestDone := func() error { requestDone := func() error {
@@ -644,6 +646,12 @@ func TransportTCP(ctx context.Context, sessionPolicy policy.Session, reader buf.
return buf.Copy(link.Reader, writer, buf.UpdateActivity(timer)) return buf.Copy(link.Reader, writer, buf.UpdateActivity(timer))
} }
responseDoneAndCloseWriter := task.OnSuccess(responseDone, task.Close(link.Writer)) requestDoneAndCloseWriter := task.OnSuccess(requestDone, task.Close(link.Writer))
return task.Run(ctx, requestDone, responseDoneAndCloseWriter) if err := task.Run(ctx, requestDoneAndCloseWriter, responseDone); err != nil {
common.Interrupt(link.Reader)
common.Interrupt(link.Writer)
return errors.New("connection ends").Base(err)
}
return nil
} }