lzhr/v2fly
1
0
Fork 0
mirror of https://github.com/v2fly/v2ray-core.git synced 2025-01-02 07:26:24 -05:00
Code Issues Releases Wiki Activity

refine dispatcher

Browse Source
This commit is contained in:
Darien Raymond 2016-11-13 14:33:00 +01:00
parent 5dc05d6352
commit 9471b5b066
No known key found for this signature in database
GPG Key ID: 7251FFA14BB18169
13 changed files with 31 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
app/dispatcher/dispatcher.go
View File

@ -12,5 +12,5 @@ const (
// PacketDispatcher dispatch a packet and possibly further network payload to its destination.
type PacketDispatcher interface {
DispatchToOutbound(meta *proxy.InboundHandlerMeta, session *proxy.SessionInfo) ray.InboundRay
DispatchToOutbound(session *proxy.SessionInfo) ray.InboundRay
}

4
app/dispatcher/impl/default.go
View File

@ -43,7 +43,7 @@ func (this *DefaultDispatcher) Release() {
}
func (this *DefaultDispatcher) DispatchToOutbound(meta *proxy.InboundHandlerMeta, session *proxy.SessionInfo) ray.InboundRay {
func (this *DefaultDispatcher) DispatchToOutbound(session *proxy.SessionInfo) ray.InboundRay {
direct := ray.NewRay()
dispatcher := this.ohm.GetDefaultHandler()
destination := session.Destination
@ -61,7 +61,7 @@ func (this *DefaultDispatcher) DispatchToOutbound(meta *proxy.InboundHandlerMeta
}
}
if meta.AllowPassiveConnection {
if session.Inbound != nil && session.Inbound.AllowPassiveConnection {
go dispatcher.Dispatch(destination, alloc.NewLocalBuffer(32).Clear(), direct)
} else {
go this.FilterPacketAndDispatch(destination, direct, dispatcher)

2
app/dispatcher/testing/dispatcher.go
View File

@ -30,7 +30,7 @@ func NewTestPacketDispatcher(handler func(destination v2net.Destination, traffic
}
}
func (this *TestPacketDispatcher) DispatchToOutbound(meta *proxy.InboundHandlerMeta, session *proxy.SessionInfo) ray.InboundRay {
func (this *TestPacketDispatcher) DispatchToOutbound(session *proxy.SessionInfo) ray.InboundRay {
traffic := ray.NewRay()
this.Destination <- session.Destination
go this.Handler(session.Destination, traffic)

8
app/dns/nameserver.go
View File

@ -50,11 +50,9 @@ type UDPNameServer struct {
func NewUDPNameServer(address v2net.Destination, dispatcher dispatcher.PacketDispatcher) *UDPNameServer {
s := &UDPNameServer{
address: address,
requests: make(map[uint16]*PendingRequest),
udpServer: udp.NewUDPServer(&proxy.InboundHandlerMeta{
AllowPassiveConnection: false,
}, dispatcher),
address: address,
requests: make(map[uint16]*PendingRequest),
udpServer: udp.NewUDPServer(dispatcher),
}
return s
}

6
proxy/dokodemo/dokodemo.go
View File

@ -90,7 +90,7 @@ func (this *DokodemoDoor) Start() error {
}
func (this *DokodemoDoor) ListenUDP() error {
this.udpServer = udp.NewUDPServer(this.meta, this.packetDispatcher)
this.udpServer = udp.NewUDPServer(this.packetDispatcher)
udpHub, err := udp.ListenUDP(
this.meta.Address, this.meta.Port, udp.ListenOption{
Callback: this.handleUDPPackets,
@ -114,6 +114,7 @@ func (this *DokodemoDoor) handleUDPPackets(payload *alloc.Buffer, session *proxy
log.Info("Dokodemo: Unknown destination, stop forwarding...")
return
}
session.Inbound = this.meta
this.udpServer.Dispatch(session, payload, this.handleUDPResponse)
}
@ -160,9 +161,10 @@ func (this *DokodemoDoor) HandleTCPConnection(conn internet.Connection) {
}
log.Info("Dokodemo: Handling request to ", dest)
ray := this.packetDispatcher.DispatchToOutbound(this.meta, &proxy.SessionInfo{
ray := this.packetDispatcher.DispatchToOutbound(&proxy.SessionInfo{
Source: v2net.DestinationFromAddr(conn.RemoteAddr()),
Destination: dest,
Inbound: this.meta,
})
defer ray.InboundOutput().Release()

5
proxy/http/server.go
View File

@ -124,6 +124,7 @@ func (this *Server) handleConnection(conn internet.Connection) {
session := &proxy.SessionInfo{
Source: v2net.DestinationFromAddr(conn.RemoteAddr()),
Destination: dest,
Inbound: this.meta,
}
if strings.ToUpper(request.Method) == "CONNECT" {
this.handleConnect(request, session, reader, conn)
@ -146,7 +147,7 @@ func (this *Server) handleConnect(request *http.Request, session *proxy.SessionI
}
response.Write(writer)
ray := this.packetDispatcher.DispatchToOutbound(this.meta, session)
ray := this.packetDispatcher.DispatchToOutbound(session)
this.transport(reader, writer, ray)
}
@ -226,7 +227,7 @@ func (this *Server) handlePlainHTTP(request *http.Request, session *proxy.Sessio
request.Host = request.URL.Host
StripHopByHopHeaders(request)
ray := this.packetDispatcher.DispatchToOutbound(this.meta, session)
ray := this.packetDispatcher.DispatchToOutbound(session)
defer ray.InboundInput().Close()
defer ray.InboundOutput().Release()

1
proxy/proxy.go
View File

@ -20,6 +20,7 @@ type SessionInfo struct {
Source v2net.Destination
Destination v2net.Destination
User *protocol.User
Inbound *InboundHandlerMeta
}
type InboundHandlerMeta struct {

7
proxy/shadowsocks/server.go
View File

@ -90,7 +90,7 @@ func (this *Server) Start() error {
this.tcpHub = tcpHub
if this.config.UdpEnabled {
this.udpServer = udp.NewUDPServer(this.meta, this.packetDispatcher)
this.udpServer = udp.NewUDPServer(this.packetDispatcher)
udpHub, err := udp.ListenUDP(this.meta.Address, this.meta.Port, udp.ListenOption{Callback: this.handlerUDPPayload})
if err != nil {
log.Error("Shadowsocks: Failed to listen UDP on ", this.meta.Address, ":", this.meta.Port, ": ", err)
@ -130,7 +130,7 @@ func (this *Server) handlerUDPPayload(payload *alloc.Buffer, session *proxy.Sess
log.Access(source, dest, log.AccessAccepted, "")
log.Info("Shadowsocks|Server: Tunnelling request to ", dest)
this.udpServer.Dispatch(&proxy.SessionInfo{Source: source, Destination: dest, User: request.User}, data, func(destination v2net.Destination, payload *alloc.Buffer) {
this.udpServer.Dispatch(&proxy.SessionInfo{Source: source, Destination: dest, User: request.User, Inbound: this.meta}, data, func(destination v2net.Destination, payload *alloc.Buffer) {
defer payload.Release()
data, err := EncodeUDPPacket(request, payload)
@ -171,10 +171,11 @@ func (this *Server) handleConnection(conn internet.Connection) {
log.Access(conn.RemoteAddr(), dest, log.AccessAccepted, "")
log.Info("Shadowsocks|Server: Tunnelling request to ", dest)
ray := this.packetDispatcher.DispatchToOutbound(this.meta, &proxy.SessionInfo{
ray := this.packetDispatcher.DispatchToOutbound(&proxy.SessionInfo{
Source: v2net.DestinationFromAddr(conn.RemoteAddr()),
Destination: dest,
User: request.User,
Inbound: this.meta,
})
defer ray.InboundOutput().Release()

3
proxy/socks/server.go
View File

@ -285,6 +285,7 @@ func (this *Server) handleSocks4(clientAddr v2net.Destination, reader *v2io.Buff
session := &proxy.SessionInfo{
Source: clientAddr,
Destination: dest,
Inbound: this.meta,
}
log.Access(clientAddr, dest, log.AccessAccepted, "")
this.transport(reader, writer, session)
@ -292,7 +293,7 @@ func (this *Server) handleSocks4(clientAddr v2net.Destination, reader *v2io.Buff
}
func (this *Server) transport(reader io.Reader, writer io.Writer, session *proxy.SessionInfo) {
ray := this.packetDispatcher.DispatchToOutbound(this.meta, session)
ray := this.packetDispatcher.DispatchToOutbound(session)
input := ray.InboundInput()
output := ray.InboundOutput()

4
proxy/socks/server_udp.go
View File

@ -10,7 +10,7 @@ import (
)
func (this *Server) listenUDP() error {
this.udpServer = udp.NewUDPServer(this.meta, this.packetDispatcher)
this.udpServer = udp.NewUDPServer(this.packetDispatcher)
udpHub, err := udp.ListenUDP(this.meta.Address, this.meta.Port, udp.ListenOption{Callback: this.handleUDPPayload})
if err != nil {
log.Error("Socks: Failed to listen on udp ", this.meta.Address, ":", this.meta.Port)
@ -46,7 +46,7 @@ func (this *Server) handleUDPPayload(payload *alloc.Buffer, session *proxy.Sessi
log.Info("Socks: Send packet to ", request.Destination(), " with ", request.Data.Len(), " bytes")
log.Access(source, request.Destination, log.AccessAccepted, "")
this.udpServer.Dispatch(&proxy.SessionInfo{Source: source, Destination: request.Destination()}, request.Data, func(destination v2net.Destination, payload *alloc.Buffer) {
this.udpServer.Dispatch(&proxy.SessionInfo{Source: source, Destination: request.Destination(), Inbound: this.meta}, request.Data, func(destination v2net.Destination, payload *alloc.Buffer) {
response := &protocol.Socks5UDPRequest{
Fragment: 0,
Address: request.Destination().Address,

7
proxy/testing/mocks/inboundhandler.go
View File

@ -31,11 +31,12 @@ func (this *InboundConnectionHandler) Close() {
}
func (this *InboundConnectionHandler) Communicate(destination v2net.Destination) error {
ray := this.PacketDispatcher.DispatchToOutbound(&proxy.InboundHandlerMeta{
AllowPassiveConnection: false,
}, &proxy.SessionInfo{
ray := this.PacketDispatcher.DispatchToOutbound(&proxy.SessionInfo{
Source: v2net.TCPDestination(v2net.LocalHostIP, v2net.Port(0)),
Destination: destination,
Inbound: &proxy.InboundHandlerMeta{
AllowPassiveConnection: false,
},
})
input := ray.InboundInput()

3
proxy/vmess/inbound/inbound.go
View File

@ -166,10 +166,11 @@ func (this *VMessInboundHandler) HandleConnection(connection internet.Connection
connection.SetReusable(request.Option.Has(protocol.RequestOptionConnectionReuse))
ray := this.packetDispatcher.DispatchToOutbound(this.meta, &proxy.SessionInfo{
ray := this.packetDispatcher.DispatchToOutbound(&proxy.SessionInfo{
Source: v2net.DestinationFromAddr(connection.RemoteAddr()),
Destination: request.Destination(),
User: request.User,
Inbound: this.meta,
})
input := ray.InboundInput()
output := ray.InboundOutput()

6
transport/internet/udp/udp_server.go
View File

@ -96,14 +96,12 @@ type UDPServer struct {
sync.RWMutex
conns map[string]*TimedInboundRay
packetDispatcher dispatcher.PacketDispatcher
meta *proxy.InboundHandlerMeta
}
func NewUDPServer(meta *proxy.InboundHandlerMeta, packetDispatcher dispatcher.PacketDispatcher) *UDPServer {
func NewUDPServer(packetDispatcher dispatcher.PacketDispatcher) *UDPServer {
return &UDPServer{
conns: make(map[string]*TimedInboundRay),
packetDispatcher: packetDispatcher,
meta: meta,
}
}
@ -144,7 +142,7 @@ func (this *UDPServer) Dispatch(session *proxy.SessionInfo, payload *alloc.Buffe
}
log.Info("UDP Server: establishing new connection for ", destString)
inboundRay := this.packetDispatcher.DispatchToOutbound(this.meta, session)
inboundRay := this.packetDispatcher.DispatchToOutbound(session)
timedInboundRay := NewTimedInboundRay(destString, inboundRay, this)
outputStream := timedInboundRay.InboundInput()
if outputStream != nil {