sdk: separate types and client
Kim committed
Nov 18, 2025 at 10:56 UTC
54a03fda2b578db732f1e9334f15438b88455852
3 files changed
+234
-221
sdk/client.go
renamed
+7
-221
@@ -3,10 +3,8 @@ package sdk
3
import (
4
"context"
5
"encoding/json"
6
- "errors"
6
"fmt"
7
"io"
9
- "net"
8
"strings"
9
"sync"
10
"time"
@@ -19,190 +17,18 @@ import (
17
"gosuda.org/portal/portal/core/proto/rdverb"
18
)
19
22
-type ClientConfig struct {
23
- BootstrapServers []string
24
- Dialer func(context.Context, string) (io.ReadWriteCloser, error)
25
- HealthCheckInterval time.Duration // Interval for health checks (default: 10 seconds)
26
- ReconnectMaxRetries int // Maximum reconnection attempts (default: 0 = infinite)
27
- ReconnectInterval time.Duration // Interval between reconnection attempts (default: 5 seconds)
28
-}
29
-
30
-type Option func(*ClientConfig)
31
-
32
-func WithBootstrapServers(servers []string) Option {
33
- return func(c *ClientConfig) {
34
- c.BootstrapServers = servers
35
- }
36
-}
37
-
38
-func WithDialer(dialer func(context.Context, string) (io.ReadWriteCloser, error)) Option {
39
- return func(c *ClientConfig) {
40
- c.Dialer = dialer
41
- }
42
-}
43
-
44
-func WithHealthCheckInterval(interval time.Duration) Option {
45
- return func(c *ClientConfig) {
46
- c.HealthCheckInterval = interval
47
- }
48
-}
49
-
50
-func WithReconnectMaxRetries(retries int) Option {
51
- return func(c *ClientConfig) {
52
- c.ReconnectMaxRetries = retries
53
- }
54
-}
55
-
56
-func WithReconnectInterval(interval time.Duration) Option {
57
- return func(c *ClientConfig) {
58
- c.ReconnectInterval = interval
59
- }
60
-}
61
-
62
-type connRelay struct {
63
- addr string
64
- client *portal.RelayClient
65
- dialer func(context.Context, string) (io.ReadWriteCloser, error)
66
- stop chan struct{}
67
- stopOnce sync.Once // Ensure stop channel is closed only once
68
- mu sync.Mutex
69
-}
70
-
71
-var _ net.Conn = (*connection)(nil)
72
-
73
-type connection struct {
74
- via *connRelay
75
- localAddr string
76
- remoteAddr string
77
- conn *cryptoops.SecureConnection
78
-}
79
-
80
-func (r *connection) Read(b []byte) (n int, err error) {
81
- return r.conn.Read(b)
82
-}
83
-
84
-func (r *connection) Write(b []byte) (n int, err error) {
85
- return r.conn.Write(b)
86
-}
87
-
88
-func (r *connection) Close() error {
89
- return r.conn.Close()
90
-}
91
-
92
-func (r *connection) LocalAddr() net.Addr {
93
- return addr(r.localAddr)
94
-}
95
-
96
-func (r *connection) RemoteAddr() net.Addr {
97
- return addr(r.remoteAddr)
98
-}
99
-
100
-func (r *connection) SetDeadline(t time.Time) error {
101
- return r.conn.SetDeadline(t)
102
-}
103
-
104
-func (r *connection) SetReadDeadline(t time.Time) error {
105
- return r.conn.SetReadDeadline(t)
106
-}
107
-
108
-func (r *connection) SetWriteDeadline(t time.Time) error {
109
- return r.conn.SetWriteDeadline(t)
110
-}
111
-
112
-var _ net.Addr = (*addr)(nil)
113
-
114
-type addr string
115
-
116
-func (a addr) Network() string {
117
- return "portal"
118
-}
119
-
120
-func (a addr) String() string {
121
- return string(a)
122
-}
123
-
124
-type Listener struct {
125
- mu sync.Mutex
126
-
127
- cred *cryptoops.Credential
128
- lease *rdverb.Lease
129
-
130
- conns map[*connection]struct{}
131
-
132
- connCh chan *connection
133
- closed bool
134
-}
135
-
136
-type Metadata struct {
137
- Description string `json:"description"`
138
- Tags []string `json:"tags"`
139
- Thumbnail string `json:"thumbnail"`
140
- Owner string `json:"owner"`
141
- Hide bool `json:"hide"`
142
-}
143
-
144
-func (m Metadata) isEmpty() bool {
145
- return m.Description == "" &&
146
- len(m.Tags) == 0 &&
147
- m.Thumbnail == "" &&
148
- m.Owner == ""
149
-}
150
-
151
-type MetadataOption func(*Metadata)
152
-
153
-func WithDescription(description string) MetadataOption {
154
- return func(m *Metadata) {
155
- m.Description = description
156
- }
157
-}
158
-
159
-func WithTags(tags []string) MetadataOption {
160
- return func(m *Metadata) {
161
- m.Tags = tags
162
- }
163
-}
164
-
165
-func WithThumbnail(thumbnail string) MetadataOption {
166
- return func(m *Metadata) {
167
- m.Thumbnail = thumbnail
168
- }
169
-}
170
-
171
-func WithOwner(owner string) MetadataOption {
172
- return func(m *Metadata) {
173
- m.Owner = owner
174
- }
175
-}
176
-
177
-func WithHide(hide bool) MetadataOption {
178
- return func(m *Metadata) {
179
- m.Hide = hide
180
- }
181
-}
182
-
20
type Client struct {
184
- mu sync.Mutex
21
+ config *ClientConfig
22
+ mu sync.Mutex
23
24
relays map[string]*connRelay
187
- listeners map[string]*Listener
188
- config *ClientConfig
25
+ listeners map[string]*listener
26
27
stopch chan struct{}
28
stopOnce sync.Once // Ensure stopch is closed only once
29
waitGroup sync.WaitGroup // Track all listener workers
30
}
31
195
-var (
196
- ErrNoAvailableRelay = errors.New("no available relay")
197
- ErrClientClosed = errors.New("client is closed")
198
- ErrListenerExists = errors.New("listener already exists for this credential")
199
- ErrRelayExists = errors.New("relay already exists")
200
- ErrRelayNotFound = errors.New("relay not found")
201
- ErrInvalidName = errors.New("lease name contains invalid characters (only alphanumeric, hyphen, underscore allowed)")
202
- ErrFailedToCreateClient = errors.New("failed to create relay client")
203
- ErrInvalidMetadata = errors.New("invalid metadata")
204
-)
205
-
32
func NewClient(opt ...Option) (*Client, error) {
33
log.Debug().Msg("[SDK] Creating new Client")
34
@@ -219,7 +45,7 @@ func NewClient(opt ...Option) (*Client, error) {
45
46
client := &Client{
47
relays: make(map[string]*connRelay),
222
- listeners: make(map[string]*Listener),
48
+ listeners: make(map[string]*listener),
49
config: config,
50
stopch: make(chan struct{}),
51
}
@@ -336,7 +162,7 @@ func (g *Client) Dial(cred *cryptoops.Credential, leaseID string, alpn string) (
162
return nil, ErrNoAvailableRelay
163
}
164
339
-func (g *Client) Listen(cred *cryptoops.Credential, name string, alpns []string, options ...MetadataOption) (*Listener, error) {
165
+func (g *Client) Listen(cred *cryptoops.Credential, name string, alpns []string, options ...MetadataOption) (*listener, error) {
166
log.Debug().
167
Str("lease_id", cred.ID()).
168
Str("name", name).
@@ -395,7 +221,7 @@ func (g *Client) Listen(cred *cryptoops.Credential, name string, alpns []string,
221
}
222
223
// Create listener with lease metadata for re-registration
398
- listener := &Listener{
224
+ listener := &listener{
225
cred: cred,
226
lease: lease,
227
conns: make(map[*connection]struct{}),
@@ -515,7 +341,7 @@ func (g *Client) Close() error {
341
})
342
343
g.mu.Lock()
518
- listeners := make([]*Listener, 0, len(g.listeners))
344
+ listeners := make([]*listener, 0, len(g.listeners))
345
for _, listener := range g.listeners {
346
listeners = append(listeners, listener)
347
}
@@ -675,46 +501,6 @@ func (g *Client) reconnectRelay(relay *connRelay) {
501
}()
502
}
503
678
-// Implement net.Listener interface for Listener
679
-func (l *Listener) Accept() (net.Conn, error) {
680
- conn, ok := <-l.connCh
681
- if !ok {
682
- return nil, net.ErrClosed
683
- }
684
- return conn, nil
685
-}
686
-
687
-func (l *Listener) Close() error {
688
- l.mu.Lock()
689
- defer l.mu.Unlock()
690
-
691
- if l.closed {
692
- return nil
693
- }
694
-
695
- l.closed = true
696
-
697
- // Close the connection channel first to prevent new connections
698
- close(l.connCh)
699
-
700
- // Close all active connections
701
- for conn := range l.conns {
702
- if err := conn.Close(); err != nil {
703
- log.Error().Err(err).Msg("[SDK] Error closing connection")
704
- }
705
- delete(l.conns, conn)
706
- }
707
-
708
- // Clear the connections map
709
- l.conns = make(map[*connection]struct{})
710
-
711
- return nil
712
-}
713
-
714
-func (l *Listener) Addr() net.Addr {
715
- return addr(l.cred.ID())
716
-}
717
-
504
// AddRelay adds a new relay server to the client
505
func (g *Client) AddRelay(addr string, dialer func(context.Context, string) (io.ReadWriteCloser, error)) error {
506
g.mu.Lock()
sdk/client_e2e_test.go
renamed
sdk/types.go
new
+227
@@ -0,0 +1,227 @@
1
+package sdk
2
+
3
+import (
4
+ "context"
5
+ "errors"
6
+ "io"
7
+ "net"
8
+ "sync"
9
+ "time"
10
+
11
+ "github.com/rs/zerolog/log"
12
+ "gosuda.org/portal/portal"
13
+ "gosuda.org/portal/portal/core/cryptoops"
14
+ "gosuda.org/portal/portal/core/proto/rdverb"
15
+)
16
+
17
+var (
18
+ ErrNoAvailableRelay = errors.New("no available relay")
19
+ ErrClientClosed = errors.New("client is closed")
20
+ ErrListenerExists = errors.New("listener already exists for this credential")
21
+ ErrRelayExists = errors.New("relay already exists")
22
+ ErrRelayNotFound = errors.New("relay not found")
23
+ ErrInvalidName = errors.New("lease name contains invalid characters (only alphanumeric, hyphen, underscore allowed)")
24
+ ErrFailedToCreateClient = errors.New("failed to create relay client")
25
+ ErrInvalidMetadata = errors.New("invalid metadata")
26
+)
27
+
28
+type ClientConfig struct {
29
+ BootstrapServers []string
30
+ Dialer func(context.Context, string) (io.ReadWriteCloser, error)
31
+ HealthCheckInterval time.Duration // Interval for health checks (default: 10 seconds)
32
+ ReconnectMaxRetries int // Maximum reconnection attempts (default: 0 = infinite)
33
+ ReconnectInterval time.Duration // Interval between reconnection attempts (default: 5 seconds)
34
+}
35
+
36
+type Option func(*ClientConfig)
37
+
38
+func WithBootstrapServers(servers []string) Option {
39
+ return func(c *ClientConfig) {
40
+ c.BootstrapServers = servers
41
+ }
42
+}
43
+
44
+func WithDialer(dialer func(context.Context, string) (io.ReadWriteCloser, error)) Option {
45
+ return func(c *ClientConfig) {
46
+ c.Dialer = dialer
47
+ }
48
+}
49
+
50
+func WithHealthCheckInterval(interval time.Duration) Option {
51
+ return func(c *ClientConfig) {
52
+ c.HealthCheckInterval = interval
53
+ }
54
+}
55
+
56
+func WithReconnectMaxRetries(retries int) Option {
57
+ return func(c *ClientConfig) {
58
+ c.ReconnectMaxRetries = retries
59
+ }
60
+}
61
+
62
+func WithReconnectInterval(interval time.Duration) Option {
63
+ return func(c *ClientConfig) {
64
+ c.ReconnectInterval = interval
65
+ }
66
+}
67
+
68
+type Metadata struct {
69
+ Description string `json:"description"`
70
+ Tags []string `json:"tags"`
71
+ Thumbnail string `json:"thumbnail"`
72
+ Owner string `json:"owner"`
73
+ Hide bool `json:"hide"`
74
+}
75
+
76
+func (m Metadata) isEmpty() bool {
77
+ return m.Description == "" &&
78
+ len(m.Tags) == 0 &&
79
+ m.Thumbnail == "" &&
80
+ m.Owner == ""
81
+}
82
+
83
+type MetadataOption func(*Metadata)
84
+
85
+func WithDescription(description string) MetadataOption {
86
+ return func(m *Metadata) {
87
+ m.Description = description
88
+ }
89
+}
90
+
91
+func WithTags(tags []string) MetadataOption {
92
+ return func(m *Metadata) {
93
+ m.Tags = tags
94
+ }
95
+}
96
+
97
+func WithThumbnail(thumbnail string) MetadataOption {
98
+ return func(m *Metadata) {
99
+ m.Thumbnail = thumbnail
100
+ }
101
+}
102
+
103
+func WithOwner(owner string) MetadataOption {
104
+ return func(m *Metadata) {
105
+ m.Owner = owner
106
+ }
107
+}
108
+
109
+func WithHide(hide bool) MetadataOption {
110
+ return func(m *Metadata) {
111
+ m.Hide = hide
112
+ }
113
+}
114
+
115
+type listener struct {
116
+ mu sync.Mutex
117
+
118
+ cred *cryptoops.Credential
119
+ lease *rdverb.Lease
120
+
121
+ conns map[*connection]struct{}
122
+
123
+ connCh chan *connection
124
+ closed bool
125
+}
126
+
127
+// Implement net.Listener interface for Listener
128
+func (l *listener) Accept() (net.Conn, error) {
129
+ conn, ok := <-l.connCh
130
+ if !ok {
131
+ return nil, net.ErrClosed
132
+ }
133
+ return conn, nil
134
+}
135
+
136
+func (l *listener) Close() error {
137
+ l.mu.Lock()
138
+ defer l.mu.Unlock()
139
+
140
+ if l.closed {
141
+ return nil
142
+ }
143
+
144
+ l.closed = true
145
+
146
+ // Close the connection channel first to prevent new connections
147
+ close(l.connCh)
148
+
149
+ // Close all active connections
150
+ for conn := range l.conns {
151
+ if err := conn.Close(); err != nil {
152
+ log.Error().Err(err).Msg("[SDK] Error closing connection")
153
+ }
154
+ delete(l.conns, conn)
155
+ }
156
+
157
+ // Clear the connections map
158
+ l.conns = make(map[*connection]struct{})
159
+
160
+ return nil
161
+}
162
+
163
+func (l *listener) Addr() net.Addr {
164
+ return addr(l.cred.ID())
165
+}
166
+
167
+type connRelay struct {
168
+ addr string
169
+ client *portal.RelayClient
170
+ dialer func(context.Context, string) (io.ReadWriteCloser, error)
171
+ stop chan struct{}
172
+ stopOnce sync.Once // Ensure stop channel is closed only once
173
+ mu sync.Mutex
174
+}
175
+
176
+var _ net.Conn = (*connection)(nil)
177
+
178
+type connection struct {
179
+ via *connRelay
180
+ localAddr string
181
+ remoteAddr string
182
+ conn *cryptoops.SecureConnection
183
+}
184
+
185
+func (r *connection) Read(b []byte) (n int, err error) {
186
+ return r.conn.Read(b)
187
+}
188
+
189
+func (r *connection) Write(b []byte) (n int, err error) {
190
+ return r.conn.Write(b)
191
+}
192
+
193
+func (r *connection) Close() error {
194
+ return r.conn.Close()
195
+}
196
+
197
+func (r *connection) LocalAddr() net.Addr {
198
+ return addr(r.localAddr)
199
+}
200
+
201
+func (r *connection) RemoteAddr() net.Addr {
202
+ return addr(r.remoteAddr)
203
+}
204
+
205
+func (r *connection) SetDeadline(t time.Time) error {
206
+ return r.conn.SetDeadline(t)
207
+}
208
+
209
+func (r *connection) SetReadDeadline(t time.Time) error {
210
+ return r.conn.SetReadDeadline(t)
211
+}
212
+
213
+func (r *connection) SetWriteDeadline(t time.Time) error {
214
+ return r.conn.SetWriteDeadline(t)
215
+}
216
+
217
+var _ net.Addr = (*addr)(nil)
218
+
219
+type addr string
220
+
221
+func (a addr) Network() string {
222
+ return "portal"
223
+}
224
+
225
+func (a addr) String() string {
226
+ return string(a)
227
+}