fix: 流量用户为null，xray无法断开失效用户连接

core/xray/app/dispatcher/default.go

@@ -102,13 +102,13 @@ func (r *cachedReader) Interrupt() {
 
 // DefaultDispatcher is a default implementation of Dispatcher.
 type DefaultDispatcher struct {
-	ohm     outbound.Manager
-	router  routing.Router
-	policy  policy.Manager
-	stats   stats.Manager
-	fdns    dns.FakeDNSEngine
-	Wm      *WriterManager
-	Counter sync.Map
+	ohm          outbound.Manager
+	router       routing.Router
+	policy       policy.Manager
+	stats        stats.Manager
+	fdns         dns.FakeDNSEngine
+	Counter      sync.Map
+	LinkManagers sync.Map // map[string]*LinkManager
 }
 
 func init() {
@@ -132,9 +132,6 @@ func (d *DefaultDispatcher) Init(config *Config, om outbound.Manager, router rou
 	d.router = router
 	d.policy = pm
 	d.stats = sm
-	d.Wm = &WriterManager{
-		writers: make(map[string]map[*ManagedWriter]struct{}),
-	}
 	return nil
 }
 
@@ -197,13 +194,27 @@ func (d *DefaultDispatcher) getLink(ctx context.Context, network net.Network) (*
 			common.Interrupt(inboundLink.Reader)
 			return nil, nil, nil, errors.New("Limited ", user.Email, " by conn or ip")
 		}
+		var lm *LinkManager
+		if lmloaded, ok := d.LinkManagers.Load(user.Email); !ok {
+			lm = &LinkManager{
+				writers: make(map[*ManagedWriter]struct{}),
+				readers: make(map[*ManagedReader]struct{}),
+			}
+			d.LinkManagers.Store(user.Email, lm)
+		} else {
+			lm = lmloaded.(*LinkManager)
+		}
 		managedWriter := &ManagedWriter{
 			writer:  uplinkWriter,
-			email:   user.Email,
-			manager: d.Wm,
+			manager: lm,
 		}
-		d.Wm.AddWriter(managedWriter)
+		managedReader := &ManagedReader{
+			reader:  downlinkReader,
+			manager: lm,
+		}
+		lm.AddLink(managedWriter, managedReader)
 		inboundLink.Writer = managedWriter
+		outboundLink.Reader = managedReader
 		if w != nil {
 			inboundLink.Writer = rate.NewRateLimitWriter(inboundLink.Writer, w)
 			outboundLink.Writer = rate.NewRateLimitWriter(outboundLink.Writer, w)
@@ -305,7 +316,7 @@ func (d *DefaultDispatcher) Dispatch(ctx context.Context, destination net.Destin
 	} else {
 		go func() {
 			cReader := &cachedReader{
-				reader: outbound.Reader.(*pipe.Reader),
+				reader: outbound.Reader.(*ManagedReader),
 			}
 			outbound.Reader = cReader
 			result, err := sniffer(ctx, cReader, sniffingRequest.MetadataOnly, destination.Network)
@@ -382,13 +393,27 @@ func (d *DefaultDispatcher) DispatchLink(ctx context.Context, destination net.De
 			common.Interrupt(outbound.Reader)
 			return errors.New("Limited ", user.Email, " by conn or ip")
 		}
+		var lm *LinkManager
+		if lmloaded, ok := d.LinkManagers.Load(user.Email); !ok {
+			lm = &LinkManager{
+				writers: make(map[*ManagedWriter]struct{}),
+				readers: make(map[*ManagedReader]struct{}),
+			}
+			d.LinkManagers.Store(user.Email, lm)
+		} else {
+			lm = lmloaded.(*LinkManager)
+		}
 		managedWriter := &ManagedWriter{
 			writer:  outbound.Writer,
-			email:   user.Email,
-			manager: d.Wm,
+			manager: lm,
 		}
-		d.Wm.AddWriter(managedWriter)
+		managedReader := &ManagedReader{
+			reader:  &buf.TimeoutWrapperReader{Reader: outbound.Reader},
+			manager: lm,
+		}
+		lm.AddLink(managedWriter, managedReader)
 		outbound.Writer = managedWriter
+		outbound.Reader = managedReader
 		if w != nil {
 			outbound.Writer = rate.NewRateLimitWriter(outbound.Writer, w)
 		}
@@ -403,7 +428,7 @@ func (d *DefaultDispatcher) DispatchLink(ctx context.Context, destination net.De
 		ts := t.GetCounter(user.Email)
 		downcounter := &counter.XrayTrafficCounter{V: &ts.DownCounter}
 		outbound.Reader = &CounterReader{
-			Reader:  &buf.TimeoutWrapperReader{Reader: outbound.Reader},
+			Reader:  managedReader,
 			Counter: &ts.UpCounter,
 		}
 		outbound.Writer = &dispatcher.SizeStatWriter{

core/xray/app/dispatcher/linkmanager.go

@@ -2,53 +2,15 @@ package dispatcher
 
 import (
 	sync "sync"
+	"time"
 
 	"github.com/xtls/xray-core/common"
 	"github.com/xtls/xray-core/common/buf"
 )
 
-type WriterManager struct {
-	writers map[string]map[*ManagedWriter]struct{}
-	mu      sync.Mutex
-}
-
-func (m *WriterManager) AddWriter(writer *ManagedWriter) {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-	if _, exists := m.writers[writer.email]; !exists {
-		m.writers[writer.email] = make(map[*ManagedWriter]struct{})
-	}
-	m.writers[writer.email][writer] = struct{}{}
-}
-
-func (m *WriterManager) RemoveWriter(writer *ManagedWriter) {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-
-	if _, exists := m.writers[writer.email]; !exists {
-		return
-	}
-	delete(m.writers[writer.email], writer)
-}
-
-func (m *WriterManager) RemoveWritersForUser(email string) {
-	m.mu.Lock()
-	defer m.mu.Unlock()
-
-	if _, exists := m.writers[email]; !exists {
-		return
-	}
-	for writer := range m.writers[email] {
-		delete(m.writers[email], writer)
-		common.Close(writer.writer)
-	}
-	delete(m.writers, email)
-}
-
 type ManagedWriter struct {
 	writer  buf.Writer
-	email   string
-	manager *WriterManager
+	manager *LinkManager
 }
 
 func (w *ManagedWriter) WriteMultiBuffer(mb buf.MultiBuffer) error {
@@ -59,3 +21,57 @@ func (w *ManagedWriter) Close() error {
 	w.manager.RemoveWriter(w)
 	return common.Close(w.writer)
 }
+
+type ManagedReader struct {
+	reader  buf.TimeoutReader
+	manager *LinkManager
+}
+
+func (r *ManagedReader) ReadMultiBuffer() (buf.MultiBuffer, error) {
+	return r.reader.ReadMultiBuffer()
+}
+
+func (r *ManagedReader) ReadMultiBufferTimeout(t time.Duration) (buf.MultiBuffer, error) {
+	return r.reader.ReadMultiBufferTimeout(t)
+}
+
+func (r *ManagedReader) Interrupt() {
+	r.manager.RemoveReader(r)
+	common.Interrupt(r.reader)
+}
+
+type LinkManager struct {
+	writers map[*ManagedWriter]struct{}
+	readers map[*ManagedReader]struct{}
+	mu      sync.Mutex
+}
+
+func (m *LinkManager) AddLink(writer *ManagedWriter, reader *ManagedReader) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.writers[writer] = struct{}{}
+	m.readers[reader] = struct{}{}
+}
+
+func (m *LinkManager) RemoveWriter(writer *ManagedWriter) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	delete(m.writers, writer)
+}
+
+func (m *LinkManager) RemoveReader(reader *ManagedReader) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	delete(m.readers, reader)
+}
+
+func (m *LinkManager) CloseAll() {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	for w := range m.writers {
+		common.Close(w)
+	}
+	for r := range m.readers {
+		common.Interrupt(r)
+	}
+}