stream_js.go raw
1 //go:build js && wasm
2
3 package jsws
4
5 import (
6 "errors"
7 "fmt"
8 "net"
9 "sync"
10 "syscall/js"
11 "time"
12 )
13
14 const maxStreamQueue = 8192
15
16 // StreamConn adapts a browser WebSocket to net.Conn. It sends each Write as
17 // one binary message and reads message payloads as a byte stream.
18 type StreamConn struct {
19 ws js.Value
20
21 mu sync.Mutex
22 queue [][]byte
23 buf []byte
24 closed bool
25 err error
26 notify chan struct{}
27
28 funcs []js.Func
29 local net.Addr
30 remote net.Addr
31 }
32
33 // DialStream opens a binary WebSocket and returns it as a net.Conn.
34 func DialStream(rawURL string) (*StreamConn, error) {
35 ctor := js.Global().Get("WebSocket")
36 if ctor.Type() == js.TypeUndefined {
37 return nil, errors.New("jsws: WebSocket is not available")
38 }
39 c := &StreamConn{
40 notify: make(chan struct{}, 1),
41 local: streamAddr("stream:local"),
42 remote: streamAddr(rawURL),
43 }
44 ws, err := newWebSocket(ctor, rawURL)
45 if err != nil {
46 return nil, err
47 }
48 c.ws = ws
49 ws.Set("binaryType", "arraybuffer")
50
51 opened := make(chan struct{})
52 failed := make(chan error, 1)
53 var openOnce sync.Once
54 onOpen := js.FuncOf(func(this js.Value, args []js.Value) any {
55 openOnce.Do(func() { close(opened) })
56 return nil
57 })
58 onMessage := js.FuncOf(func(this js.Value, args []js.Value) any {
59 if len(args) == 0 {
60 return nil
61 }
62 data := args[0].Get("data")
63 if data.Type() != js.TypeObject {
64 return nil
65 }
66 view := js.Global().Get("Uint8Array").New(data)
67 buf := make([]byte, view.Length())
68 js.CopyBytesToGo(buf, view)
69 c.push(buf)
70 return nil
71 })
72 onClose := js.FuncOf(func(this js.Value, args []js.Value) any {
73 err := errors.New("jsws: stream closed")
74 c.fail(err)
75 select {
76 case failed <- err:
77 default:
78 }
79 return nil
80 })
81 onError := js.FuncOf(func(this js.Value, args []js.Value) any {
82 err := errors.New("jsws: stream error")
83 c.fail(err)
84 select {
85 case failed <- err:
86 default:
87 }
88 return nil
89 })
90 c.funcs = []js.Func{onOpen, onMessage, onClose, onError}
91 ws.Set("onopen", onOpen)
92 ws.Set("onmessage", onMessage)
93 ws.Set("onclose", onClose)
94 ws.Set("onerror", onError)
95
96 select {
97 case <-opened:
98 return c, nil
99 case err := <-failed:
100 return nil, err
101 case <-time.After(15 * time.Second):
102 func() {
103 defer func() { recover() }()
104 ws.Call("close")
105 }()
106 return nil, errors.New("jsws: stream connect timeout")
107 }
108 }
109
110 func (c *StreamConn) push(buf []byte) {
111 c.mu.Lock()
112 if c.closed {
113 c.mu.Unlock()
114 return
115 }
116 if len(c.queue) >= maxStreamQueue {
117 c.closed = true
118 c.err = errors.New("jsws: stream queue overflow")
119 c.mu.Unlock()
120 c.signal()
121 return
122 }
123 c.queue = append(c.queue, buf)
124 c.mu.Unlock()
125 c.signal()
126 }
127
128 func (c *StreamConn) fail(err error) {
129 c.mu.Lock()
130 if !c.closed {
131 c.closed = true
132 c.err = err
133 }
134 c.mu.Unlock()
135 c.signal()
136 }
137
138 func (c *StreamConn) signal() {
139 select {
140 case c.notify <- struct{}{}:
141 default:
142 }
143 }
144
145 // Read implements net.Conn. It waits for WebSocket messages; a close or error
146 // event signals the same channel, so a dead socket does not leave the caller
147 // asleep. The P2P read deadline is advisory here because the browser has no
148 // portable socket timeout; the bridge's idle bound closes the stream.
149 func (c *StreamConn) Read(p []byte) (int, error) {
150 if len(p) == 0 {
151 return 0, nil
152 }
153 for {
154 c.mu.Lock()
155 if len(c.buf) > 0 {
156 n := copy(p, c.buf)
157 c.buf = c.buf[n:]
158 c.mu.Unlock()
159 return n, nil
160 }
161 if len(c.queue) > 0 {
162 c.buf = c.queue[0]
163 c.queue = c.queue[1:]
164 c.mu.Unlock()
165 continue
166 }
167 if c.closed {
168 err := c.err
169 c.mu.Unlock()
170 if err == nil {
171 err = errors.New("jsws: closed")
172 }
173 return 0, err
174 }
175 c.mu.Unlock()
176 <-c.notify
177 }
178 }
179
180 // Write implements net.Conn.
181 func (c *StreamConn) Write(p []byte) (n int, err error) {
182 c.mu.Lock()
183 closed := c.closed
184 c.mu.Unlock()
185 if closed {
186 return 0, errors.New("jsws: closed")
187 }
188 defer func() {
189 if r := recover(); r != nil {
190 n, err = 0, fmt.Errorf("jsws: send: %v", r)
191 }
192 }()
193 arr := js.Global().Get("Uint8Array").New(len(p))
194 js.CopyBytesToJS(arr, p)
195 c.ws.Call("send", arr)
196 return len(p), nil
197 }
198
199 // Close implements net.Conn.
200 func (c *StreamConn) Close() error {
201 c.mu.Lock()
202 already := c.closed
203 c.closed = true
204 c.mu.Unlock()
205 if !already {
206 func() {
207 defer func() { recover() }()
208 c.ws.Call("close")
209 }()
210 }
211 c.signal()
212 return nil
213 }
214
215 // LocalAddr implements net.Conn.
216 func (c *StreamConn) LocalAddr() net.Addr { return c.local }
217
218 // RemoteAddr implements net.Conn.
219 func (c *StreamConn) RemoteAddr() net.Addr { return c.remote }
220
221 // SetDeadline implements net.Conn.
222 func (c *StreamConn) SetDeadline(t time.Time) error { return nil }
223
224 // SetReadDeadline implements net.Conn.
225 func (c *StreamConn) SetReadDeadline(t time.Time) error { return nil }
226
227 // SetWriteDeadline implements net.Conn.
228 func (c *StreamConn) SetWriteDeadline(t time.Time) error { return nil }
229
230 type streamAddr string
231
232 func (a streamAddr) Network() string { return "ws" }
233 func (a streamAddr) String() string { return string(a) }
234