apply mux in inbound handler

app/proxyman/inbound/always.go

@@ -5,6 +5,7 @@ import (
 
 	"v2ray.com/core/app/log"
 	"v2ray.com/core/app/proxyman"
+	"v2ray.com/core/app/proxyman/mux"
 	"v2ray.com/core/common/dice"
 	"v2ray.com/core/common/net"
 	"v2ray.com/core/proxy"
@@ -13,7 +14,7 @@ import (
 type AlwaysOnInboundHandler struct {
 	proxy   proxy.Inbound
 	workers []worker
-	mux     *mux
+	mux     *mux.Server
 }
 
 func NewAlwaysOnInboundHandler(ctx context.Context, tag string, receiverConfig *proxyman.ReceiverConfig, proxyConfig interface{}) (*AlwaysOnInboundHandler, error) {
@@ -24,7 +25,7 @@ func NewAlwaysOnInboundHandler(ctx context.Context, tag string, receiverConfig *
 
 	h := &AlwaysOnInboundHandler{
 		proxy: p,
-		mux:   newMux(ctx),
+		mux:   mux.NewServer(ctx),
 	}
 
 	nl := p.Network()

app/proxyman/inbound/dynamic.go

@@ -7,6 +7,7 @@ import (
 
 	"v2ray.com/core/app/log"
 	"v2ray.com/core/app/proxyman"
+	"v2ray.com/core/app/proxyman/mux"
 	"v2ray.com/core/common/dice"
 	v2net "v2ray.com/core/common/net"
 	"v2ray.com/core/proxy"
@@ -23,7 +24,7 @@ type DynamicInboundHandler struct {
 	workerMutex    sync.RWMutex
 	worker         []worker
 	lastRefresh    time.Time
-	mux            *mux
+	mux            *mux.Server
 }
 
 func NewDynamicInboundHandler(ctx context.Context, tag string, receiverConfig *proxyman.ReceiverConfig, proxyConfig interface{}) (*DynamicInboundHandler, error) {
@@ -35,7 +36,7 @@ func NewDynamicInboundHandler(ctx context.Context, tag string, receiverConfig *p
 		proxyConfig:    proxyConfig,
 		receiverConfig: receiverConfig,
 		portsInUse:     make(map[v2net.Port]bool),
-		mux:            newMux(ctx),
+		mux:            mux.NewServer(ctx),
 	}
 
 	return h, nil

app/proxyman/inbound/mux.go

@@ -1,33 +0,0 @@
-package inbound
-
-import (
-	"context"
-	"errors"
-
-	"v2ray.com/core/app"
-	"v2ray.com/core/app/dispatcher"
-	"v2ray.com/core/common/net"
-	"v2ray.com/core/transport/ray"
-)
-
-type mux struct {
-	dispatcher dispatcher.Interface
-}
-
-func newMux(ctx context.Context) *mux {
-	m := &mux{}
-	space := app.SpaceFromContext(ctx)
-	space.OnInitialize(func() error {
-		d := dispatcher.FromSpace(space)
-		if d == nil {
-			return errors.New("Proxyman|DefaultInboundHandler: No dispatcher in space.")
-		}
-		m.dispatcher = d
-		return nil
-	})
-	return m
-}
-
-func (m *mux) Dispatch(ctx context.Context, dest net.Destination) (ray.InboundRay, error) {
-	return m.dispatcher.Dispatch(ctx, dest)
-}

app/proxyman/mux/mux.go

@@ -5,9 +5,11 @@ import (
 	"sync"
 	"time"
 
+	"v2ray.com/core/app"
 	"v2ray.com/core/app/dispatcher"
 	"v2ray.com/core/app/log"
 	"v2ray.com/core/common/buf"
+	"v2ray.com/core/common/errors"
 	"v2ray.com/core/common/net"
 	"v2ray.com/core/common/signal"
 	"v2ray.com/core/proxy"
@@ -194,13 +196,32 @@ type Server struct {
 	dispatcher dispatcher.Interface
 }
 
+func NewServer(ctx context.Context) *Server {
+	s := &Server{}
+	space := app.SpaceFromContext(ctx)
+	space.OnInitialize(func() error {
+		d := dispatcher.FromSpace(space)
+		if d == nil {
+			return errors.New("Proxyman|Mux: No dispatcher in space.")
+		}
+		s.dispatcher = d
+		return nil
+	})
+	return s
+}
+
 func (s *Server) Dispatch(ctx context.Context, dest net.Destination) (ray.InboundRay, error) {
 	if dest != muxCoolDestination {
 		return s.dispatcher.Dispatch(ctx, dest)
 	}
 
 	ray := ray.NewRay(ctx)
-
+	worker := &ServerWorker{
+		dispatcher:  s.dispatcher,
+		outboundRay: ray,
+		sessions:    make(map[uint16]*session),
+	}
+	go worker.run(ctx)
 	return ray, nil
 }