1 //go:build js && wasm
2 3 package msgport
4 5 import (
6 "errors"
7 "sync"
8 "syscall/js"
9 )
10 11 // messagePort is a Port over a browser MessagePort.
12 //
13 // The frames are transferred, not copied: the Uint8Array that carries a frame is created for
14 // that one send and its buffer is handed to the far end, so a filter range crosses the worker
15 // boundary without a second copy of itself.
16 type messagePort struct {
17 port js.Value
18 listener js.Func
19 20 in chan []byte
21 done chan struct{}
22 23 mu sync.Mutex
24 dead bool
25 }
26 27 // NewMessagePort wraps a MessagePort, or any object with addEventListener/postMessage.
28 func NewMessagePort(port js.Value) (Port, error) {
29 if port.IsUndefined() || port.IsNull() {
30 return nil, errors.New("msgport: no port was given")
31 }
32 33 wrapped := &messagePort{
34 port: port,
35 in: make(chan []byte, Buffer),
36 done: make(chan struct{}),
37 }
38 39 wrapped.listener = js.FuncOf(wrapped.onMessage)
40 port.Call("addEventListener", "message", wrapped.listener)
41 42 // A MessagePort received through postMessage starts paused: the queue only flows once
43 // start() is called or an onmessage handler is set. addEventListener alone does not
44 // start it, so a worker that only listens would receive nothing at all — the port stays
45 // paused, no frame ever arrives, and the far end waits with no error anywhere.
46 if start := port.Get("start"); start.Type() == js.TypeFunction {
47 port.Call("start")
48 }
49 50 return wrapped, nil
51 }
52 53 // onMessage copies an arriving frame out of the JS heap.
54 //
55 // A transferred ArrayBuffer arrives as the buffer; a plain Uint8Array arrives as a view. The
56 // Uint8Array constructor copies either, which is the one place a copy is unavoidable: the
57 // bytes have to become Go memory before the frame can be decoded.
58 func (p *messagePort) onMessage(_ js.Value, args []js.Value) any {
59 if len(args) == 0 {
60 return nil
61 }
62 63 frame, err := toBytes(args[0].Get("data"))
64 65 if err != nil {
66 return nil
67 }
68 69 p.mu.Lock()
70 defer p.mu.Unlock()
71 72 if p.dead {
73 return nil
74 }
75 76 select {
77 case p.in <- frame:
78 default:
79 // The protocol cannot drain a full buffer, so a full buffer is a dead channel
80 // rather than backpressure the event loop would have to block for.
81 p.closeLocked()
82 }
83 84 return nil
85 }
86 87 // Send posts one frame, transferring its buffer.
88 func (p *messagePort) Send(frame []byte) error {
89 p.mu.Lock()
90 dead := p.dead
91 p.mu.Unlock()
92 93 if dead {
94 return ErrClosed
95 }
96 97 array := js.Global().Get("Uint8Array").New(len(frame))
98 99 if len(frame) > 0 {
100 _ = js.CopyBytesToJS(array, frame)
101 }
102 103 // The second argument is the transfer list: the buffer moves instead of being copied.
104 p.port.Call("postMessage", array, []any{array.Get("buffer")})
105 106 return nil
107 }
108 109 // Receive is the incoming frames, closed when the port is gone.
110 func (p *messagePort) Receive() <-chan []byte { return p.in }
111 112 // Close detaches the listener and ends the receive side.
113 func (p *messagePort) Close() error {
114 p.mu.Lock()
115 defer p.mu.Unlock()
116 117 p.closeLocked()
118 119 return nil
120 }
121 122 // closeLocked ends the port. It runs under the lock the message handler takes, so a frame can
123 // never be handed to a closed channel.
124 func (p *messagePort) closeLocked() {
125 if p.dead {
126 return
127 }
128 129 p.dead = true
130 131 close(p.in)
132 close(p.done)
133 134 p.port.Call("removeEventListener", "message", p.listener)
135 p.listener.Release()
136 }
137 138 // toBytes reads an ArrayBuffer or a typed-array view into Go memory.
139 func toBytes(value js.Value) ([]byte, error) {
140 if value.IsUndefined() || value.IsNull() {
141 return nil, errors.New("msgport: an empty message")
142 }
143 144 switch value.Type() {
145 case js.TypeObject:
146 default:
147 return nil, errors.New("msgport: a message that is not binary")
148 }
149 150 // new Uint8Array(ArrayBuffer) views the buffer; new Uint8Array(typedArray) copies it.
151 view := js.Global().Get("Uint8Array").New(value)
152 153 out := make([]byte, view.Length())
154 155 if len(out) > 0 {
156 _ = js.CopyBytesToGo(out, view)
157 }
158 159 return out, nil
160 }
161 162 // Done is closed when the port is gone.
163 func (p *messagePort) Done() <-chan struct{} { return p.done }
164