-
Notifications
You must be signed in to change notification settings - Fork 4.6k
/
hub.go
126 lines (107 loc) · 3.14 KB
/
hub.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
//go:build !confonly
// +build !confonly
package grpc
import (
"context"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials"
"github.com/v2fly/v2ray-core/v4/common"
"github.com/v2fly/v2ray-core/v4/common/net"
"github.com/v2fly/v2ray-core/v4/common/session"
"github.com/v2fly/v2ray-core/v4/transport/internet"
"github.com/v2fly/v2ray-core/v4/transport/internet/grpc/encoding"
"github.com/v2fly/v2ray-core/v4/transport/internet/tls"
)
type Listener struct {
encoding.UnimplementedGunServiceServer
ctx context.Context
handler internet.ConnHandler
local net.Addr
config *Config
locker *internet.FileLocker // for unix domain socket
s *grpc.Server
}
func (l Listener) Tun(server encoding.GunService_TunServer) error {
tunCtx, cancel := context.WithCancel(l.ctx)
l.handler(encoding.NewGunConn(server, cancel))
<-tunCtx.Done()
return nil
}
func (l Listener) Close() error {
l.s.Stop()
return nil
}
func (l Listener) Addr() net.Addr {
return l.local
}
func Listen(ctx context.Context, address net.Address, port net.Port, settings *internet.MemoryStreamConfig, handler internet.ConnHandler) (internet.Listener, error) {
grpcSettings := settings.ProtocolSettings.(*Config)
var listener *Listener
if port == net.Port(0) { // unix
listener = &Listener{
handler: handler,
local: &net.UnixAddr{
Name: address.Domain(),
Net: "unix",
},
config: grpcSettings,
}
} else { // tcp
listener = &Listener{
handler: handler,
local: &net.TCPAddr{
IP: address.IP(),
Port: int(port),
},
config: grpcSettings,
}
}
listener.ctx = ctx
config := tls.ConfigFromStreamSettings(settings)
var s *grpc.Server
if config == nil {
s = grpc.NewServer()
} else {
// gRPC server may silently ignore TLS errors
s = grpc.NewServer(grpc.Creds(credentials.NewTLS(config.GetTLSConfig(tls.WithNextProto("h2")))))
}
listener.s = s
if settings.SocketSettings != nil && settings.SocketSettings.AcceptProxyProtocol {
newError("accepting PROXY protocol").AtWarning().WriteToLog(session.ExportIDToError(ctx))
}
go func() {
var streamListener net.Listener
var err error
if port == net.Port(0) { // unix
streamListener, err = internet.ListenSystem(ctx, &net.UnixAddr{
Name: address.Domain(),
Net: "unix",
}, settings.SocketSettings)
if err != nil {
newError("failed to listen on ", address).Base(err).AtError().WriteToLog(session.ExportIDToError(ctx))
return
}
locker := ctx.Value(address.Domain())
if locker != nil {
listener.locker = locker.(*internet.FileLocker)
}
} else { // tcp
streamListener, err = internet.ListenSystem(ctx, &net.TCPAddr{
IP: address.IP(),
Port: int(port),
}, settings.SocketSettings)
if err != nil {
newError("failed to listen on ", address, ":", port).Base(err).AtError().WriteToLog(session.ExportIDToError(ctx))
return
}
}
encoding.RegisterGunServiceServerX(s, listener, grpcSettings.ServiceName)
if err = s.Serve(streamListener); err != nil {
newError("Listener for grpc ended").Base(err).WriteToLog()
}
}()
return listener, nil
}
func init() {
common.Must(internet.RegisterTransportListener(protocolName, Listen))
}