1
0
mirror of https://github.com/v2fly/v2ray-core.git synced 2024-12-22 10:08:15 -05:00

unified dispatcher interface

This commit is contained in:
Darien Raymond 2017-02-03 22:50:01 +01:00
parent c4d0227977
commit 43fb425fd7
No known key found for this signature in database
GPG Key ID: 7251FFA14BB18169
6 changed files with 43 additions and 52 deletions

View File

@ -11,8 +11,6 @@ import (
// Interface dispatch a packet and possibly further network payload to its destination. // Interface dispatch a packet and possibly further network payload to its destination.
type Interface interface { type Interface interface {
Dispatch(ctx context.Context, dest net.Destination) (ray.InboundRay, error) Dispatch(ctx context.Context, dest net.Destination) (ray.InboundRay, error)
Start() error
Close()
} }
func FromSpace(space app.Space) Interface { func FromSpace(space app.Space) Interface {

View File

@ -9,8 +9,6 @@ import (
// A Server is a DNS server for responding DNS queries. // A Server is a DNS server for responding DNS queries.
type Server interface { type Server interface {
Get(domain string) []net.IP Get(domain string) []net.IP
Start() error
Close()
} }
func FromSpace(space app.Space) Server { func FromSpace(space app.Space) Server {

View File

@ -3,21 +3,17 @@ package inbound
import ( import (
"context" "context"
"v2ray.com/core/app"
"v2ray.com/core/app/dispatcher"
"v2ray.com/core/app/log" "v2ray.com/core/app/log"
"v2ray.com/core/app/proxyman" "v2ray.com/core/app/proxyman"
"v2ray.com/core/common/dice" "v2ray.com/core/common/dice"
"v2ray.com/core/common/errors"
"v2ray.com/core/common/net" "v2ray.com/core/common/net"
"v2ray.com/core/proxy" "v2ray.com/core/proxy"
"v2ray.com/core/transport/ray"
) )
type AlwaysOnInboundHandler struct { type AlwaysOnInboundHandler struct {
proxy proxy.Inbound proxy proxy.Inbound
workers []worker workers []worker
dispatcher dispatcher.Interface mux *mux
} }
func NewAlwaysOnInboundHandler(ctx context.Context, tag string, receiverConfig *proxyman.ReceiverConfig, proxyConfig interface{}) (*AlwaysOnInboundHandler, error) { func NewAlwaysOnInboundHandler(ctx context.Context, tag string, receiverConfig *proxyman.ReceiverConfig, proxyConfig interface{}) (*AlwaysOnInboundHandler, error) {
@ -28,18 +24,9 @@ func NewAlwaysOnInboundHandler(ctx context.Context, tag string, receiverConfig *
h := &AlwaysOnInboundHandler{ h := &AlwaysOnInboundHandler{
proxy: p, proxy: p,
mux: newMux(ctx),
} }
space := app.SpaceFromContext(ctx)
space.OnInitialize(func() error {
d := dispatcher.FromSpace(space)
if d == nil {
return errors.New("Proxyman|DefaultInboundHandler: No dispatcher in space.")
}
h.dispatcher = d
return nil
})
nl := p.Network() nl := p.Network()
pr := receiverConfig.PortRange pr := receiverConfig.PortRange
address := receiverConfig.Listen.AsAddress() address := receiverConfig.Listen.AsAddress()
@ -57,7 +44,7 @@ func NewAlwaysOnInboundHandler(ctx context.Context, tag string, receiverConfig *
recvOrigDest: receiverConfig.ReceiveOriginalDestination, recvOrigDest: receiverConfig.ReceiveOriginalDestination,
tag: tag, tag: tag,
allowPassiveConn: receiverConfig.AllowPassiveConnection, allowPassiveConn: receiverConfig.AllowPassiveConnection,
dispatcher: h, dispatcher: h.mux,
} }
h.workers = append(h.workers, worker) h.workers = append(h.workers, worker)
} }
@ -69,7 +56,7 @@ func NewAlwaysOnInboundHandler(ctx context.Context, tag string, receiverConfig *
address: address, address: address,
port: net.Port(port), port: net.Port(port),
recvOrigDest: receiverConfig.ReceiveOriginalDestination, recvOrigDest: receiverConfig.ReceiveOriginalDestination,
dispatcher: h, dispatcher: h.mux,
} }
h.workers = append(h.workers, worker) h.workers = append(h.workers, worker)
} }
@ -97,7 +84,3 @@ func (h *AlwaysOnInboundHandler) GetRandomInboundProxy() (proxy.Inbound, net.Por
w := h.workers[dice.Roll(len(h.workers))] w := h.workers[dice.Roll(len(h.workers))]
return w.Proxy(), w.Port(), 9999 return w.Proxy(), w.Port(), 9999
} }
func (h *AlwaysOnInboundHandler) Dispatch(ctx context.Context, dest net.Destination) (ray.InboundRay, error) {
return h.dispatcher.Dispatch(ctx, dest)
}

View File

@ -2,18 +2,14 @@ package inbound
import ( import (
"context" "context"
"errors"
"sync" "sync"
"time" "time"
"v2ray.com/core/app"
"v2ray.com/core/app/dispatcher"
"v2ray.com/core/app/log" "v2ray.com/core/app/log"
"v2ray.com/core/app/proxyman" "v2ray.com/core/app/proxyman"
"v2ray.com/core/common/dice" "v2ray.com/core/common/dice"
v2net "v2ray.com/core/common/net" v2net "v2ray.com/core/common/net"
"v2ray.com/core/proxy" "v2ray.com/core/proxy"
"v2ray.com/core/transport/ray"
) )
type DynamicInboundHandler struct { type DynamicInboundHandler struct {
@ -27,7 +23,7 @@ type DynamicInboundHandler struct {
workerMutex sync.RWMutex workerMutex sync.RWMutex
worker []worker worker []worker
lastRefresh time.Time lastRefresh time.Time
dispatcher dispatcher.Interface mux *mux
} }
func NewDynamicInboundHandler(ctx context.Context, tag string, receiverConfig *proxyman.ReceiverConfig, proxyConfig interface{}) (*DynamicInboundHandler, error) { func NewDynamicInboundHandler(ctx context.Context, tag string, receiverConfig *proxyman.ReceiverConfig, proxyConfig interface{}) (*DynamicInboundHandler, error) {
@ -39,18 +35,9 @@ func NewDynamicInboundHandler(ctx context.Context, tag string, receiverConfig *p
proxyConfig: proxyConfig, proxyConfig: proxyConfig,
receiverConfig: receiverConfig, receiverConfig: receiverConfig,
portsInUse: make(map[v2net.Port]bool), portsInUse: make(map[v2net.Port]bool),
mux: newMux(ctx),
} }
space := app.SpaceFromContext(ctx)
space.OnInitialize(func() error {
d := dispatcher.FromSpace(space)
if d == nil {
return errors.New("Proxyman|DefaultInboundHandler: No dispatcher in space.")
}
h.dispatcher = d
return nil
})
return h, nil return h, nil
} }
@ -117,7 +104,7 @@ func (h *DynamicInboundHandler) refresh() error {
stream: h.receiverConfig.StreamSettings, stream: h.receiverConfig.StreamSettings,
recvOrigDest: h.receiverConfig.ReceiveOriginalDestination, recvOrigDest: h.receiverConfig.ReceiveOriginalDestination,
allowPassiveConn: h.receiverConfig.AllowPassiveConnection, allowPassiveConn: h.receiverConfig.AllowPassiveConnection,
dispatcher: h, dispatcher: h.mux,
} }
if err := worker.Start(); err != nil { if err := worker.Start(); err != nil {
log.Warning("Proxyman:InboundHandler: Failed to create TCP worker: ", err) log.Warning("Proxyman:InboundHandler: Failed to create TCP worker: ", err)
@ -133,7 +120,7 @@ func (h *DynamicInboundHandler) refresh() error {
address: address, address: address,
port: port, port: port,
recvOrigDest: h.receiverConfig.ReceiveOriginalDestination, recvOrigDest: h.receiverConfig.ReceiveOriginalDestination,
dispatcher: h, dispatcher: h.mux,
} }
if err := worker.Start(); err != nil { if err := worker.Start(); err != nil {
log.Warning("Proxyman:InboundHandler: Failed to create UDP worker: ", err) log.Warning("Proxyman:InboundHandler: Failed to create UDP worker: ", err)
@ -181,7 +168,3 @@ func (h *DynamicInboundHandler) GetRandomInboundProxy() (proxy.Inbound, v2net.Po
expire := h.receiverConfig.AllocationStrategy.GetRefreshValue() - uint32(time.Since(h.lastRefresh)/time.Minute) expire := h.receiverConfig.AllocationStrategy.GetRefreshValue() - uint32(time.Since(h.lastRefresh)/time.Minute)
return w.Proxy(), w.Port(), int(expire) return w.Proxy(), w.Port(), int(expire)
} }
func (h *DynamicInboundHandler) Dispatch(ctx context.Context, dest v2net.Destination) (ray.InboundRay, error) {
return h.dispatcher.Dispatch(ctx, dest)
}

View File

@ -0,0 +1,33 @@
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)
}

View File

@ -13,8 +13,6 @@ import (
type InboundHandlerManager interface { type InboundHandlerManager interface {
GetHandler(ctx context.Context, tag string) (InboundHandler, error) GetHandler(ctx context.Context, tag string) (InboundHandler, error)
AddHandler(ctx context.Context, config *InboundHandlerConfig) error AddHandler(ctx context.Context, config *InboundHandlerConfig) error
Start() error
Close()
} }
type InboundHandler interface { type InboundHandler interface {
@ -29,8 +27,6 @@ type OutboundHandlerManager interface {
GetHandler(tag string) OutboundHandler GetHandler(tag string) OutboundHandler
GetDefaultHandler() OutboundHandler GetDefaultHandler() OutboundHandler
AddHandler(ctx context.Context, config *OutboundHandlerConfig) error AddHandler(ctx context.Context, config *OutboundHandlerConfig) error
Start() error
Close()
} }
type OutboundHandler interface { type OutboundHandler interface {