@samitouri / QOSAMI-X402Tunnel-1 / commits / 0b104f1f

sdk: add exposure helper for multi relay

Chang committed Mar 8, 2026 at 11:17 UTC 0b104f1f837a4c7a548fa55454da4428be37579a
5 files changed +902 -255
cmd/demo-app/main.go
+25 -33
@@ -17,22 +17,22 @@ import (
17 )
18
19 var (
20 - flagServerURL string
21 - flagAddr string
22 - flagName string
23 - flagDesc string
24 - flagTags string
25 - flagOwner string
26 - flagHide bool
27 - flagThumbnail string
20 + flagServerURLs string
21 + flagAddr string
22 + flagName string
23 + flagDesc string
24 + flagTags string
25 + flagOwner string
26 + flagHide bool
27 + flagThumbnail string
28 )
29
30 func main() {
31 log.Logger = log.Output(zerolog.ConsoleWriter{Out: os.Stdout, TimeFormat: time.RFC3339})
32 logger := log.With().Str("component", "demo-app").Logger()
33
34 - flag.StringVar(&flagServerURL, "server-url", "https://localhost:4017", "relay API URL (https only)")
35 - flag.StringVar(&flagAddr, "addr", "127.0.0.1:8092", "local demo HTTP listen address (disable if empty)")
34 + flag.StringVar(&flagServerURLs, "server-urls", "https://localhost:4017", "relay API URLs (comma-separated; scheme omitted defaults to https)")
35 + flag.StringVar(&flagAddr, "addr", "127.0.0.1:8092", "local demo HTTP listen address (host:port or URL; disable if empty)")
36 flag.StringVar(&flagName, "name", "demo-app", "backend display name")
37 flag.StringVar(&flagDesc, "description", "Portal demo connectivity app", "lease description")
38 flag.StringVar(&flagTags, "tags", "demo,connectivity,activity,cloud,sun,morning", "comma-separated lease tags")
@@ -53,34 +53,26 @@ func runDemo() error {
53 ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
54 defer stop()
55
56 - sdkClient, err := sdk.NewClient(flagServerURL)
56 + exposure, err := sdk.Expose(ctx, sdk.SplitCSV(flagServerURLs), flagName, types.LeaseMetadata{
57 + Description: flagDesc,
58 + Tags: sdk.SplitCSV(flagTags),
59 + Owner: flagOwner,
60 + Thumbnail: flagThumbnail,
61 + Hide: flagHide,
62 + })
63 if err != nil {
58 - return fmt.Errorf("new client: %w", err)
64 + return fmt.Errorf("exposure listen error: %w", err)
65 + }
66 + defer exposure.Close()
67 + if exposure == nil {
68 + logger.Info().Msg("demo app running without relay")
69 }
60 - defer sdkClient.Close()
70
62 - listener, err := sdkClient.Listen(ctx, sdk.ListenRequest{
63 - Name: flagName,
64 - Metadata: types.LeaseMetadata{
65 - Description: flagDesc,
66 - Tags: sdk.SplitCSV(flagTags),
67 - Owner: flagOwner,
68 - Thumbnail: flagThumbnail,
69 - Hide: flagHide,
70 - },
71 - })
71 + flagAddr, err := sdk.NormalizeTargetAddr(flagAddr)
72 if err != nil {
73 - return fmt.Errorf("listen: %w", err)
73 + return fmt.Errorf("invalid --addr value %q: %w", flagAddr, err)
74 }
75 - defer listener.Close()
76 -
77 - logger.Info().
78 - Str("relay", flagServerURL).
79 - Str("lease_id", listener.LeaseID()).
80 - Strs("public_urls", listener.PublicURLs()).
81 - Str("local_addr", flagAddr).
82 - Msg("demo app registered with relay")
83 - if err := sdk.RunHTTP(ctx, listener, newHandler(), flagAddr); err != nil {
75 + if err := exposure.RunHTTP(ctx, newHandler(), flagAddr); err != nil {
76 return err
77 }
78
cmd/portal-tunnel/main.go
+14 -41
@@ -5,7 +5,6 @@ import (
5 "errors"
6 "flag"
7 "fmt"
8 - "net"
8 "os"
9 "os/signal"
10 "sync"
@@ -41,7 +40,7 @@ func main() {
40 defaultRelayURLs = "https://localhost:4017"
41 }
42
44 - flag.StringVar(&flagRelayURLs, "relays", defaultRelayURLs, "Portal relay server API URLs (comma-separated, https only) [env: RELAYS]")
43 + flag.StringVar(&flagRelayURLs, "relays", defaultRelayURLs, "Portal relay server API URLs (comma-separated; scheme omitted defaults to https) [env: RELAYS]")
44 flag.StringVar(&flagHost, "host", os.Getenv("APP_HOST"), "Target host to proxy to (host:port or URL) [env: APP_HOST]")
45 flag.StringVar(&flagName, "name", os.Getenv("APP_NAME"), "Service name [env: APP_NAME]")
46 flag.StringVar(&flagDesc, "description", os.Getenv("APP_DESCRIPTION"), "Service description metadata [env: APP_DESCRIPTION]")
@@ -63,64 +62,38 @@ func runTunnel() error {
62 ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
63 defer stop()
64
66 - relayURLs, err := normalizeRelayURLs(flagRelayURLs)
65 + exposure, err := sdk.Expose(ctx, sdk.SplitCSV(flagRelayURLs), flagName, types.LeaseMetadata{
66 + Description: flagDesc,
67 + Tags: sdk.SplitCSV(flagTags),
68 + Owner: flagOwner,
69 + Thumbnail: flagThumbnail,
70 + Hide: flagHide,
71 + })
72 if err != nil {
68 - return err
73 + return fmt.Errorf("service %s: failed to start relays: %w", flagName, err)
74 }
70 - if len(relayURLs) == 0 {
75 + if exposure == nil {
76 return errors.New("no relay URLs provided")
77 }
78 + defer exposure.Close()
79
80 logger.Info().
81 Str("local", flagHost).
76 - Int("relay_count", len(relayURLs)).
77 - Strs("relays", relayURLs).
82 Msg("starting portal tunnel")
83
80 - listenReq := sdk.ListenRequest{
81 - Name: flagName,
82 - Metadata: types.LeaseMetadata{
83 - Description: flagDesc,
84 - Tags: sdk.SplitCSV(flagTags),
85 - Owner: flagOwner,
86 - Thumbnail: flagThumbnail,
87 - Hide: flagHide,
88 - },
89 - }
90 -
91 - runtimes, err := startRelayRuntimes(ctx, relayURLs, listenReq)
92 - if err != nil {
93 - return fmt.Errorf("service %s: failed to start relays: %w", flagName, err)
94 - }
95 -
84 var connWG sync.WaitGroup
85 var connCount atomic.Int64
98 - listeners := make([]net.Listener, 0, len(runtimes))
99 -
100 - for _, runtime := range runtimes {
101 - logger.Info().
102 - Str("relay", runtime.relayURL).
103 - Str("lease_id", runtime.listener.LeaseID()).
104 - Strs("public_urls", runtime.listener.PublicURLs()).
105 - Msg("relay tunnel ready")
106 - listeners = append(listeners, runtime.listener)
107 - }
86
109 - relayListener, err := sdk.MergeListeners(listeners...)
110 - if err != nil {
111 - closeErr := closeRelayRuntimes(runtimes)
112 - return errors.Join(fmt.Errorf("merge relay listeners: %w", err), closeErr)
113 - }
87 go func() {
88 <-ctx.Done()
116 - _ = relayListener.Close()
89 + _ = exposure.Close()
90 }()
91
119 - waitErr := proxyRelayConnections(ctx, relayListener, flagHost, &connWG, &connCount)
92 + waitErr := proxyRelayConnections(ctx, exposure, flagHost, &connWG, &connCount)
93 if waitErr != nil {
94 stop()
95 }
123 - closeErr := errors.Join(relayListener.Close(), closeRelayRuntimes(runtimes))
96 + closeErr := exposure.Close()
97 if waitErr != nil {
98 logger.Error().Err(waitErr).Msg("relay supervisor exited with error")
99 }
cmd/portal-tunnel/relays.go
+1 -125
@@ -6,8 +6,6 @@ import (
6 "fmt"
7 "io"
8 "net"
9 - "net/url"
10 - "strings"
9 "sync"
10 "sync/atomic"
11 "time"
@@ -17,53 +15,6 @@ import (
15 "github.com/gosuda/portal/v2/sdk"
16 )
17
20 -type relayRuntime struct {
21 - client *sdk.Client
22 - listener *sdk.Listener
23 - relayURL string
24 -}
25 -
26 -func startRelayRuntimes(ctx context.Context, relayURLs []string, req sdk.ListenRequest) ([]*relayRuntime, error) {
27 - runtimes := make([]*relayRuntime, 0, len(relayURLs))
28 - for _, relayURL := range relayURLs {
29 - client, err := sdk.NewClient(relayURL)
30 - if err != nil {
31 - _ = closeRelayRuntimes(runtimes)
32 - return nil, fmt.Errorf("create relay client %s: %w", relayURL, err)
33 - }
34 -
35 - listener, err := client.Listen(ctx, req)
36 - if err != nil {
37 - client.Close()
38 - _ = closeRelayRuntimes(runtimes)
39 - return nil, fmt.Errorf("register relay lease %s: %w", relayURL, err)
40 - }
41 -
42 - runtimes = append(runtimes, &relayRuntime{
43 - relayURL: relayURL,
44 - client: client,
45 - listener: listener,
46 - })
47 - }
48 - return runtimes, nil
49 -}
50 -
51 -func closeRelayRuntimes(runtimes []*relayRuntime) error {
52 - var closeErr error
53 - for _, runtime := range runtimes {
54 - if runtime == nil {
55 - continue
56 - }
57 - if runtime.listener != nil {
58 - closeErr = errors.Join(closeErr, runtime.listener.Close())
59 - }
60 - if runtime.client != nil {
61 - runtime.client.Close()
62 - }
63 - }
64 - return closeErr
65 -}
66 -
18 func proxyRelayConnections(ctx context.Context, relayListener net.Listener, localAddr string, connWG *sync.WaitGroup, connCount *atomic.Int64) error {
19 logger := log.With().Str("component", "portal-tunnel").Logger()
20
@@ -97,40 +48,6 @@ func proxyRelayConnections(ctx context.Context, relayListener net.Listener, loca
48 }
49 }
50
100 -func normalizeRelayURLs(raw string) ([]string, error) {
101 - seen := make(map[string]struct{})
102 - var relayURLs []string
103 - for _, relayURL := range sdk.SplitCSV(raw) {
104 - normalized, err := normalizeRelayURL(relayURL)
105 - if err != nil {
106 - return nil, err
107 - }
108 - if _, ok := seen[normalized]; ok {
109 - continue
110 - }
111 - seen[normalized] = struct{}{}
112 - relayURLs = append(relayURLs, normalized)
113 - }
114 - return relayURLs, nil
115 -}
116 -
117 -func normalizeRelayURL(raw string) (string, error) {
118 - u, err := url.Parse(strings.TrimSpace(raw))
119 - if err != nil {
120 - return "", fmt.Errorf("parse relay url: %w", err)
121 - }
122 - if !strings.EqualFold(u.Scheme, "https") {
123 - return "", fmt.Errorf("relay url must use https: %q", raw)
124 - }
125 - if u.Host == "" {
126 - return "", fmt.Errorf("relay url host is empty: %q", raw)
127 - }
128 - u.Path = strings.TrimRight(u.Path, "/")
129 - u.RawQuery = ""
130 - u.Fragment = ""
131 - return u.String(), nil
132 -}
133 -
51 var bufferPool = sync.Pool{
52 New: func() any {
53 b := make([]byte, 64*1024)
@@ -141,7 +58,7 @@ var bufferPool = sync.Pool{
58 func proxyConnection(ctx context.Context, localAddr string, relayConn net.Conn) error {
59 defer relayConn.Close()
60
144 - targetAddr, err := normalizeTargetAddr(localAddr)
61 + targetAddr, err := sdk.NormalizeTargetAddr(localAddr)
62 if err != nil {
63 return fmt.Errorf("invalid --host value %q: %w", localAddr, err)
64 }
@@ -215,44 +132,3 @@ func writeEmptyHTTPResponse(conn net.Conn) error {
132 _, err := conn.Write([]byte(response))
133 return err
134 }
218 -
219 -func normalizeTargetAddr(raw string) (string, error) {
220 - raw = strings.TrimSpace(raw)
221 - if raw == "" {
222 - return "", errors.New("target address is required")
223 - }
224 -
225 - if strings.Contains(raw, "://") {
226 - targetURL, err := url.Parse(raw)
227 - if err != nil {
228 - return "", fmt.Errorf("parse target url: %w", err)
229 - }
230 - if !strings.EqualFold(targetURL.Scheme, "http") && !strings.EqualFold(targetURL.Scheme, "https") {
231 - return "", fmt.Errorf("unsupported target url scheme %q", targetURL.Scheme)
232 - }
233 - if targetURL.Host == "" {
234 - return "", errors.New("target url host is empty")
235 - }
236 - if targetURL.Path != "" && targetURL.Path != "/" {
237 - return "", errors.New("target url path is not supported")
238 - }
239 - if targetURL.RawQuery != "" {
240 - return "", errors.New("target url query is not supported")
241 - }
242 - if targetURL.Fragment != "" {
243 - return "", errors.New("target url fragment is not supported")
244 - }
245 - raw = targetURL.Host
246 - }
247 -
248 - if _, _, err := net.SplitHostPort(raw); err == nil {
249 - return raw, nil
250 - }
251 - if strings.Count(raw, ":") == 0 {
252 - return net.JoinHostPort(raw, "80"), nil
253 - }
254 - if ip := net.ParseIP(raw); ip != nil {
255 - return net.JoinHostPort(raw, "80"), nil
256 - }
257 - return "", fmt.Errorf("invalid target address %q", raw)
258 -}
sdk/helper.go
+470 -41
@@ -6,73 +6,378 @@ import (
6 "fmt"
7 "net"
8 "net/http"
9 + "net/url"
10 "strings"
11 "sync"
12 + "sync/atomic"
13
12 - "golang.org/x/sync/errgroup"
14 + "github.com/rs/zerolog/log"
15 +
16 + "github.com/gosuda/portal/v2/types"
17 )
18
15 -// RunHTTP serves one handler on the relay listener and, when localAddr is set,
16 -// on the provided local HTTP address for app-local access.
19 +// Exposure owns the lifecycle of one or more relay listeners plus their
20 +// clients and accepts traffic from all of them through one net.Listener.
21 +type Exposure struct {
22 + listener net.Listener
23 + relays []exposureRelay
24 +
25 + closeOnce sync.Once
26 + connSeq atomic.Uint64
27 +}
28 +
29 +// Expose creates relay listeners for each normalized relay URL and exposes a
30 +// merged listener for accepting traffic from all of them. Empty relay input
31 +// returns nil, nil so callers can fall back to local-only serving.
32 +func Expose(ctx context.Context, relayUrls []string, name string, metadata types.LeaseMetadata) (*Exposure, error) {
33 + relayURLs, err := NormalizeRelayURLs(relayUrls)
34 + if err != nil {
35 + return nil, err
36 + }
37 + if len(relayURLs) == 0 {
38 + return nil, nil
39 + }
40 +
41 + relays := make([]exposureRelay, 0, len(relayURLs))
42 + cleanup := func() error {
43 + var closeErr error
44 + for _, relay := range relays {
45 + if relay.listener != nil {
46 + closeErr = errors.Join(closeErr, relay.listener.Close())
47 + }
48 + if relay.client != nil {
49 + relay.client.Close()
50 + }
51 + }
52 + return closeErr
53 + }
54 +
55 + for _, relayURL := range relayURLs {
56 + client, err := NewClient(relayURL)
57 + if err != nil {
58 + return nil, errors.Join(fmt.Errorf("new client %q: %w", relayURL, err), cleanup())
59 + }
60 +
61 + listener, err := client.Listen(ctx, ListenRequest{
62 + Name: name,
63 + Metadata: metadata,
64 + })
65 + if err != nil {
66 + client.Close()
67 + return nil, errors.Join(fmt.Errorf("listen %q: %w", relayURL, err), cleanup())
68 + }
69 +
70 + relays = append(relays, exposureRelay{
71 + relayURL: relayURL,
72 + publicURLs: append([]string(nil), listener.PublicURLs()...),
73 + client: client,
74 + listener: listener,
75 + })
76 + }
77 +
78 + listeners := make([]net.Listener, 0, len(relays))
79 + for _, relay := range relays {
80 + listeners = append(listeners, relay.listener)
81 + }
82 +
83 + merged, err := mergeListeners(listeners...)
84 + if err != nil {
85 + return nil, errors.Join(fmt.Errorf("merge listeners: %w", err), cleanup())
86 + }
87 +
88 + exposure := &Exposure{
89 + listener: merged,
90 + relays: relays,
91 + }
92 +
93 + logger := log.With().Str("component", "sdk-exposure").Logger()
94 + logger.Info().
95 + Int("relay_count", len(exposure.relays)).
96 + Strs("relays", exposure.RelayURLs()).
97 + Strs("public_urls", exposure.PublicURLs()).
98 + Msg("exposure ready")
99 +
100 + return exposure, nil
101 +}
102 +
103 +// Accept implements net.Listener by accepting from the merged relay listener.
104 +func (e *Exposure) Accept() (net.Conn, error) {
105 + if e == nil || e.listener == nil {
106 + return nil, net.ErrClosed
107 + }
108 +
109 + conn, err := e.listener.Accept()
110 + if err != nil {
111 + if !errors.Is(err, net.ErrClosed) {
112 + logger := log.With().Str("component", "sdk-exposure").Logger()
113 + logger.Warn().
114 + Err(err).
115 + Str("local_addr", exposureAddrString(e.listener.Addr())).
116 + Msg("exposure accept failed")
117 + }
118 + return nil, err
119 + }
120 +
121 + connID := e.connSeq.Add(1)
122 + logger := log.With().Str("component", "sdk-exposure").Logger()
123 + logger.Info().
124 + Uint64("conn_id", connID).
125 + Str("local_addr", exposureAddrString(conn.LocalAddr())).
126 + Str("remote_addr", exposureAddrString(conn.RemoteAddr())).
127 + Msg("exposure connection accepted")
128 +
129 + return &exposureConn{
130 + Conn: conn,
131 + id: connID,
132 + localAddr: exposureAddrString(conn.LocalAddr()),
133 + remoteAddr: exposureAddrString(conn.RemoteAddr()),
134 + }, nil
135 +}
136 +
137 +// Addr implements net.Listener.
138 +func (e *Exposure) Addr() net.Addr {
139 + if e == nil || e.listener == nil {
140 + return listenerAddr("portal:exposure")
141 + }
142 + return e.listener.Addr()
143 +}
144 +
145 +// RelayURLs returns the normalized relay URLs backing the exposure.
146 +func (e *Exposure) RelayURLs() []string {
147 + if e == nil || len(e.relays) == 0 {
148 + return nil
149 + }
150 +
151 + out := make([]string, 0, len(e.relays))
152 + for _, relay := range e.relays {
153 + out = append(out, relay.relayURL)
154 + }
155 + return out
156 +}
157 +
158 +// PublicURLs returns the de-duplicated public URLs exposed by the exposure.
159 +func (e *Exposure) PublicURLs() []string {
160 + if e == nil || len(e.relays) == 0 {
161 + return nil
162 + }
163 +
164 + out := make([]string, 0, len(e.relays))
165 + seen := make(map[string]struct{})
166 + for _, relay := range e.relays {
167 + for _, rawURL := range relay.publicURLs {
168 + if _, ok := seen[rawURL]; ok {
169 + continue
170 + }
171 + seen[rawURL] = struct{}{}
172 + out = append(out, rawURL)
173 + }
174 + }
175 + if len(out) == 0 {
176 + return nil
177 + }
178 + return out
179 +}
180 +
181 +// RunHTTP serves one handler on the exposure and, when localAddr is set, on
182 +// the provided local HTTP address for app-local access. Nil exposures support
183 +// local-only serving.
184 +func (e *Exposure) RunHTTP(ctx context.Context, handler http.Handler, localAddr string) error {
185 + var relayListener net.Listener
186 + if e != nil {
187 + relayListener = e
188 + }
189 + return RunHTTP(ctx, relayListener, handler, localAddr)
190 +}
191 +
192 +// Close closes the merged listener and all underlying SDK clients.
193 +func (e *Exposure) Close() error {
194 + if e == nil {
195 + return nil
196 + }
197 +
198 + var closeErr error
199 + e.closeOnce.Do(func() {
200 + if e.listener != nil {
201 + closeErr = errors.Join(closeErr, e.listener.Close())
202 + }
203 + for _, relay := range e.relays {
204 + if relay.client != nil {
205 + relay.client.Close()
206 + }
207 + }
208 +
209 + logger := log.With().Str("component", "sdk-exposure").Logger()
210 + event := logger.Info().
211 + Int("relay_count", len(e.relays)).
212 + Strs("relays", e.RelayURLs())
213 + if closeErr != nil {
214 + event = logger.Warn().
215 + Err(closeErr).
216 + Int("relay_count", len(e.relays)).
217 + Strs("relays", e.RelayURLs())
218 + }
219 + event.Msg("exposure closed")
220 + })
221 + return closeErr
222 +}
223 +
224 +// RunHTTP serves one handler on relayListener and, when localAddr is set, on
225 +// the provided local HTTP address for app-local access.
226 func RunHTTP(ctx context.Context, relayListener net.Listener, handler http.Handler, localAddr string) error {
18 - relaySrv := &http.Server{
19 - Handler: handler,
20 - ReadHeaderTimeout: defaultRequestTimeout,
227 + localAddr = strings.TrimSpace(localAddr)
228 + if ctx == nil {
229 + ctx = context.Background()
230 + }
231 +
232 + if relayListener == nil && localAddr == "" {
233 + return errors.New("relay listener or local address is required")
234 + }
235 +
236 + var relaySrv *http.Server
237 + if relayListener != nil {
238 + relaySrv = &http.Server{
239 + Handler: handler,
240 + ReadHeaderTimeout: defaultRequestTimeout,
241 + }
242 }
243
244 var localSrv *http.Server
24 - if strings.TrimSpace(localAddr) != "" {
245 + if localAddr != "" {
246 localSrv = &http.Server{
26 - Addr: strings.TrimSpace(localAddr),
247 + Addr: localAddr,
248 Handler: handler,
249 ReadHeaderTimeout: defaultRequestTimeout,
250 }
251 }
252
32 - group, groupCtx := errgroup.WithContext(ctx)
253 + serverCount := 0
254 + if relaySrv != nil {
255 + serverCount++
256 + }
257 if localSrv != nil {
34 - group.Go(func() error {
35 - if err := localSrv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
36 - return fmt.Errorf("serve local http: %w", err)
37 - }
258 + serverCount++
259 + }
260 +
261 + results := make(chan error, serverCount)
262 + normalizeServeErr := func(err error, prefix string) error {
263 + if err == nil || errors.Is(err, http.ErrServerClosed) || errors.Is(err, net.ErrClosed) {
264 return nil
265 + }
266 + return fmt.Errorf("%s: %w", prefix, err)
267 + }
268 +
269 + var (
270 + shutdownOnce sync.Once
271 + shutdownErr error
272 + )
273 + shutdown := func() error {
274 + shutdownOnce.Do(func() {
275 + shutdownCtx, cancel := context.WithTimeout(context.Background(), defaultHTTPShutdownTimeout)
276 + defer cancel()
277 +
278 + var localErr error
279 + if localSrv != nil {
280 + localErr = localSrv.Shutdown(shutdownCtx)
281 + if errors.Is(localErr, http.ErrServerClosed) {
282 + localErr = nil
283 + }
284 + }
285 +
286 + var relayErr error
287 + if relaySrv != nil {
288 + relayErr = relaySrv.Shutdown(shutdownCtx)
289 + if errors.Is(relayErr, http.ErrServerClosed) {
290 + relayErr = nil
291 + }
292 + }
293 +
294 + shutdownErr = errors.Join(localErr, relayErr)
295 })
296 + return shutdownErr
297 }
41 - group.Go(func() error {
42 - if err := relaySrv.Serve(relayListener); err != nil && !errors.Is(err, http.ErrServerClosed) {
43 - return fmt.Errorf("serve relay http: %w", err)
44 - }
45 - return nil
46 - })
47 - group.Go(func() error {
48 - <-groupCtx.Done()
298
50 - shutdownCtx, cancel := context.WithTimeout(context.Background(), defaultHTTPShutdownTimeout)
51 - defer cancel()
299 + if localSrv != nil {
300 + go func() {
301 + results <- normalizeServeErr(localSrv.ListenAndServe(), "serve local http")
302 + }()
303 + }
304 + if relaySrv != nil {
305 + go func() {
306 + results <- normalizeServeErr(relaySrv.Serve(relayListener), "serve relay http")
307 + }()
308 + }
309
53 - var localErr error
54 - if localSrv != nil {
55 - localErr = localSrv.Shutdown(shutdownCtx)
56 - if errors.Is(localErr, http.ErrServerClosed) {
57 - localErr = nil
310 + var serveErr error
311 + remaining := serverCount
312 + ctxDone := ctx.Done()
313 + for remaining > 0 {
314 + select {
315 + case err := <-results:
316 + remaining--
317 + if err != nil {
318 + serveErr = errors.Join(serveErr, err)
319 + _ = shutdown()
320 }
321 + case <-ctxDone:
322 + _ = shutdown()
323 + ctxDone = nil
324 }
325 + }
326 +
327 + return errors.Join(serveErr, shutdownErr)
328 +}
329 +
330 +type exposureRelay struct {
331 + relayURL string
332 + publicURLs []string
333 + client *Client
334 + listener *Listener
335 +}
336 +
337 +type exposureConn struct {
338 + net.Conn
339 + id uint64
340 + localAddr string
341 + remoteAddr string
342 + closeOnce sync.Once
343 +}
344
61 - relayErr := relaySrv.Shutdown(shutdownCtx)
62 - if errors.Is(relayErr, http.ErrServerClosed) {
63 - relayErr = nil
345 +func (c *exposureConn) Close() error {
346 + var closeErr error
347 + c.closeOnce.Do(func() {
348 + closeErr = c.Conn.Close()
349 + if errors.Is(closeErr, net.ErrClosed) {
350 + closeErr = nil
351 }
352
66 - return errors.Join(localErr, relayErr)
353 + logger := log.With().Str("component", "sdk-exposure").Logger()
354 + event := logger.Info().
355 + Uint64("conn_id", c.id).
356 + Str("local_addr", c.localAddr).
357 + Str("remote_addr", c.remoteAddr)
358 + if closeErr != nil {
359 + event = logger.Warn().
360 + Err(closeErr).
361 + Uint64("conn_id", c.id).
362 + Str("local_addr", c.localAddr).
363 + Str("remote_addr", c.remoteAddr)
364 + }
365 + event.Msg("exposure connection closed")
366 })
367 + return closeErr
368 +}
369
69 - return group.Wait()
370 +func exposureAddrString(addr net.Addr) string {
371 + if addr == nil {
372 + return ""
373 + }
374 + return addr.String()
375 }
376
72 -// MergeListeners fans in multiple listeners into one net.Listener. It keeps
377 +// mergeListeners fans in multiple listeners into one net.Listener. It keeps
378 // serving accepts from remaining listeners when one listener stops, and returns
379 // a terminal error only after all source listeners have stopped.
75 -func MergeListeners(listeners ...net.Listener) (net.Listener, error) {
380 +func mergeListeners(listeners ...net.Listener) (net.Listener, error) {
381 if len(listeners) == 0 {
382 return nil, errors.New("at least one listener is required")
383 }
@@ -113,15 +418,16 @@ type mergedListener struct {
418 func (l *mergedListener) Accept() (net.Conn, error) {
419 conn, ok := <-l.accepted
420 if ok {
421 + select {
422 + case <-l.closed:
423 + _ = conn.Close()
424 + return nil, l.terminalErrorOr(net.ErrClosed)
425 + default:
426 + }
427 return conn, nil
428 }
429
119 - l.mu.Lock()
120 - defer l.mu.Unlock()
121 - if l.terminalErr == nil {
122 - return nil, net.ErrClosed
123 - }
124 - return nil, l.terminalErr
430 + return nil, l.terminalErrorOr(net.ErrClosed)
431 }
432
433 func (l *mergedListener) Close() error {
@@ -167,6 +473,14 @@ func (l *mergedListener) runAcceptLoop(listener net.Listener) {
473 return
474 }
475
476 + select {
477 + case <-l.closed:
478 + _ = conn.Close()
479 + l.finishWorker()
480 + return
481 + default:
482 + }
483 +
484 select {
485 case l.accepted <- conn:
486 case <-l.closed:
@@ -201,6 +515,15 @@ func (l *mergedListener) recordTerminalError(err error) {
515 l.mu.Unlock()
516 }
517
518 +func (l *mergedListener) terminalErrorOr(fallback error) error {
519 + l.mu.Lock()
520 + defer l.mu.Unlock()
521 + if l.terminalErr == nil {
522 + return fallback
523 + }
524 + return l.terminalErr
525 +}
526 +
527 // SplitCSV splits a comma-separated string, trimming whitespace and dropping
528 // empty entries.
529 func SplitCSV(raw string) []string {
@@ -218,3 +541,109 @@ func SplitCSV(raw string) []string {
541 }
542 return out
543 }
544 +
545 +// NormalizeRelayURLs splits, normalizes, and de-duplicates relay URLs while
546 +// preserving input order. Empty inputs return nil, nil.
547 +func NormalizeRelayURLs(inputs []string) ([]string, error) {
548 + out := make([]string, 0, len(inputs))
549 + seen := make(map[string]struct{}, len(inputs))
550 +
551 + for _, input := range inputs {
552 + for _, part := range SplitCSV(input) {
553 + normalized, err := NormalizeRelayURL(part)
554 + if err != nil {
555 + return nil, err
556 + }
557 + if _, ok := seen[normalized]; ok {
558 + continue
559 + }
560 + seen[normalized] = struct{}{}
561 + out = append(out, normalized)
562 + }
563 + }
564 +
565 + if len(out) == 0 {
566 + return nil, nil
567 + }
568 + return out, nil
569 +}
570 +
571 +// NormalizeRelayURL accepts host[:port] or https URLs and returns the canonical
572 +// relay base URL used by the SDK.
573 +func NormalizeRelayURL(raw string) (string, error) {
574 + trimmed := strings.TrimSpace(raw)
575 + if trimmed == "" {
576 + return "", errors.New("relay url is empty")
577 + }
578 + if !strings.Contains(trimmed, "://") {
579 + trimmed = "https://" + strings.TrimPrefix(trimmed, "//")
580 + }
581 +
582 + parsed, err := url.Parse(trimmed)
583 + if err != nil {
584 + return "", fmt.Errorf("parse relay url %q: %w", raw, err)
585 + }
586 + if parsed.Host == "" && parsed.Path != "" && !strings.Contains(parsed.Path, "/") {
587 + parsed, err = url.Parse("https://" + strings.TrimSpace(parsed.Path))
588 + if err != nil {
589 + return "", fmt.Errorf("parse relay url %q: %w", raw, err)
590 + }
591 + }
592 + if parsed.Host == "" {
593 + return "", fmt.Errorf("relay url host is empty: %q", raw)
594 + }
595 + if !strings.EqualFold(parsed.Scheme, "https") {
596 + return "", fmt.Errorf("relay url must use https: %q", raw)
597 + }
598 +
599 + parsed.RawQuery = ""
600 + parsed.Fragment = ""
601 + parsed.Path = strings.TrimRight(parsed.Path, "/")
602 + if strings.HasSuffix(strings.ToLower(parsed.Path), "/relay") {
603 + parsed.Path = strings.TrimSuffix(parsed.Path, "/relay")
604 + }
605 + return parsed.String(), nil
606 +}
607 +
608 +// NormalizeTargetAddr accepts host[:port] or http/https URLs and returns a
609 +// canonical host:port target address for local dialing.
610 +func NormalizeTargetAddr(raw string) (string, error) {
611 + raw = strings.TrimSpace(raw)
612 + if raw == "" {
613 + return "", errors.New("target address is required")
614 + }
615 +
616 + if strings.Contains(raw, "://") {
617 + targetURL, err := url.Parse(raw)
618 + if err != nil {
619 + return "", fmt.Errorf("parse target url: %w", err)
620 + }
621 + if !strings.EqualFold(targetURL.Scheme, "http") && !strings.EqualFold(targetURL.Scheme, "https") {
622 + return "", fmt.Errorf("unsupported target url scheme %q", targetURL.Scheme)
623 + }
624 + if targetURL.Host == "" {
625 + return "", errors.New("target url host is empty")
626 + }
627 + if targetURL.Path != "" && targetURL.Path != "/" {
628 + return "", errors.New("target url path is not supported")
629 + }
630 + if targetURL.RawQuery != "" {
631 + return "", errors.New("target url query is not supported")
632 + }
633 + if targetURL.Fragment != "" {
634 + return "", errors.New("target url fragment is not supported")
635 + }
636 + raw = targetURL.Host
637 + }
638 +
639 + if _, _, err := net.SplitHostPort(raw); err == nil {
640 + return raw, nil
641 + }
642 + if strings.Count(raw, ":") == 0 {
643 + return net.JoinHostPort(raw, "80"), nil
644 + }
645 + if ip := net.ParseIP(raw); ip != nil {
646 + return net.JoinHostPort(raw, "80"), nil
647 + }
648 + return "", fmt.Errorf("invalid target address %q", raw)
649 +}
sdk/helper_test.go
+392 -15
@@ -6,12 +6,15 @@ import (
6 "io"
7 "net"
8 "net/http"
9 + "reflect"
10 "sort"
11 "testing"
12 "time"
13 +
14 + "github.com/gosuda/portal/v2/types"
15 )
16
14 -func TestRunHTTPAppRelayOnly(t *testing.T) {
17 +func TestRunHTTPRelayOnly(t *testing.T) {
18 t.Parallel()
19
20 listener, err := net.Listen("tcp", "127.0.0.1:0")
@@ -36,14 +39,57 @@ func TestRunHTTPAppRelayOnly(t *testing.T) {
39 select {
40 case err := <-errCh:
41 if err != nil {
39 - t.Fatalf("RunHTTPApp() error = %v", err)
42 + t.Fatalf("RunHTTP() error = %v", err)
43 }
44 case <-time.After(3 * time.Second):
42 - t.Fatal("RunHTTPApp() did not exit after context cancellation")
45 + t.Fatal("RunHTTP() did not exit after context cancellation")
46 }
47 }
48
46 -func TestRunHTTPAppLocalAndRelay(t *testing.T) {
49 +func TestExposureRunHTTPLocalOnly(t *testing.T) {
50 + t.Parallel()
51 +
52 + localListener, err := net.Listen("tcp", "127.0.0.1:0")
53 + if err != nil {
54 + t.Fatalf("Listen() error = %v", err)
55 + }
56 + localAddr := localListener.Addr().String()
57 + _ = localListener.Close()
58 +
59 + ctx, cancel := context.WithCancel(context.Background())
60 + defer cancel()
61 +
62 + var exposure *Exposure
63 + errCh := make(chan error, 1)
64 + go func() {
65 + errCh <- exposure.RunHTTP(ctx, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
66 + _, _ = io.WriteString(w, "ok")
67 + }), localAddr)
68 + }()
69 +
70 + waitForHTTP(t, "http://"+localAddr)
71 + cancel()
72 +
73 + select {
74 + case err := <-errCh:
75 + if err != nil {
76 + t.Fatalf("RunHTTP() error = %v", err)
77 + }
78 + case <-time.After(3 * time.Second):
79 + t.Fatal("RunHTTP() did not exit after context cancellation")
80 + }
81 +}
82 +
83 +func TestRunHTTPRequiresRelayOrLocal(t *testing.T) {
84 + t.Parallel()
85 +
86 + err := RunHTTP(context.Background(), nil, http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}), "")
87 + if err == nil {
88 + t.Fatal("RunHTTP() error = nil, want error")
89 + }
90 +}
91 +
92 +func TestRunHTTPLocalAndRelay(t *testing.T) {
93 t.Parallel()
94
95 relayListener, err := net.Listen("tcp", "127.0.0.1:0")
@@ -76,10 +122,96 @@ func TestRunHTTPAppLocalAndRelay(t *testing.T) {
122 select {
123 case err := <-errCh:
124 if err != nil {
79 - t.Fatalf("RunHTTPApp() error = %v", err)
125 + t.Fatalf("RunHTTP() error = %v", err)
126 + }
127 + case <-time.After(3 * time.Second):
128 + t.Fatal("RunHTTP() did not exit after context cancellation")
129 + }
130 +}
131 +
132 +func TestRunHTTPRelayListenerCloseIsNormal(t *testing.T) {
133 + t.Parallel()
134 +
135 + listener, err := net.Listen("tcp", "127.0.0.1:0")
136 + if err != nil {
137 + t.Fatalf("Listen() error = %v", err)
138 + }
139 +
140 + ctx, cancel := context.WithCancel(context.Background())
141 + defer cancel()
142 +
143 + errCh := make(chan error, 1)
144 + go func() {
145 + errCh <- RunHTTP(ctx, listener, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
146 + _, _ = io.WriteString(w, "ok")
147 + }), "")
148 + }()
149 +
150 + waitForHTTP(t, "http://"+listener.Addr().String())
151 +
152 + if err := listener.Close(); err != nil {
153 + t.Fatalf("listener.Close() error = %v", err)
154 + }
155 +
156 + select {
157 + case err := <-errCh:
158 + if err != nil {
159 + t.Fatalf("RunHTTP() error = %v", err)
160 }
161 case <-time.After(3 * time.Second):
82 - t.Fatal("RunHTTPApp() did not exit after context cancellation")
162 + t.Fatal("RunHTTP() did not exit after listener close")
163 + }
164 +}
165 +
166 +func TestRunHTTPRelayListenerCloseKeepsLocalRunning(t *testing.T) {
167 + t.Parallel()
168 +
169 + relayListener, err := net.Listen("tcp", "127.0.0.1:0")
170 + if err != nil {
171 + t.Fatalf("Listen() error = %v", err)
172 + }
173 +
174 + localListener, err := net.Listen("tcp", "127.0.0.1:0")
175 + if err != nil {
176 + t.Fatalf("Listen() error = %v", err)
177 + }
178 + localAddr := localListener.Addr().String()
179 + _ = localListener.Close()
180 +
181 + ctx, cancel := context.WithCancel(context.Background())
182 + defer cancel()
183 +
184 + errCh := make(chan error, 1)
185 + go func() {
186 + errCh <- RunHTTP(ctx, relayListener, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
187 + _, _ = io.WriteString(w, "ok")
188 + }), localAddr)
189 + }()
190 +
191 + waitForHTTP(t, "http://"+relayListener.Addr().String())
192 + waitForHTTP(t, "http://"+localAddr)
193 +
194 + if err := relayListener.Close(); err != nil {
195 + t.Fatalf("relayListener.Close() error = %v", err)
196 + }
197 +
198 + waitForHTTP(t, "http://"+localAddr)
199 +
200 + select {
201 + case err := <-errCh:
202 + t.Fatalf("RunHTTP() exited early with %v", err)
203 + case <-time.After(200 * time.Millisecond):
204 + }
205 +
206 + cancel()
207 +
208 + select {
209 + case err := <-errCh:
210 + if err != nil {
211 + t.Fatalf("RunHTTP() error = %v", err)
212 + }
213 + case <-time.After(3 * time.Second):
214 + t.Fatal("RunHTTP() did not exit after context cancellation")
215 }
216 }
217
@@ -105,12 +237,12 @@ func waitForHTTP(t *testing.T, rawURL string) {
237 func TestMergeListenersRequiresInput(t *testing.T) {
238 t.Parallel()
239
108 - listener, err := MergeListeners()
240 + listener, err := mergeListeners()
241 if err == nil {
110 - t.Fatal("MergeListeners() error = nil, want error")
242 + t.Fatal("mergeListeners() error = nil, want error")
243 }
244 if listener != nil {
113 - t.Fatalf("MergeListeners() listener = %#v, want nil", listener)
245 + t.Fatalf("mergeListeners() listener = %#v, want nil", listener)
246 }
247 }
248
@@ -129,9 +261,9 @@ func TestMergeListenersAcceptsFromAllSources(t *testing.T) {
261 }
262 defer listener2.Close()
263
132 - merged, err := MergeListeners(listener1, listener2)
264 + merged, err := mergeListeners(listener1, listener2)
265 if err != nil {
134 - t.Fatalf("MergeListeners() error = %v", err)
266 + t.Fatalf("mergeListeners() error = %v", err)
267 }
268 defer merged.Close()
269
@@ -180,9 +312,9 @@ func TestMergeListenersContinuesAfterOneSourceCloses(t *testing.T) {
312 }
313 defer listener2.Close()
314
183 - merged, err := MergeListeners(listener1, listener2)
315 + merged, err := mergeListeners(listener1, listener2)
316 if err != nil {
185 - t.Fatalf("MergeListeners() error = %v", err)
317 + t.Fatalf("mergeListeners() error = %v", err)
318 }
319 defer merged.Close()
320
@@ -229,9 +361,9 @@ func TestMergeListenersCloseUnblocksAccept(t *testing.T) {
361 }
362 defer listener.Close()
363
232 - merged, err := MergeListeners(listener)
364 + merged, err := mergeListeners(listener)
365 if err != nil {
234 - t.Fatalf("MergeListeners() error = %v", err)
366 + t.Fatalf("mergeListeners() error = %v", err)
367 }
368
369 errCh := make(chan error, 1)
@@ -253,3 +385,248 @@ func TestMergeListenersCloseUnblocksAccept(t *testing.T) {
385 t.Fatal("Accept() did not unblock after Close()")
386 }
387 }
388 +
389 +func TestMergedListenerAcceptDoesNotReturnQueuedConnAfterClose(t *testing.T) {
390 + t.Parallel()
391 +
392 + conn := &stubConn{}
393 + closed := make(chan struct{})
394 + close(closed)
395 +
396 + merged := &mergedListener{
397 + accepted: make(chan net.Conn, 1),
398 + closed: closed,
399 + }
400 + merged.accepted <- conn
401 +
402 + gotConn, err := merged.Accept()
403 + if gotConn != nil {
404 + t.Fatalf("Accept() conn = %#v, want nil", gotConn)
405 + }
406 + if !errors.Is(err, net.ErrClosed) {
407 + t.Fatalf("Accept() error = %v, want net.ErrClosed", err)
408 + }
409 + if conn.closeCount != 1 {
410 + t.Fatalf("conn close count = %d, want 1", conn.closeCount)
411 + }
412 +}
413 +
414 +func TestNormalizeRelayURLs(t *testing.T) {
415 + t.Parallel()
416 +
417 + got, err := NormalizeRelayURLs([]string{
418 + " localhost:4017 , https://relay.example.com/base/relay?x=1#frag ",
419 + "https://relay.example.com/base",
420 + })
421 + if err != nil {
422 + t.Fatalf("NormalizeRelayURLs() error = %v", err)
423 + }
424 +
425 + want := []string{
426 + "https://localhost:4017",
427 + "https://relay.example.com/base",
428 + }
429 + if !reflect.DeepEqual(got, want) {
430 + t.Fatalf("NormalizeRelayURLs() = %v, want %v", got, want)
431 + }
432 +}
433 +
434 +func TestNormalizeRelayURLRejectsNonHTTPS(t *testing.T) {
435 + t.Parallel()
436 +
437 + _, err := NormalizeRelayURL("http://relay.example.com")
438 + if err == nil {
439 + t.Fatal("NormalizeRelayURL() error = nil, want error")
440 + }
441 +}
442 +
443 +func TestNormalizeTargetAddr(t *testing.T) {
444 + t.Parallel()
445 +
446 + tests := []struct {
447 + name string
448 + input string
449 + want string
450 + }{
451 + {
452 + name: "host only",
453 + input: "localhost",
454 + want: "localhost:80",
455 + },
456 + {
457 + name: "host and port",
458 + input: "127.0.0.1:8080",
459 + want: "127.0.0.1:8080",
460 + },
461 + {
462 + name: "http url",
463 + input: "http://localhost:3000",
464 + want: "localhost:3000",
465 + },
466 + {
467 + name: "https url default port preserved by host parsing",
468 + input: "https://example.com",
469 + want: "example.com:80",
470 + },
471 + {
472 + name: "ipv6 host",
473 + input: "::1",
474 + want: "[::1]:80",
475 + },
476 + }
477 +
478 + for _, tt := range tests {
479 + t.Run(tt.name, func(t *testing.T) {
480 + t.Parallel()
481 +
482 + got, err := NormalizeTargetAddr(tt.input)
483 + if err != nil {
484 + t.Fatalf("NormalizeTargetAddr() error = %v", err)
485 + }
486 + if got != tt.want {
487 + t.Fatalf("NormalizeTargetAddr() = %q, want %q", got, tt.want)
488 + }
489 + })
490 + }
491 +}
492 +
493 +func TestNormalizeTargetAddrRejectsInvalidInput(t *testing.T) {
494 + t.Parallel()
495 +
496 + inputs := []string{
497 + "",
498 + "ftp://example.com",
499 + "http://example.com/path",
500 + "http://example.com?a=1",
501 + "http://example.com#frag",
502 + "host:port:extra",
503 + }
504 +
505 + for _, input := range inputs {
506 + t.Run(input, func(t *testing.T) {
507 + t.Parallel()
508 +
509 + if _, err := NormalizeTargetAddr(input); err == nil {
510 + t.Fatalf("NormalizeTargetAddr(%q) error = nil, want error", input)
511 + }
512 + })
513 + }
514 +}
515 +
516 +func TestExposeNoRelayInputs(t *testing.T) {
517 + t.Parallel()
518 +
519 + exposure, err := Expose(context.Background(), nil, "demo", types.LeaseMetadata{})
520 + if err != nil {
521 + t.Fatalf("Expose() error = %v", err)
522 + }
523 + if exposure != nil {
524 + t.Fatalf("Expose() exposure = %#v, want nil", exposure)
525 + }
526 +}
527 +
528 +func TestExposureAccessorsReturnCopies(t *testing.T) {
529 + t.Parallel()
530 +
531 + exposure := &Exposure{
532 + relays: []exposureRelay{
533 + {
534 + relayURL: "https://relay-1.example.com",
535 + publicURLs: []string{"https://app.example.com"},
536 + },
537 + {
538 + relayURL: "https://relay-2.example.com",
539 + publicURLs: []string{"https://app.example.com"},
540 + },
541 + },
542 + }
543 +
544 + relayURLs := exposure.RelayURLs()
545 + publicURLs := exposure.PublicURLs()
546 +
547 + relayURLs[0] = "changed"
548 + publicURLs[0] = "changed"
549 +
550 + if got, want := exposure.RelayURLs(), []string{"https://relay-1.example.com", "https://relay-2.example.com"}; !reflect.DeepEqual(got, want) {
551 + t.Fatalf("RelayURLs() = %v, want %v", got, want)
552 + }
553 + if got, want := exposure.PublicURLs(), []string{"https://app.example.com"}; !reflect.DeepEqual(got, want) {
554 + t.Fatalf("PublicURLs() = %v, want %v", got, want)
555 + }
556 + if got, want := exposure.relays[0].relayURL, "https://relay-1.example.com"; got != want {
557 + t.Fatalf("relays[0].relayURL = %q, want %q", got, want)
558 + }
559 + if got, want := exposure.relays[0].publicURLs, []string{"https://app.example.com"}; !reflect.DeepEqual(got, want) {
560 + t.Fatalf("relays[0].publicURLs = %v, want %v", got, want)
561 + }
562 +}
563 +
564 +func TestExposureCloseIsIdempotent(t *testing.T) {
565 + t.Parallel()
566 +
567 + listener := &stubListener{}
568 + exposure := &Exposure{listener: listener}
569 +
570 + if err := exposure.Close(); err != nil {
571 + t.Fatalf("Close() error = %v", err)
572 + }
573 + if err := exposure.Close(); err != nil {
574 + t.Fatalf("second Close() error = %v", err)
575 + }
576 + if listener.closeCount != 1 {
577 + t.Fatalf("listener close count = %d, want 1", listener.closeCount)
578 + }
579 +}
580 +
581 +func TestExposureImplementsListener(t *testing.T) {
582 + t.Parallel()
583 +
584 + listener := &stubListener{addr: listenerAddr("merged:test")}
585 + exposure := &Exposure{listener: listener}
586 +
587 + if got := exposure.Addr().String(); got != "merged:test" {
588 + t.Fatalf("Addr().String() = %q, want %q", got, "merged:test")
589 + }
590 +
591 + _, err := exposure.Accept()
592 + if !errors.Is(err, net.ErrClosed) {
593 + t.Fatalf("Accept() error = %v, want net.ErrClosed", err)
594 + }
595 +}
596 +
597 +type stubListener struct {
598 + closeCount int
599 + addr net.Addr
600 +}
601 +
602 +func (l *stubListener) Accept() (net.Conn, error) {
603 + return nil, net.ErrClosed
604 +}
605 +
606 +func (l *stubListener) Close() error {
607 + l.closeCount++
608 + if l.closeCount > 1 {
609 + return errors.New("listener closed more than once")
610 + }
611 + return nil
612 +}
613 +
614 +func (l *stubListener) Addr() net.Addr {
615 + if l.addr != nil {
616 + return l.addr
617 + }
618 + return listenerAddr("stub")
619 +}
620 +
621 +type stubConn struct {
622 + closeCount int
623 +}
624 +
625 +func (c *stubConn) Read(_ []byte) (int, error) { return 0, io.EOF }
626 +func (c *stubConn) Write(b []byte) (int, error) { return len(b), nil }
627 +func (c *stubConn) Close() error { c.closeCount++; return nil }
628 +func (c *stubConn) LocalAddr() net.Addr { return listenerAddr("stub-local") }
629 +func (c *stubConn) RemoteAddr() net.Addr { return listenerAddr("stub-remote") }
630 +func (c *stubConn) SetDeadline(_ time.Time) error { return nil }
631 +func (c *stubConn) SetReadDeadline(_ time.Time) error { return nil }
632 +func (c *stubConn) SetWriteDeadline(_ time.Time) error { return nil }