conn_js.go raw
1 //go:build js && wasm
2
3 package jsws
4
5 import (
6 "errors"
7 "fmt"
8 "sync"
9 "syscall/js"
10 "time"
11 )
12
13 // maxQueuedMessages bounds the receive queue. A relay message is small; a
14 // client that stops reading while a bridge streams filters is bounded here
15 // rather than by growing the wasm heap without limit.
16 const maxQueuedMessages = 4096
17
18 // Conn is a browser WebSocket implementing relay.Transport.
19 type Conn struct {
20 ws js.Value
21
22 mu sync.Mutex
23 queue [][]byte
24 closed bool
25 err error
26 notify chan struct{}
27 funcs []js.Func
28 }
29
30 // Dial opens a WebSocket to a ws:// or wss:// URL using the browser's
31 // WebSocket implementation.
32 func Dial(rawURL string) (*Conn, error) {
33 ctor := js.Global().Get("WebSocket")
34 if ctor.Type() == js.TypeUndefined {
35 return nil, errors.New("jsws: WebSocket is not available")
36 }
37 c := &Conn{notify: make(chan struct{}, 1)}
38 opened := make(chan struct{})
39 failed := make(chan error, 1)
40 openOnce := sync.Once{}
41 openedSignal := func() { openOnce.Do(func() { close(opened) }) }
42
43 ws, err := newWebSocket(ctor, rawURL)
44 if err != nil {
45 return nil, err
46 }
47 c.ws = ws
48
49 onOpen := js.FuncOf(func(this js.Value, args []js.Value) any {
50 openedSignal()
51 return nil
52 })
53 onMessage := js.FuncOf(func(this js.Value, args []js.Value) any {
54 if len(args) == 0 {
55 return nil
56 }
57 data := args[0].Get("data")
58 if data.Type() != js.TypeString {
59 return nil
60 }
61 c.push([]byte(data.String()))
62 return nil
63 })
64 onClose := js.FuncOf(func(this js.Value, args []js.Value) any {
65 err := errors.New("jsws: connection closed")
66 c.fail(err)
67 select {
68 case failed <- err:
69 default:
70 }
71 return nil
72 })
73 onError := js.FuncOf(func(this js.Value, args []js.Value) any {
74 err := errors.New("jsws: websocket error")
75 c.fail(err)
76 select {
77 case failed <- err:
78 default:
79 }
80 return nil
81 })
82 c.funcs = []js.Func{onOpen, onMessage, onClose, onError}
83 ws.Set("onopen", onOpen)
84 ws.Set("onmessage", onMessage)
85 ws.Set("onclose", onClose)
86 ws.Set("onerror", onError)
87
88 select {
89 case <-opened:
90 return c, nil
91 case err := <-failed:
92 return nil, err
93 case <-time.After(15 * time.Second):
94 func() {
95 defer func() { recover() }()
96 ws.Call("close")
97 }()
98 return nil, errors.New("jsws: connect timeout")
99 }
100 }
101
102 func newWebSocket(ctor js.Value, rawURL string) (ws js.Value, err error) {
103 defer func() {
104 if r := recover(); r != nil {
105 err = fmt.Errorf("jsws: websocket: %v", r)
106 }
107 }()
108 return ctor.New(rawURL), nil
109 }
110
111 func (c *Conn) push(msg []byte) {
112 c.mu.Lock()
113 if c.closed {
114 c.mu.Unlock()
115 return
116 }
117 if len(c.queue) >= maxQueuedMessages {
118 c.err = errors.New("jsws: receive queue overflow")
119 c.closed = true
120 c.mu.Unlock()
121 select {
122 case c.notify <- struct{}{}:
123 default:
124 }
125 return
126 }
127 c.queue = append(c.queue, msg)
128 c.mu.Unlock()
129 select {
130 case c.notify <- struct{}{}:
131 default:
132 }
133 }
134
135 func (c *Conn) fail(err error) {
136 c.mu.Lock()
137 if !c.closed {
138 c.closed = true
139 c.err = err
140 }
141 c.mu.Unlock()
142 select {
143 case c.notify <- struct{}{}:
144 default:
145 }
146 }
147
148 // ReadMessage returns the next text message.
149 func (c *Conn) ReadMessage() ([]byte, error) {
150 for {
151 c.mu.Lock()
152 if len(c.queue) > 0 {
153 msg := c.queue[0]
154 c.queue = c.queue[1:]
155 c.mu.Unlock()
156 return msg, nil
157 }
158 if c.closed {
159 err := c.err
160 c.mu.Unlock()
161 if err == nil {
162 err = errors.New("jsws: closed")
163 }
164 return nil, err
165 }
166 c.mu.Unlock()
167 <-c.notify
168 }
169 }
170
171 // WriteText sends one text message.
172 func (c *Conn) WriteText(payload []byte) (err error) {
173 c.mu.Lock()
174 closed := c.closed
175 c.mu.Unlock()
176 if closed {
177 return errors.New("jsws: closed")
178 }
179 defer func() {
180 if r := recover(); r != nil {
181 err = fmt.Errorf("jsws: send: %v", r)
182 }
183 }()
184 c.ws.Call("send", string(payload))
185 return nil
186 }
187
188 // Close closes the WebSocket.
189 func (c *Conn) Close() error {
190 c.mu.Lock()
191 already := c.closed
192 c.closed = true
193 c.mu.Unlock()
194 if !already {
195 func() {
196 defer func() { recover() }()
197 c.ws.Call("close")
198 }()
199 }
200 select {
201 case c.notify <- struct{}{}:
202 default:
203 }
204 return nil
205 }
206