-
Notifications
You must be signed in to change notification settings - Fork 4.6k
/
dialer.go
98 lines (83 loc) · 2.27 KB
/
dialer.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
package kcp
import (
"context"
"crypto/tls"
"io"
"sync/atomic"
"v2ray.com/core/common"
"v2ray.com/core/common/buf"
"v2ray.com/core/common/dice"
"v2ray.com/core/common/net"
"v2ray.com/core/transport/internet"
v2tls "v2ray.com/core/transport/internet/tls"
)
var (
globalConv = uint32(dice.RollUint16())
)
func fetchInput(ctx context.Context, input io.Reader, reader PacketReader, conn *Connection) {
cache := make(chan *buf.Buffer, 1024)
go func() {
for {
payload := buf.New()
if _, err := payload.ReadFrom(input); err != nil {
payload.Release()
close(cache)
return
}
select {
case cache <- payload:
default:
payload.Release()
}
}
}()
for payload := range cache {
segments := reader.Read(payload.Bytes())
payload.Release()
if len(segments) > 0 {
conn.Input(segments)
}
}
}
func DialKCP(ctx context.Context, dest net.Destination, streamSettings *internet.MemoryStreamConfig) (internet.Connection, error) {
dest.Network = net.Network_UDP
newError("dialing mKCP to ", dest).WriteToLog()
rawConn, err := internet.DialSystem(ctx, dest, streamSettings.SocketSettings)
if err != nil {
return nil, newError("failed to dial to dest: ", err).AtWarning().Base(err)
}
kcpSettings := streamSettings.ProtocolSettings.(*Config)
header, err := kcpSettings.GetPackerHeader()
if err != nil {
return nil, newError("failed to create packet header").Base(err)
}
security, err := kcpSettings.GetSecurity()
if err != nil {
return nil, newError("failed to create security").Base(err)
}
reader := &KCPPacketReader{
Header: header,
Security: security,
}
writer := &KCPPacketWriter{
Header: header,
Security: security,
Writer: rawConn,
}
conv := uint16(atomic.AddUint32(&globalConv, 1))
session := NewConnection(ConnMetadata{
LocalAddr: rawConn.LocalAddr(),
RemoteAddr: rawConn.RemoteAddr(),
Conversation: conv,
}, writer, rawConn, kcpSettings)
go fetchInput(ctx, rawConn, reader, session)
var iConn internet.Connection = session
if config := v2tls.ConfigFromStreamSettings(streamSettings); config != nil {
tlsConn := tls.Client(iConn, config.GetTLSConfig(v2tls.WithDestination(dest)))
iConn = tlsConn
}
return iConn, nil
}
func init() {
common.Must(internet.RegisterTransportDialer(protocolName, DialKCP))
}