1/**
2 * Marks encoded worker message fields that should move through `postMessage` as
3 * transfer-list entries.
4 *
5 * Worker messages still pass through schema encoding and structured clone, but
6 * schemas wrapped with `schema` can also report backing resources such as
7 * `ArrayBuffer`, `ImageData.data.buffer`, or `MessagePort` to a `Collector`.
8 * Worker platforms then pass the collected values as the transfer list for the
9 * same `postMessage` call, avoiding copies for large payloads and ports.
10 *
11 * @stability unstable
12 * @since 4.0.0
13 */
14import * as Context from "../Context.ts"
15import * as Effect from "../Effect.ts"
16import { dual } from "../Function.ts"
17import * as Schema from "../Schema.ts"
18import * as SchemaGetter from "../SchemaGetter.ts"
19
20/**
21 * Service for collecting `Transferable` objects while encoding worker messages
22 * so they can be passed to `postMessage` transfer lists.
23 *
24 * @stability unstable
25 * @category services
26 * @since 4.0.0
27 */
28export class Collector extends Context.Service<Collector, {
29 readonly addAll: (
30 _: Iterable<globalThis.Transferable>
31 ) => Effect.Effect<void>
32 readonly addAllUnsafe: (_: Iterable<globalThis.Transferable>) => void
33 readonly read: Effect.Effect<Array<globalThis.Transferable>>
34 readonly readUnsafe: () => Array<globalThis.Transferable>
35 readonly clearUnsafe: () => Array<globalThis.Transferable>
36 readonly clear: Effect.Effect<Array<globalThis.Transferable>>
37}>()("effect/workers/Transferable/Collector") {}
38
39/**
40 * Creates a mutable `Collector` service directly, exposing unsafe synchronous
41 * methods for reading, adding, and clearing collected transferables.
42 *
43 * @stability unstable
44 * @category constructors
45 * @since 4.0.0
46 */
47export const makeCollectorUnsafe = (): Collector["Service"] => {
48 let tranferables: Array<globalThis.Transferable> = []
49 const unsafeAddAll = (transfers: Iterable<globalThis.Transferable>): void => {
50 tranferables.push(...transfers)
51 }
52 const unsafeRead = (): Array<globalThis.Transferable> => tranferables
53 const unsafeClear = (): Array<globalThis.Transferable> => {
54 const prev = tranferables
55 tranferables = []
56 return prev
57 }
58 return Collector.of({
59 addAllUnsafe: unsafeAddAll,
60 addAll: (transferables) => Effect.sync(() => unsafeAddAll(transferables)),
61 readUnsafe: unsafeRead,
62 read: Effect.sync(unsafeRead),
63 clearUnsafe: unsafeClear,
64 clear: Effect.sync(unsafeClear)
65 })
66}
67
68/**
69 * Effect that creates a fresh `Collector` service for accumulating
70 * transferables.
71 *
72 * @stability unstable
73 * @category constructors
74 * @since 4.0.0
75 */
76export const makeCollector: Effect.Effect<Collector["Service"]> = Effect.sync(makeCollectorUnsafe)
77
78/**
79 * Adds transferables to the current `Collector` when one is present in the
80 * context, and does nothing otherwise.
81 *
82 * @stability unstable
83 * @category accessors
84 * @since 4.0.0
85 */
86export const addAll = (
87 tranferables: Iterable<globalThis.Transferable>
88): Effect.Effect<void> =>
89 Effect.contextWith((services) => {
90 const collector = Context.getOrUndefined(services, Collector)
91 if (!collector) return Effect.void
92 collector.addAllUnsafe(tranferables)
93 return Effect.void
94 })
95
96/**
97 * Creates a schema getter that records transferables derived from a value in
98 * the current `Collector` while passing the value through unchanged.
99 *
100 * @stability unstable
101 * @category getters
102 * @since 4.0.0
103 */
104export const getterAddAll = <A>(
105 f: (_: A) => Iterable<globalThis.Transferable>
106): SchemaGetter.Getter<A, A> =>
107 SchemaGetter.transformEffect((e: A) =>
108 Effect.contextWith((services) => {
109 const collector = Context.getOrUndefined(services, Collector)
110 if (!collector) return Effect.succeed(e)
111 collector.addAllUnsafe(f(e))
112 return Effect.succeed(e)
113 })
114 )
115
116/**
117 * Schema wrapper whose encode path can record transferables with a `Collector`
118 * while preserving the wrapped schema's decoded type.
119 *
120 * @stability unstable
121 * @category schemas
122 * @since 4.0.0
123 */
124export interface Transferable<S extends Schema.Top> extends
125 Schema.decodeTo<
126 Schema.toType<S["Rebuild"]>,
127 S["Rebuild"]
128 >
129{}
130
131/**
132 * Wraps a schema so encoding records transferables selected from the encoded
133 * value, enabling worker messages to populate a `postMessage` transfer list.
134 *
135 * @stability unstable
136 * @category schemas
137 * @since 4.0.0
138 */
139export const schema: {
140 <S extends Schema.Top>(
141 f: (_: S["Encoded"]) => Iterable<globalThis.Transferable>
142 ): (self: S) => Transferable<S>
143 <S extends Schema.Top>(
144 self: S,
145 f: (_: S["Encoded"]) => Iterable<globalThis.Transferable>
146 ): Transferable<S>
147} = dual(
148 2,
149 <S extends Schema.Top>(
150 self: S,
151 f: (_: S["Encoded"]) => Iterable<globalThis.Transferable>
152 ): Transferable<S> =>
153 self.annotate({
154 toCodecJson: () => passthroughLink
155 }).pipe(
156 Schema.decode({
157 decode: SchemaGetter.passthrough(),
158 encode: getterAddAll(f)
159 })
160 )
161)
162
163const passthroughLink = Schema.link()(Schema.Any, {
164 decode: SchemaGetter.passthrough(),
165 encode: SchemaGetter.passthrough()
166})
167
168/**
169 * Schema for transferring `ImageData` values with their pixel data buffer.
170 *
171 * @stability unstable
172 * @category schemas
173 * @since 4.0.0
174 */
175export const ImageData: Transferable<Schema.declare<ImageData>> = schema(
176 Schema.Any as any as Schema.declare<globalThis.ImageData>,
177 (_) => [_.data.buffer]
178)
179
180/**
181 * Schema for transferring `MessagePort` values as transferable objects.
182 *
183 * @stability unstable
184 * @category schemas
185 * @since 4.0.0
186 */
187export const MessagePort: Transferable<Schema.declare<MessagePort>> = schema(
188 Schema.Any as any as Schema.declare<MessagePort>,
189 (_) => [_]
190)
191
192/**
193 * Schema for transferring `Uint8Array` values with their backing buffer.
194 *
195 * @stability unstable
196 * @category schemas
197 * @since 4.0.0
198 */
199export const Uint8Array: Transferable<Schema.instanceOf<globalThis.Uint8Array<ArrayBuffer>>> = schema(
200 Schema.Uint8Array as any,
201 (_) => [_.buffer]
202)