From 5686566139ebc8535637f77172d167db0b498919 Mon Sep 17 00:00:00 2001 From: Darien Raymond Date: Sun, 2 Apr 2017 21:30:21 +0200 Subject: [PATCH] apply mux in client --- app/proxyman/mux/mux.go | 56 ++++++++++++++++++++++++++++++-- app/proxyman/outbound/handler.go | 40 ++++++++++++++++------- tools/conf/v2ray.go | 18 ++++++++++ 3 files changed, 101 insertions(+), 13 deletions(-) diff --git a/app/proxyman/mux/mux.go b/app/proxyman/mux/mux.go index 67451247f..248da9700 100644 --- a/app/proxyman/mux/mux.go +++ b/app/proxyman/mux/mux.go @@ -57,6 +57,52 @@ func (s *session) closeDownlink() { s.checkAndRemove() } +type ClientManager struct { + access sync.Mutex + clients []*Client + proxy proxy.Outbound + dialer proxy.Dialer +} + +func NewClientManager(p proxy.Outbound, d proxy.Dialer) *ClientManager { + return &ClientManager{ + proxy: p, + dialer: d, + } +} + +func (m *ClientManager) Dispatch(ctx context.Context, outboundRay ray.OutboundRay) error { + m.access.Lock() + defer m.access.Unlock() + + for _, client := range m.clients { + if client.Dispatch(ctx, outboundRay) { + return nil + } + } + + client, err := NewClient(m.proxy, m.dialer, m) + if err != nil { + return err + } + m.clients = append(m.clients, client) + client.Dispatch(ctx, outboundRay) + return nil +} + +func (m *ClientManager) onClientFinish() { + m.access.Lock() + defer m.access.Unlock() + + nActive := 0 + for idx, client := range m.clients { + if nActive != idx && !client.Closed() { + m.clients[nActive] = client + } + } + m.clients = m.clients[:nActive] +} + type Client struct { access sync.RWMutex count uint16 @@ -64,11 +110,12 @@ type Client struct { inboundRay ray.InboundRay ctx context.Context cancel context.CancelFunc + manager *ClientManager } var muxCoolDestination = net.TCPDestination(net.DomainAddress("v1.mux.cool"), net.Port(9527)) -func NewClient(p proxy.Outbound, dialer proxy.Dialer) (*Client, error) { +func NewClient(p proxy.Outbound, dialer proxy.Dialer, m *ClientManager) (*Client, error) { ctx, cancel := context.WithCancel(context.Background()) ctx = proxy.ContextWithTarget(ctx, muxCoolDestination) pipe := ray.NewRay(ctx) @@ -82,6 +129,7 @@ func NewClient(p proxy.Outbound, dialer proxy.Dialer) (*Client, error) { inboundRay: pipe, ctx: ctx, cancel: cancel, + manager: m, }, nil } @@ -101,6 +149,7 @@ func (m *Client) remove(id uint16) { if len(m.sessions) == 0 { m.cancel() m.inboundRay.InboundInput().Close() + go m.manager.onClientFinish() } } @@ -121,7 +170,10 @@ func (m *Client) fetchInput(ctx context.Context, s *session) { writer: m.inboundRay.InboundInput(), } _, timer := signal.CancelAfterInactivity(ctx, time.Minute*5) - buf.PipeUntilEOF(timer, s.input, writer) + if err := buf.PipeUntilEOF(timer, s.input, writer); err != nil { + log.Info("Proxyman|Mux|Client: Failed to fetch all input: ", err) + } + writer.Close() s.closeUplink() } diff --git a/app/proxyman/outbound/handler.go b/app/proxyman/outbound/handler.go index 69b693769..721275c50 100644 --- a/app/proxyman/outbound/handler.go +++ b/app/proxyman/outbound/handler.go @@ -9,6 +9,7 @@ import ( "v2ray.com/core/app" "v2ray.com/core/app/log" "v2ray.com/core/app/proxyman" + "v2ray.com/core/app/proxyman/mux" "v2ray.com/core/common/buf" "v2ray.com/core/common/errors" v2net "v2ray.com/core/common/net" @@ -22,6 +23,7 @@ type Handler struct { senderSettings *proxyman.SenderConfig proxy proxy.Outbound outboundManager proxyman.OutboundHandlerManager + mux *mux.ClientManager } func NewHandler(ctx context.Context, config *proxyman.OutboundHandlerConfig) (*Handler, error) { @@ -54,6 +56,10 @@ func NewHandler(ctx context.Context, config *proxyman.OutboundHandlerConfig) (*H } } + if h.senderSettings != nil && h.senderSettings.MultiplexSettings != nil && h.senderSettings.MultiplexSettings.Enabled { + h.mux = mux.NewClientManager(h.proxy, h) + } + proxyHandler, err := config.GetProxyHandler(ctx) if err != nil { return nil, err @@ -64,20 +70,32 @@ func NewHandler(ctx context.Context, config *proxyman.OutboundHandlerConfig) (*H } func (h *Handler) Dispatch(ctx context.Context, outboundRay ray.OutboundRay) { - err := h.proxy.Process(ctx, outboundRay, h) - // Ensure outbound ray is properly closed. - if err != nil && errors.Cause(err) != io.EOF { - err = errors.Base(err).Message("Proxyman|OutboundHandler: Failed to process outbound traffic.") - if errors.IsActionRequired(err) { - log.Warning(err) - } else { - log.Info(err) + if h.mux != nil { + err := h.mux.Dispatch(ctx, outboundRay) + if err != nil { + err = errors.Base(err).Message("Proxyman|OutboundHandler: Failed to process outbound traffic.") + if errors.IsActionRequired(err) { + log.Warning(err) + } else { + log.Info(err) + } } - outboundRay.OutboundOutput().CloseError() } else { - outboundRay.OutboundOutput().Close() + err := h.proxy.Process(ctx, outboundRay, h) + // Ensure outbound ray is properly closed. + if err != nil && errors.Cause(err) != io.EOF { + err = errors.Base(err).Message("Proxyman|OutboundHandler: Failed to process outbound traffic.") + if errors.IsActionRequired(err) { + log.Warning(err) + } else { + log.Info(err) + } + outboundRay.OutboundOutput().CloseError() + } else { + outboundRay.OutboundOutput().Close() + } + outboundRay.OutboundInput().CloseError() } - outboundRay.OutboundInput().CloseError() } // Dial implements proxy.Dialer.Dial(). diff --git a/tools/conf/v2ray.go b/tools/conf/v2ray.go index ebd80adae..c5270a78a 100644 --- a/tools/conf/v2ray.go +++ b/tools/conf/v2ray.go @@ -80,6 +80,10 @@ func (v *InboundConnectionConfig) Build() (*proxyman.InboundHandlerConfig, error }, nil } +type MuxConfig struct { + Enabled bool `json:"enabled"` +} + type OutboundConnectionConfig struct { Protocol string `json:"protocol"` SendThrough *Address `json:"sendThrough"` @@ -87,6 +91,7 @@ type OutboundConnectionConfig struct { ProxySettings *ProxyConfig `json:"proxySettings"` Settings json.RawMessage `json:"settings"` Tag string `json:"tag"` + MuxSettings *MuxConfig `json:"mux"` } func (v *OutboundConnectionConfig) Build() (*proxyman.OutboundHandlerConfig, error) { @@ -114,6 +119,12 @@ func (v *OutboundConnectionConfig) Build() (*proxyman.OutboundHandlerConfig, err senderSettings.ProxySettings = ps } + if v.MuxSettings != nil && v.MuxSettings.Enabled { + senderSettings.MultiplexSettings = &proxyman.MultiplexingConfig{ + Enabled: true, + } + } + rawConfig, err := outboundConfigLoader.LoadWithID(v.Settings, v.Protocol) if err != nil { return nil, errors.Base(err).Message("Config: Failed to parse outbound config.") @@ -228,6 +239,7 @@ type OutboundDetourConfig struct { Settings json.RawMessage `json:"settings"` StreamSetting *StreamConfig `json:"streamSettings"` ProxySettings *ProxyConfig `json:"proxySettings"` + MuxSettings *MuxConfig `json:"mux"` } func (v *OutboundDetourConfig) Build() (*proxyman.OutboundHandlerConfig, error) { @@ -257,6 +269,12 @@ func (v *OutboundDetourConfig) Build() (*proxyman.OutboundHandlerConfig, error) senderSettings.ProxySettings = ps } + if v.MuxSettings != nil && v.MuxSettings.Enabled { + senderSettings.MultiplexSettings = &proxyman.MultiplexingConfig{ + Enabled: true, + } + } + rawConfig, err := outboundConfigLoader.LoadWithID(v.Settings, v.Protocol) if err != nil { return nil, errors.Base(err).Message("Config: Failed to parse to outbound detour config.")