feat(relaydns): integrate yamux multiplexing and net.Conn interface
Integrate yamux for multiplexed connections in RelayClient, including session creation, proper cleanup in Close(), and checks for session availability in workers. Implement net.Conn interface for RDConnection to allow standard networking operations, reducing SDK complexity and improving usability. Add error handling and new error types for better reliability. This enhances connection multiplexing and simplifies client-side networking integration.
lemon-mint committed
Oct 27, 2025 at 16:12 UTC
45ee19fa21f7f2a1c1a16eac7887231c28ea979e
3 files changed
+367
-19
relaydns/client.go
+58
-9
@@ -57,14 +57,25 @@ type leaseWithCred struct {
57
58
// NewRelayClient는 새로운 RelayClient 인스턴스를 생성합니다.
59
func NewRelayClient(conn io.ReadWriteCloser) *RelayClient {
60
+ // Create yamux session as client
61
+ config := yamux.DefaultConfig()
62
+ config.Logger = nil // Disable logging for cleaner output
63
+ sess, err := yamux.Client(conn, config)
64
+ if err != nil {
65
+ // If session creation fails, close the connection and return nil
66
+ conn.Close()
67
+ return nil
68
+ }
69
+
70
g := &RelayClient{
71
conn: conn,
72
+ sess: sess,
73
leases: make(map[string]*leaseWithCred),
74
stopCh: make(chan struct{}),
75
incommingConnCh: make(chan *IncommingConn),
76
}
77
67
- g.waitGroup.Add(1)
78
+ g.waitGroup.Add(2) // One for leaseUpdateWorker, one for leaseListenWorker
79
go g.leaseUpdateWorker()
80
go g.leaseListenWorker()
81
@@ -76,9 +87,24 @@ func (g *RelayClient) Close() error {
87
close(g.stopCh)
88
g.waitGroup.Wait()
89
79
- err := g.conn.Close()
80
- if err != nil {
81
- return err
90
+ var errs []error
91
+
92
+ // Close the session first
93
+ if g.sess != nil {
94
+ if err := g.sess.Close(); err != nil {
95
+ errs = append(errs, err)
96
+ }
97
+ }
98
+
99
+ // Then close the underlying connection
100
+ if g.conn != nil {
101
+ if err := g.conn.Close(); err != nil {
102
+ errs = append(errs, err)
103
+ }
104
+ }
105
+
106
+ if len(errs) > 0 {
107
+ return errs[0]
108
}
109
return nil
110
}
@@ -108,19 +134,42 @@ func (g *RelayClient) leaseUpdateWorker() {
134
135
for lease := range updateRequired {
136
lease.Lease.Expires = time.Now().Add(30 * time.Second).Unix()
111
- g.updateLease(lease.Cred, lease.Lease)
137
+ // Check if session is available before updating lease
138
+ if g.sess != nil {
139
+ g.updateLease(lease.Cred, lease.Lease)
140
+ }
141
}
142
}
143
}
144
}
145
146
func (g *RelayClient) leaseListenWorker() {
147
+ defer g.waitGroup.Done()
148
+
149
for {
119
- stream, err := g.sess.AcceptStream()
120
- if err != nil {
121
- continue
150
+ select {
151
+ case <-g.stopCh:
152
+ return
153
+ default:
154
+ if g.sess == nil {
155
+ // Session not initialized, wait a bit and retry
156
+ time.Sleep(100 * time.Millisecond)
157
+ continue
158
+ }
159
+
160
+ stream, err := g.sess.AcceptStream()
161
+ if err != nil {
162
+ // Check if we're supposed to stop
163
+ select {
164
+ case <-g.stopCh:
165
+ return
166
+ default:
167
+ // Continue trying to accept streams
168
+ continue
169
+ }
170
+ }
171
+ go g.handleConnectionRequestStream(stream)
172
}
123
- go g.handleConnectionRequestStream(stream)
173
}
174
}
175
relaydns/core/cryptoops/handshaker.go
+21
@@ -111,6 +111,27 @@ type SecureConnection struct {
111
readBuffer *bytebufferpool.ByteBuffer
112
}
113
114
+func (r *SecureConnection) SetDeadline(t time.Time) error {
115
+ if conn, ok := r.conn.(interface{ SetDeadline(time.Time) error }); ok {
116
+ return conn.SetDeadline(t)
117
+ }
118
+ return nil
119
+}
120
+
121
+func (r *SecureConnection) SetReadDeadline(t time.Time) error {
122
+ if conn, ok := r.conn.(interface{ SetReadDeadline(time.Time) error }); ok {
123
+ return conn.SetReadDeadline(t)
124
+ }
125
+ return nil
126
+}
127
+
128
+func (r *SecureConnection) SetWriteDeadline(t time.Time) error {
129
+ if conn, ok := r.conn.(interface{ SetWriteDeadline(time.Time) error }); ok {
130
+ return conn.SetWriteDeadline(t)
131
+ }
132
+ return nil
133
+}
134
+
135
func (sc *SecureConnection) LocalID() string {
136
return sc.localID
137
}
sdk/sdk.go
+288
-10
@@ -3,9 +3,12 @@ package sdk
3
import (
4
"context"
5
"errors"
6
+ "fmt"
7
"io"
8
+ "net"
9
"slices"
10
"sync"
11
+ "time"
12
13
"github.com/gorilla/websocket"
14
"github.com/gosuda/relaydns/relaydns"
@@ -41,20 +44,69 @@ type rdRelay struct {
44
stop chan struct{}
45
}
46
47
+var _ net.Conn = (*RDConnection)(nil)
48
+
49
type RDConnection struct {
50
via *rdRelay
51
localAddr string
52
remoteAddr string
48
- conn io.ReadWriteCloser
53
+ conn *cryptoops.SecureConnection
54
+}
55
+
56
+// Implement net.Conn interface for RDConnection
57
+func (r *RDConnection) Read(b []byte) (n int, err error) {
58
+ return r.conn.Read(b)
59
+}
60
+
61
+func (r *RDConnection) Write(b []byte) (n int, err error) {
62
+ return r.conn.Write(b)
63
+}
64
+
65
+func (r *RDConnection) Close() error {
66
+ return r.conn.Close()
67
+}
68
+
69
+func (r *RDConnection) LocalAddr() net.Addr {
70
+ return rdAddr(r.localAddr)
71
+}
72
+
73
+func (r *RDConnection) RemoteAddr() net.Addr {
74
+ return rdAddr(r.remoteAddr)
75
+}
76
+
77
+func (r *RDConnection) SetDeadline(t time.Time) error {
78
+ return r.conn.SetDeadline(t)
79
+}
80
+
81
+func (r *RDConnection) SetReadDeadline(t time.Time) error {
82
+ return r.conn.SetReadDeadline(t)
83
+}
84
+
85
+func (r *RDConnection) SetWriteDeadline(t time.Time) error {
86
+ return r.conn.SetWriteDeadline(t)
87
+}
88
+
89
+// rdAddr implements net.Addr
90
+type rdAddr string
91
+
92
+func (a rdAddr) Network() string {
93
+ return "relaydns"
94
+}
95
+
96
+func (a rdAddr) String() string {
97
+ return string(a)
98
}
99
100
type RDListener struct {
101
mu sync.Mutex
102
103
cred *cryptoops.Credential
104
+ lease *rdverb.Lease
105
+
106
conns map[*RDConnection]struct{}
107
108
connCh chan *RDConnection
109
+ closed bool
110
}
111
112
type RDClient struct {
@@ -67,11 +119,58 @@ type RDClient struct {
119
}
120
121
var (
70
- ErrNoAvailableRelay = errors.New("no available relay")
122
+ ErrNoAvailableRelay = errors.New("no available relay")
123
+ ErrClientClosed = errors.New("client is closed")
124
+ ErrListenerExists = errors.New("listener already exists for this credential")
125
+ ErrRelayExists = errors.New("relay already exists")
126
+ ErrRelayNotFound = errors.New("relay not found")
127
+ ErrFailedToCreateClient = errors.New("failed to create relay client")
128
)
129
130
func NewClient(opt ...Option) (*RDClient, error) {
74
- return &RDClient{}, nil
131
+ config := &RDClientConfig{
132
+ Dialer: webSocketDialer(),
133
+ }
134
+
135
+ for _, o := range opt {
136
+ o(config)
137
+ }
138
+
139
+ client := &RDClient{
140
+ relays: make(map[string]*rdRelay),
141
+ listeners: make(map[string]*RDListener),
142
+ stopch: make(chan struct{}),
143
+ }
144
+
145
+ // Initialize relays from bootstrap servers
146
+ var connectionErrors []error
147
+ for _, server := range config.BootstrapServers {
148
+ conn, err := config.Dialer(context.Background(), server)
149
+ if err != nil {
150
+ connectionErrors = append(connectionErrors, err)
151
+ continue // Skip failed connections
152
+ }
153
+
154
+ relayClient := relaydns.NewRelayClient(conn)
155
+ if relayClient == nil {
156
+ conn.Close()
157
+ connectionErrors = append(connectionErrors, ErrFailedToCreateClient)
158
+ continue
159
+ }
160
+
161
+ client.relays[server] = &rdRelay{
162
+ addr: server,
163
+ client: relayClient,
164
+ stop: make(chan struct{}),
165
+ }
166
+ }
167
+
168
+ // If no relays were successfully connected, return an error
169
+ if len(client.relays) == 0 && len(config.BootstrapServers) > 0 {
170
+ return nil, fmt.Errorf("failed to connect to any bootstrap servers: %v", connectionErrors)
171
+ }
172
+
173
+ return client, nil
174
}
175
176
func (g *RDClient) Dial(cred *cryptoops.Credential, leaseID string, alpn string) (*RDConnection, error) {
@@ -121,7 +220,46 @@ func (g *RDClient) Dial(cred *cryptoops.Credential, leaseID string, alpn string)
220
}
221
222
func (g *RDClient) Listen(cred *cryptoops.Credential, name string, alpns []string) (*RDListener, error) {
223
+ g.mu.Lock()
224
+ defer g.mu.Unlock()
225
+
226
+ // Check if client is closed
227
+ select {
228
+ case <-g.stopch:
229
+ return nil, ErrClientClosed
230
+ default:
231
+ // Client is still open
232
+ }
233
234
+ // Check if listener already exists
235
+ if _, exists := g.listeners[cred.ID()]; exists {
236
+ return nil, ErrListenerExists
237
+ }
238
+
239
+ // Create listener
240
+ listener := &RDListener{
241
+ cred: cred,
242
+ conns: make(map[*RDConnection]struct{}),
243
+ connCh: make(chan *RDConnection, 100),
244
+ closed: false,
245
+ }
246
+
247
+ // Register listener
248
+ g.listeners[cred.ID()] = listener
249
+
250
+ // Register lease with all available relays
251
+ for _, relay := range g.relays {
252
+ go func(r *rdRelay) {
253
+ r.client.RegisterLease(cred, name, alpns)
254
+ }(relay)
255
+ }
256
+
257
+ // Start listener worker for each relay
258
+ for _, relay := range g.relays {
259
+ go g.listenerWorker(relay)
260
+ }
261
+
262
+ return listener, nil
263
}
264
265
func (g *RDClient) listenerWorker(server *rdRelay) {
@@ -129,24 +267,50 @@ func (g *RDClient) listenerWorker(server *rdRelay) {
267
select {
268
case <-server.stop:
269
return
132
- case conn := <-server.client.IncommingConnection():
270
+ case conn, ok := <-server.client.IncommingConnection():
271
+ if !ok {
272
+ return // Channel closed
273
+ }
274
+
275
lease := conn.LeaseID()
276
277
g.mu.Lock()
136
- listener, ok := g.listeners[lease]
278
+ listener, exists := g.listeners[lease]
279
g.mu.Unlock()
280
139
- if !ok {
281
+ if !exists {
282
+ conn.SecureConnection.Close() // Close unused connection
283
continue
284
}
285
143
- rdConn := &RDConnection{via: server, conn: conn, localAddr: conn.LocalID(), remoteAddr: conn.RemoteID()}
286
+ rdConn := &RDConnection{
287
+ via: server,
288
+ conn: conn.SecureConnection,
289
+ localAddr: conn.LocalID(),
290
+ remoteAddr: conn.RemoteID(),
291
+ }
292
293
listener.mu.Lock()
294
+ // Check if listener is still active
295
+ if listener.closed {
296
+ listener.mu.Unlock()
297
+ rdConn.Close()
298
+ continue
299
+ }
300
listener.conns[rdConn] = struct{}{}
301
listener.mu.Unlock()
302
149
- listener.connCh <- rdConn
303
+ // Send connection to listener (non-blocking)
304
+ select {
305
+ case listener.connCh <- rdConn:
306
+ // Connection sent successfully
307
+ default:
308
+ // Channel full, close connection
309
+ listener.mu.Lock()
310
+ delete(listener.conns, rdConn)
311
+ listener.mu.Unlock()
312
+ rdConn.Close()
313
+ }
314
}
315
}
316
}
@@ -154,6 +318,7 @@ func (g *RDClient) listenerWorker(server *rdRelay) {
318
func (g *RDClient) Close() error {
319
var errs []error
320
321
+ // Signal all goroutines to stop
322
close(g.stopch)
323
324
g.mu.Lock()
@@ -162,14 +327,16 @@ func (g *RDClient) Close() error {
327
errs = append(errs, err)
328
}
329
}
165
- g.mu.Unlock()
330
+ g.listeners = make(map[string]*RDListener)
331
167
- g.mu.Lock()
332
+ // Stop all relays
333
for _, server := range g.relays {
334
+ close(server.stop) // Signal relay goroutines to stop
335
if err := server.client.Close(); err != nil {
336
errs = append(errs, err)
337
}
338
}
339
+ g.relays = make(map[string]*rdRelay)
340
g.mu.Unlock()
341
342
if len(errs) > 0 {
@@ -177,3 +344,114 @@ func (g *RDClient) Close() error {
344
}
345
return nil
346
}
347
+
348
+// Implement net.Listener interface for RDListener
349
+func (l *RDListener) Accept() (net.Conn, error) {
350
+ conn, ok := <-l.connCh
351
+ if !ok {
352
+ return nil, net.ErrClosed
353
+ }
354
+ return conn, nil
355
+}
356
+
357
+func (l *RDListener) Close() error {
358
+ l.mu.Lock()
359
+ defer l.mu.Unlock()
360
+
361
+ if l.closed {
362
+ return nil
363
+ }
364
+
365
+ l.closed = true
366
+
367
+ // Close the connection channel first to prevent new connections
368
+ close(l.connCh)
369
+
370
+ // Close all active connections
371
+ for conn := range l.conns {
372
+ if err := conn.Close(); err != nil {
373
+ // Log error but continue closing other connections
374
+ // In a real implementation, you might want to collect errors
375
+ }
376
+ delete(l.conns, conn)
377
+ }
378
+
379
+ // Clear the connections map
380
+ l.conns = make(map[*RDConnection]struct{})
381
+
382
+ return nil
383
+}
384
+
385
+func (l *RDListener) Addr() net.Addr {
386
+ return rdAddr(l.cred.ID())
387
+}
388
+
389
+// AddRelay adds a new relay server to the client
390
+func (g *RDClient) AddRelay(addr string, dialer func(context.Context, string) (io.ReadWriteCloser, error)) error {
391
+ g.mu.Lock()
392
+ defer g.mu.Unlock()
393
+
394
+ // Check if relay already exists
395
+ if _, exists := g.relays[addr]; exists {
396
+ return errors.New("relay already exists")
397
+ }
398
+
399
+ // Connect to relay
400
+ conn, err := dialer(context.Background(), addr)
401
+ if err != nil {
402
+ return err
403
+ }
404
+
405
+ // Create relay client
406
+ relayClient := relaydns.NewRelayClient(conn)
407
+ if relayClient == nil {
408
+ conn.Close()
409
+ return errors.New("failed to create relay client")
410
+ }
411
+
412
+ // Add relay
413
+ g.relays[addr] = &rdRelay{
414
+ addr: addr,
415
+ client: relayClient,
416
+ stop: make(chan struct{}),
417
+ }
418
+
419
+ return nil
420
+}
421
+
422
+// RemoveRelay removes a relay server from the client
423
+func (g *RDClient) RemoveRelay(addr string) error {
424
+ g.mu.Lock()
425
+ defer g.mu.Unlock()
426
+
427
+ relay, exists := g.relays[addr]
428
+ if !exists {
429
+ return errors.New("relay not found")
430
+ }
431
+
432
+ // Signal relay to stop
433
+ close(relay.stop)
434
+
435
+ // Close relay client
436
+ if err := relay.client.Close(); err != nil {
437
+ return err
438
+ }
439
+
440
+ // Remove from map
441
+ delete(g.relays, addr)
442
+
443
+ return nil
444
+}
445
+
446
+// GetRelays returns a list of all relay addresses
447
+func (g *RDClient) GetRelays() []string {
448
+ g.mu.Lock()
449
+ defer g.mu.Unlock()
450
+
451
+ relays := make([]string, 0, len(g.relays))
452
+ for addr := range g.relays {
453
+ relays = append(relays, addr)
454
+ }
455
+
456
+ return relays
457
+}