test: Xray内核删除用户时尝试关闭连接

core/xray/app/dispatcher/default.go

@@ -102,8 +102,8 @@ type DefaultDispatcher struct {
 	router routing.Router
 	policy policy.Manager
 	stats  stats.Manager
-	dns    dns.Client
 	fdns   dns.FakeDNSEngine
+	Wm     *WriterManager
 }
 
 func init() {
@@ -127,6 +127,9 @@ 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
 }
 
@@ -148,9 +151,14 @@ func (d *DefaultDispatcher) getLink(ctx context.Context, network net.Network) (*
 	uplinkReader, uplinkWriter := pipe.New(opt...)
 	downlinkReader, downlinkWriter := pipe.New(opt...)
 
+	managedWriter := &ManagedWriter{
+		writer:  uplinkWriter,
+		manager: d.Wm,
+	}
+
 	inboundLink := &transport.Link{
 		Reader: downlinkReader,
-		Writer: uplinkWriter,
+		Writer: managedWriter,
 	}
 
 	outboundLink := &transport.Link{
@@ -214,6 +222,8 @@ func (d *DefaultDispatcher) getLink(ctx context.Context, network net.Network) (*
 			}
 		}
 	}
+	managedWriter.email = user.Email
+	d.Wm.AddWriter(managedWriter)
 
 	return inboundLink, outboundLink, limit, nil
 }

core/xray/app/dispatcher/linkmanager.go

@@ -0,0 +1,61 @@
+package dispatcher
+
+import (
+	sync "sync"
+
+	"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
+}
+
+func (w *ManagedWriter) WriteMultiBuffer(mb buf.MultiBuffer) error {
+	return w.writer.WriteMultiBuffer(mb)
+}
+
+func (w *ManagedWriter) Close() error {
+	w.manager.RemoveWriter(w)
+	return common.Close(w.writer)
+}

core/xray/user.go

@@ -43,6 +43,7 @@ func (c *Xray) DelUsers(users []panel.UserInfo, tag string, _ *panel.NodeInfo) e
 		down = "user>>>" + user + ">>>traffic>>>downlink"
 		c.shm.UnregisterCounter(up)
 		c.shm.UnregisterCounter(down)
+		c.dispatcher.Wm.RemoveWritersForUser(user)
 	}
 	return nil
 }