1import { afterEach, assert, describe, it } from "@effect/vitest"
2import { Effect, Exit, Layer, Schema, Stream } from "effect"
3import { SchemaBinary } from "effect/encoding"
4import { HttpRouter } from "effect/http"
5import * as HttpClient from "effect/http/HttpClient"
6import * as HttpClientResponse from "effect/http/HttpClientResponse"
7import { Rpc, RpcClient, RpcGroup, type RpcMessage, RpcSchema, RpcSerialization, RpcServer } from "effect/rpc"
8
9const responseExitSuccess = (requestId: string | number, value: unknown) => ({
10 _tag: "Exit",
11 requestId,
12 exit: {
13 _tag: "Success",
14 value
15 }
16})
17
18const objectPrototype = Object.prototype as Record<string, unknown>
19
20const polluteObjectPrototype = (key: string, value: unknown) => {
21 Object.defineProperty(objectPrototype, key, {
22 configurable: true,
23 value
24 })
25}
26
27const decodeJsonRpcSuccess = () =>
28 RpcSerialization.jsonRpc().makeUnsafe().decode("{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":\"ok\"}")
29
30const expectedJsonRpcSuccess = [{
31 _tag: "Exit",
32 requestId: 1,
33 exit: {
34 _tag: "Success",
35 value: "ok"
36 }
37}]
38
39const assertMaxBufferSizeExceeded = (f: () => unknown, maxBufferSize: number) => {
40 try {
41 f()
42 assert.fail("Expected MaxBufferSizeExceeded")
43 } catch (error) {
44 assert.instanceOf(error, RpcSerialization.MaxBufferSizeExceeded)
45 assert.strictEqual(error.maxBufferSize, maxBufferSize)
46 }
47}
48
49// A hole codec that is observably different from `Schema.toCodecJson`: every
50// hole is JSON-lowered and then written as a JSON string. Anything that still
51// hardcodes `Schema.toCodecJson` fails to round-trip against it.
52const codecForJsonString =
53 (<S extends Schema.Top>(schema: S) =>
54 Schema.fromJsonString(Schema.toCodecJson(schema as any))) as RpcSerialization.CodecFor
55
56const serialization: RpcSerialization.RpcSerialization["Service"] = RpcSerialization.RpcSerialization.of({
57 ...RpcSerialization.ndjson,
58 codecFor: codecForJsonString
59})
60
61const layerSerialization = Layer.succeed(RpcSerialization.RpcSerialization)(serialization)
62
63class EchoError extends Schema.Error<EchoError>("EchoError")({
64 at: Schema.Date
65}) {}
66
67const Rpcs = RpcGroup.make(
68 Rpc.make("Echo", { payload: { value: Schema.String }, success: Schema.String }),
69 Rpc.make("Counts", { payload: { to: Schema.Number }, success: RpcSchema.Stream(Schema.Number, Schema.Never) }),
70 Rpc.make("Fail", { payload: {}, success: Schema.Void, error: EchoError }),
71 Rpc.make("Boom", { payload: {}, success: Schema.Void })
72)
73
74const failedAt = new Date(0)
75
76const Handlers = Rpcs.toLayer({
77 Echo: ({ value }) => Effect.succeed(`${value}!`),
78 Counts: ({ to }) => Stream.range(1, to),
79 Fail: () => Effect.fail(new EchoError({ at: failedAt })),
80 Boom: () => Effect.die("boom")
81})
82
83const Server = RpcServer.layerHttp({
84 group: Rpcs,
85 path: "/rpc",
86 protocol: "http"
87}).pipe(
88 Layer.provide(Handlers),
89 Layer.provide(layerSerialization)
90)
91
92const makeClient = Effect.fnUntraced(function*() {
93 const requests: Array<string> = []
94 const { dispose, handler } = HttpRouter.toWebHandler(Server)
95 yield* Effect.addFinalizer(() => Effect.promise(dispose))
96
97 const httpClient = HttpClient.make((request) => {
98 const raw = (request.body as any).body as Uint8Array | string
99 const text = typeof raw === "string" ? raw : new TextDecoder().decode(raw)
100 requests.push(text)
101 return Effect.map(
102 Effect.promise(() => handler(new Request("http://test/rpc", { method: "POST", body: text }))),
103 (response) => HttpClientResponse.fromWeb(request, response)
104 )
105 })
106
107 const client = yield* RpcClient.make(Rpcs).pipe(
108 Effect.provide(
109 RpcClient.layerProtocolHttp({ url: "http://test/rpc" }).pipe(
110 Layer.provide(layerSerialization),
111 Layer.provide(Layer.succeed(HttpClient.HttpClient)(httpClient))
112 )
113 )
114 )
115
116 return { client, requests } as const
117})
118
119const BinaryServer = RpcServer.layerHttp({
120 group: Rpcs,
121 path: "/rpc",
122 protocol: "http"
123}).pipe(
124 Layer.provide(Handlers),
125 Layer.provide(RpcSerialization.layerSchemaBinary())
126)
127
128const makeBinaryClient = Effect.fnUntraced(function*() {
129 const { dispose, handler } = HttpRouter.toWebHandler(BinaryServer)
130 yield* Effect.addFinalizer(() => Effect.promise(dispose))
131
132 const httpClient = HttpClient.make((request) => {
133 const body = (request.body as any).body as Uint8Array
134 return Effect.map(
135 Effect.promise(() =>
136 handler(new Request("http://test/rpc", { method: "POST", body: new Uint8Array(body).buffer }))
137 ),
138 (response) => HttpClientResponse.fromWeb(request, response)
139 )
140 })
141
142 return yield* RpcClient.make(Rpcs).pipe(
143 Effect.provide(
144 RpcClient.layerProtocolHttp({ url: "http://test/rpc" }).pipe(
145 Layer.provide(RpcSerialization.layerSchemaBinary()),
146 Layer.provide(Layer.succeed(HttpClient.HttpClient)(httpClient))
147 )
148 )
149 )
150})
151
152const uvarint = (value: number): Uint8Array => {
153 const bytes: Array<number> = []
154 while (value >= 0x80) {
155 bytes.push((value & 0x7f) | 0x80)
156 value = Math.floor(value / 0x80)
157 }
158 bytes.push(value)
159 return Uint8Array.from(bytes)
160}
161
162describe("RpcSerialization", () => {
163 for (
164 const [name, serialization] of [
165 ["jsonRpc", RpcSerialization.jsonRpc()],
166 ["ndJsonRpc", RpcSerialization.ndJsonRpc()]
167 ] as const
168 ) {
169 describe(`${name} cause roundtrips`, () => {
170 const failure = { _tag: "Fail", error: { code: -32021, message: "Missing capability" } } as const
171 const defect = { _tag: "Die", defect: { message: "boom" } } as const
172 const interrupt = { _tag: "Interrupt", fiberId: 42 } as const
173
174 it.each(
175 [
176 ["typed failure", [failure]],
177 ["defect", [defect]],
178 ["interruption", [interrupt]],
179 ["mixed causes", [failure, defect, interrupt]],
180 ["empty cause", []]
181 ] as const
182 )("preserves %s", (_name, cause) => {
183 const response: RpcMessage.ResponseExitEncoded = {
184 _tag: "Exit",
185 requestId: 1,
186 exit: { _tag: "Failure", cause }
187 }
188 const encoded = serialization.makeUnsafe().encode(response)
189 assert.isDefined(encoded)
190 assert.deepStrictEqual(serialization.makeUnsafe().decode(encoded!), [response])
191 })
192
193 it("preserves a protocol defect", () => {
194 const response: RpcMessage.ResponseDefectEncoded = {
195 _tag: "Defect",
196 defect: { message: "protocol failed" }
197 }
198 const encoded = serialization.makeUnsafe().encode(response)
199 assert.isDefined(encoded)
200 assert.deepStrictEqual(serialization.makeUnsafe().decode(encoded!), [response])
201 })
202 })
203 }
204
205 it.each(
206 [
207 ["Ack", 0],
208 ["Ack", ""],
209 ["Interrupt", 0],
210 ["Interrupt", ""]
211 ] as const
212 )("jsonRpc preserves %s requestId %j", (_tag, requestId) => {
213 const parser = RpcSerialization.jsonRpc().makeUnsafe()
214 const encoded = JSON.stringify({
215 jsonrpc: "2.0",
216 method: `@effect/rpc/${_tag}`,
217 params: { requestId }
218 })
219
220 assert.deepStrictEqual(parser.decode(encoded), [{ _tag, requestId }])
221 })
222
223 describe("jsonRpc inherited properties", { concurrent: false }, () => {
224 afterEach(() => {
225 delete objectPrototype["method"]
226 delete objectPrototype["error"]
227 delete objectPrototype["chunk"]
228 })
229
230 it("decodes a success response with a clean prototype", () => {
231 assert.deepStrictEqual(decodeJsonRpcSuccess(), expectedJsonRpcSuccess)
232 })
233
234 it("ignores an inherited method", () => {
235 polluteObjectPrototype("method", "attacker.evil")
236 assert.deepStrictEqual(decodeJsonRpcSuccess(), expectedJsonRpcSuccess)
237 })
238
239 it("ignores an inherited defect error", () => {
240 polluteObjectPrototype("error", { _tag: "Defect", data: "pwn" })
241 assert.deepStrictEqual(decodeJsonRpcSuccess(), expectedJsonRpcSuccess)
242 })
243
244 it("ignores an inherited chunk marker", () => {
245 polluteObjectPrototype("chunk", true)
246 assert.deepStrictEqual(decodeJsonRpcSuccess(), expectedJsonRpcSuccess)
247 })
248
249 it("ignores an inherited exit error", () => {
250 polluteObjectPrototype("error", { _tag: "Cause", data: [] })
251 assert.deepStrictEqual(decodeJsonRpcSuccess(), expectedJsonRpcSuccess)
252 })
253 })
254
255 it("json decode keeps array payloads flat", () => {
256 const parser = RpcSerialization.json.makeUnsafe()
257 const decoded = parser.decode("[1,2,3]")
258 assert.strictEqual(decoded.length, 3)
259 assert.deepStrictEqual(decoded, [1, 2, 3])
260 })
261
262 it("json decode wraps non-array payloads", () => {
263 const parser = RpcSerialization.json.makeUnsafe()
264 const decoded = parser.decode("{\"a\":1}")
265 assert.strictEqual(decoded.length, 1)
266 assert.deepStrictEqual(decoded, [{ a: 1 }])
267 })
268
269 it("ndjson fails when an unterminated frame exceeds maxBufferSize", () => {
270 const parser = RpcSerialization.makeNdjson({ maxBufferSize: 4 }).makeUnsafe()
271
272 assert.deepStrictEqual(parser.decode("12"), [])
273 assert.deepStrictEqual(parser.decode("34"), [])
274 assertMaxBufferSizeExceeded(() => parser.decode("5"), 4)
275 })
276
277 it("ndjson allows an unbounded incomplete frame", () => {
278 const parser = RpcSerialization.makeNdjson({ maxBufferSize: "unbounded" }).makeUnsafe()
279
280 assert.deepStrictEqual(parser.decode("x".repeat(1024)), [])
281 })
282
283 it("ndjson decodes a multibyte character split across byte chunks", () => {
284 const parser = RpcSerialization.ndjson.makeUnsafe()
285 const message = { value: "\u20ac" }
286 const encoded = parser.encode(message)
287 assert(typeof encoded === "string")
288 const bytes = new TextEncoder().encode(encoded)
289 const split = bytes.indexOf(0xe2) + 1
290
291 assert.deepStrictEqual(parser.decode(bytes.slice(0, split)), [])
292 assert.deepStrictEqual(parser.decode(bytes.slice(split)), [message])
293 })
294
295 it("ndjson skips lines that are not JSON and keeps decoding", () => {
296 const parser = RpcSerialization.ndjson.makeUnsafe()
297
298 assert.deepStrictEqual(parser.decode("{\"id\":1}\nnot json\n{\"id\":2}\n"), [{ id: 1 }, { id: 2 }])
299 assert.deepStrictEqual(parser.decode("{\"id\":4}\n"), [{ id: 4 }])
300 })
301
302 it.effect("layerNdjsonWith forwards maxBufferSize to its decoder", () =>
303 Effect.gen(function*() {
304 const serialization = yield* RpcSerialization.RpcSerialization
305 const parser = serialization.makeUnsafe()
306
307 assert.deepStrictEqual(parser.decode("12"), [])
308 assert.deepStrictEqual(parser.decode("34"), [])
309 assertMaxBufferSizeExceeded(() => parser.decode("5"), 4)
310 }).pipe(
311 Effect.provide(RpcSerialization.layerNdjsonWith({ maxBufferSize: 4 }))
312 ))
313
314 it("ndJsonRpc forwards maxBufferSize to its ndjson framing parser", () => {
315 const parser = RpcSerialization.ndJsonRpc({ maxBufferSize: 4 }).makeUnsafe()
316
317 assert.deepStrictEqual(parser.decode("12"), [])
318 assert.deepStrictEqual(parser.decode("34"), [])
319 assert.throws(() => parser.decode("5"), RpcSerialization.MaxBufferSizeExceeded)
320 })
321
322 it("jsonRpc encodes a non-batched single response array as an object", () => {
323 const parser = RpcSerialization.jsonRpc().makeUnsafe()
324 const decoded = parser.decode("{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"users.get\"}")
325 assert.strictEqual(decoded.length, 1)
326
327 const encoded = parser.encode([responseExitSuccess("1", { id: 1 })])
328 assert(encoded !== undefined)
329
330 const message = JSON.parse(encoded as string)
331 assert.strictEqual(Array.isArray(message), false)
332 assert.deepStrictEqual(message, {
333 jsonrpc: "2.0",
334 id: "1",
335 result: {
336 id: 1
337 }
338 })
339 })
340
341 it("jsonRpc encodes batched responses as an array", () => {
342 const parser = RpcSerialization.jsonRpc().makeUnsafe()
343 const decoded = parser.decode(
344 "[{\"jsonrpc\":\"2.0\",\"id\":\"1\",\"method\":\"users.get\"},{\"jsonrpc\":\"2.0\",\"id\":2,\"method\":\"users.list\"}]"
345 )
346 assert.strictEqual(decoded.length, 2)
347
348 const encoded = parser.encode([
349 responseExitSuccess("1", "one"),
350 responseExitSuccess(2, "two")
351 ])
352 console.log(encoded)
353 assert(encoded !== undefined)
354
355 const message = JSON.parse(encoded as string)
356 assert.strictEqual(Array.isArray(message), true)
357 assert.deepStrictEqual(message, [
358 {
359 jsonrpc: "2.0",
360 id: "1",
361 result: "one"
362 },
363 {
364 jsonrpc: "2.0",
365 id: 2,
366 result: "two"
367 }
368 ])
369 })
370
371 it("jsonRpc preserves id 0 across decode and encode", () => {
372 const parser = RpcSerialization.jsonRpc().makeUnsafe()
373 const decoded = parser.decode("{\"jsonrpc\":\"2.0\",\"id\":0,\"method\":\"users.get\"}")
374 assert.deepStrictEqual(decoded, [{
375 _tag: "Request",
376 id: 0,
377 tag: "users.get",
378 payload: null,
379 headers: []
380 }])
381
382 const encoded = parser.encode({
383 _tag: "Request",
384 id: 0,
385 tag: "users.get",
386 payload: null,
387 headers: []
388 })
389 assert.strictEqual(
390 encoded,
391 "{\"jsonrpc\":\"2.0\",\"method\":\"users.get\",\"params\":null,\"id\":0}"
392 )
393 })
394
395 it("jsonRpc maps null id to internal notification sentinel", () => {
396 const parser = RpcSerialization.jsonRpc().makeUnsafe()
397 const decoded = parser.decode("{\"jsonrpc\":\"2.0\",\"id\":null,\"method\":\"users.get\"}")
398 assert.deepStrictEqual(decoded, [{
399 _tag: "Request",
400 id: "",
401 tag: "users.get",
402 payload: null,
403 headers: []
404 }])
405 })
406
407 it("jsonRpc skips non-objects in a batch without losing requests", () => {
408 const parser = RpcSerialization.jsonRpc().makeUnsafe()
409
410 assert.deepStrictEqual(parser.decode("null"), [])
411 assert.deepStrictEqual(parser.decode("7"), [])
412 assert.deepStrictEqual(
413 parser.decode(
414 "[null,{\"jsonrpc\":\"2.0\",\"method\":1},{\"jsonrpc\":\"2.0\",\"id\":2,\"method\":\"users.get\"}]"
415 ),
416 [{
417 _tag: "Request",
418 id: "",
419 tag: 1,
420 isNotification: true,
421 payload: null,
422 headers: []
423 }, {
424 _tag: "Request",
425 id: 2,
426 tag: "users.get",
427 payload: null,
428 headers: []
429 }]
430 )
431 })
432
433 it("ndJsonRpc keeps requests that share a chunk with a value that is not a message", () => {
434 const parser = RpcSerialization.ndJsonRpc().makeUnsafe()
435
436 assert.deepStrictEqual(parser.decode("null\n{\"jsonrpc\":\"2.0\",\"id\":2,\"method\":\"users.get\"}\n"), [{
437 _tag: "Request",
438 id: 2,
439 tag: "users.get",
440 payload: null,
441 headers: []
442 }])
443 })
444
445 it("jsonRpc preserves empty string id across decode and encode", () => {
446 const parser = RpcSerialization.jsonRpc().makeUnsafe()
447 const decoded = parser.decode("{\"jsonrpc\":\"2.0\",\"id\":\"\",\"method\":\"users.get\"}")
448 assert.deepStrictEqual(decoded, [{
449 _tag: "Request",
450 id: "",
451 tag: "users.get",
452 payload: null,
453 headers: []
454 }])
455
456 const encoded = parser.encode({
457 _tag: "Request",
458 id: "",
459 tag: "users.get",
460 payload: null,
461 headers: []
462 })
463 assert.strictEqual(
464 encoded,
465 "{\"jsonrpc\":\"2.0\",\"method\":\"users.get\",\"params\":null,\"id\":\"\"}"
466 )
467 })
468
469 it("jsonRpc decodes standard error objects as typed failures", () => {
470 const parser = RpcSerialization.jsonRpc().makeUnsafe()
471 const error = { code: -32021, message: "Missing capability", data: { requiredCapabilities: {} } }
472
473 assert.deepStrictEqual(
474 parser.decode(JSON.stringify({
475 jsonrpc: "2.0",
476 id: 1,
477 error
478 })),
479 [{
480 _tag: "Exit",
481 requestId: 1,
482 exit: {
483 _tag: "Failure",
484 cause: [{ _tag: "Fail", error }]
485 }
486 }]
487 )
488 })
489
490 it("jsonRpc encodes a notification without an id", () => {
491 const parser = RpcSerialization.jsonRpc().makeUnsafe()
492 const decoded = parser.decode("{\"jsonrpc\":\"2.0\",\"method\":\"notifications/message\"}")
493 assert.deepStrictEqual(decoded, [{
494 _tag: "Request",
495 id: "",
496 tag: "notifications/message",
497 payload: null,
498 headers: [],
499 isNotification: true
500 }])
501
502 const encoded = parser.encode({
503 _tag: "Request",
504 id: "",
505 tag: "notifications/message",
506 payload: { level: "info" },
507 headers: [["x-test", "value"]],
508 traceId: "trace",
509 spanId: "span",
510 sampled: true,
511 isNotification: true
512 })
513 assert.strictEqual(
514 encoded,
515 "{\"jsonrpc\":\"2.0\",\"method\":\"notifications/message\",\"params\":{\"level\":\"info\"},\"headers\":[[\"x-test\",\"value\"]],\"traceId\":\"trace\",\"spanId\":\"span\",\"sampled\":true}"
516 )
517 })
518
519 describe("SchemaBinary", () => {
520 it.effect("roundtrips requests and streamed responses over HTTP", () =>
521 Effect.gen(function*() {
522 const client = yield* makeBinaryClient()
523
524 assert.strictEqual(yield* client.Echo({ value: "hi" }), "hi!")
525 assert.deepStrictEqual(yield* Stream.runCollect(client.Counts({ to: 3 })), [1, 2, 3])
526
527 const error = yield* Effect.flip(client.Fail({}))
528 assert.instanceOf(error, EchoError)
529 assert.strictEqual(error.at.getTime(), failedAt.getTime())
530
531 const exit = yield* Effect.exit(client.Boom({}))
532 assert(Exit.isFailure(exit))
533 assert.include(String(exit.cause), "boom")
534 }))
535
536 it.effect("fingerprints envelopes", () =>
537 Effect.gen(function*() {
538 const serialization = yield* RpcSerialization.RpcSerialization
539 const parser = serialization.makeUnsafe()
540 const incompatibleEnvelope = Schema.Struct({
541 _tag: Schema.tag("Request"),
542 id: Schema.Union([Schema.String, Schema.Number]),
543 tag: Schema.String,
544 payload: Schema.Uint8Array,
545 headers: Schema.Array(Schema.Tuple([Schema.String, Schema.String])),
546 added: Schema.optional(Schema.String)
547 })
548 const frame = Schema.encodeSync(SchemaBinary.toCodec(incompatibleEnvelope, { fingerprint: true }))({
549 _tag: "Request",
550 id: 1,
551 tag: "Echo",
552 payload: Uint8Array.of(1),
553 headers: []
554 })
555
556 assert.throws(() => parser.decode(frame), /matching layout fingerprint/)
557 }).pipe(Effect.provide(RpcSerialization.layerSchemaBinary())))
558
559 it.effect("does not fingerprint payloads by default", () =>
560 Effect.gen(function*() {
561 const serialization = yield* RpcSerialization.RpcSerialization
562 const Writer = Schema.Struct({
563 value: Schema.String,
564 added: Schema.optional(Schema.String)
565 })
566 const Reader = Schema.Struct({ value: Schema.String })
567 const encoded = Schema.encodeSync(serialization.codecFor(Writer))({ value: "ok", added: "new" })
568
569 assert.instanceOf(encoded, Uint8Array)
570 assert.deepStrictEqual(Schema.decodeSync(serialization.codecFor(Reader))(encoded), { value: "ok" })
571 }).pipe(Effect.provide(RpcSerialization.layerSchemaBinary())))
572
573 it.effect("layerSchemaBinary fingerprints payloads when enabled", () =>
574 Effect.gen(function*() {
575 const serialization = yield* RpcSerialization.RpcSerialization
576 const Writer = Schema.Struct({ value: Schema.String })
577 const Reader = Schema.Struct({ value: Schema.Number })
578 const encoded = Schema.encodeSync(serialization.codecFor(Writer))({ value: "ok" })
579
580 assert.deepStrictEqual(Schema.decodeSync(serialization.codecFor(Writer))(encoded), { value: "ok" })
581 assert.throws(
582 () => Schema.decodeSync(serialization.codecFor(Reader))(encoded),
583 /matching layout fingerprint/
584 )
585 }).pipe(Effect.provide(RpcSerialization.layerSchemaBinary({ fingerprintPayloads: true }))))
586
587 it.effect("uses fingerprinted envelope framing and the binary content type", () =>
588 Effect.gen(function*() {
589 const serialization = yield* RpcSerialization.RpcSerialization
590 const parser = serialization.makeUnsafe()
591 const request: RpcMessage.RequestEncoded = {
592 _tag: "Request",
593 id: 1,
594 tag: "Echo",
595 payload: Uint8Array.of(1, 2, 3),
596 headers: []
597 }
598 const frame = parser.encode(request)
599
600 assert.strictEqual(serialization.contentType, "application/vnd.effect.rpc+schema-binary")
601 assert.strictEqual(serialization.includesFraming, true)
602 assert.instanceOf(frame, Uint8Array)
603 assert.deepStrictEqual(parser.decode(frame), [request])
604 }).pipe(Effect.provide(RpcSerialization.layerSchemaBinary())))
605
606 it.effect("keeps varied frames decodable across parser replacement", () =>
607 Effect.gen(function*() {
608 const serialization = yield* RpcSerialization.RpcSerialization
609 const encoder = serialization.makeUnsafe()
610 const requests: Array<RpcMessage.RequestEncoded> = Array.from({ length: 200 }, (_, index) => ({
611 _tag: "Request",
612 id: index % 2 === 0 ? index : `request-${index}`,
613 tag: `Echo${index % 7}`,
614 payload: Uint8Array.from({ length: index % 17 }, (_, offset) => (index + offset) & 0xFF),
615 headers: Array.from({ length: index % 4 }, (_, offset) => [`x-${offset}`, `${index}`]),
616 ...(index % 3 === 0
617 ? { traceId: `trace-${index}`, spanId: `span-${index}`, sampled: index % 2 === 0 }
618 : undefined)
619 }))
620
621 for (const request of requests) {
622 const frame = encoder.encode(request)
623 assert.instanceOf(frame, Uint8Array)
624 const split = 1 + request.tag.length % (frame.length - 1)
625 const parser = serialization.makeUnsafe()
626 assert.deepStrictEqual(parser.decode(frame.subarray(0, split)), [])
627 assert.deepStrictEqual(parser.decode(frame.subarray(split)), [request])
628 }
629 }).pipe(Effect.provide(RpcSerialization.layerSchemaBinary())))
630
631 it.effect("owns encoded frames without copying envelope holes", () =>
632 Effect.gen(function*() {
633 const serialization = yield* RpcSerialization.RpcSerialization
634 const encoder = serialization.makeUnsafe()
635 const payload = Uint8Array.of(1, 2, 3)
636 const request: RpcMessage.RequestEncoded = {
637 _tag: "Request",
638 id: 1,
639 tag: "Echo",
640 payload,
641 headers: []
642 }
643 const frame = encoder.encode(request)
644 assert(frame instanceof Uint8Array)
645 const expected = frame.slice()
646
647 payload.fill(9)
648 for (let i = 0; i < 1_000; i++) encoder.encode({ ...request, id: i })
649
650 assert.deepStrictEqual(frame, expected)
651 assert.deepStrictEqual(serialization.makeUnsafe().decode(frame), [{
652 ...request,
653 payload: Uint8Array.of(1, 2, 3)
654 }])
655 }).pipe(Effect.provide(RpcSerialization.layerSchemaBinary())))
656
657 it.effect("defaults maxFrameSize to 16 MiB", () =>
658 Effect.gen(function*() {
659 const serialization = yield* RpcSerialization.RpcSerialization
660 assert.deepStrictEqual(serialization.makeUnsafe().decode(uvarint(16 * 1024 * 1024)), [])
661 assert.throws(
662 () => serialization.makeUnsafe().decode(uvarint(16 * 1024 * 1024 + 1)),
663 /frame within maxFrameSize/
664 )
665 }).pipe(Effect.provide(RpcSerialization.layerSchemaBinary())))
666
667 it.effect("layerSchemaBinary overrides maxFrameSize", () =>
668 Effect.gen(function*() {
669 const serialization = yield* RpcSerialization.RpcSerialization
670 assert.deepStrictEqual(serialization.makeUnsafe().decode(uvarint(4)), [])
671 assert.throws(() => serialization.makeUnsafe().decode(uvarint(5)), /frame within maxFrameSize/)
672 }).pipe(Effect.provide(RpcSerialization.layerSchemaBinary({ maxFrameSize: 4 }))))
673
674 it.effect("layerSchemaBinary allows an unbounded maxFrameSize", () =>
675 Effect.gen(function*() {
676 const serialization = yield* RpcSerialization.RpcSerialization
677 assert.deepStrictEqual(serialization.makeUnsafe().decode(uvarint(16 * 1024 * 1024 + 1)), [])
678 }).pipe(Effect.provide(RpcSerialization.layerSchemaBinary({ maxFrameSize: "unbounded" }))))
679 })
680
681 describe("codecFor", () => {
682 it("text serializations JSON-lower the hole", () => {
683 const encode = Schema.encodeSync(RpcSerialization.json.codecFor(Schema.Date))
684 assert.strictEqual(encode(new Date(0)), "1970-01-01T00:00:00.000Z")
685 assert.strictEqual(RpcSerialization.ndjson.codecFor, RpcSerialization.json.codecFor)
686 assert.strictEqual(RpcSerialization.jsonRpc().codecFor, RpcSerialization.json.codecFor)
687 })
688
689 it.effect("fills the request payload hole with the serialization's codec", () =>
690 Effect.gen(function*() {
691 const { client, requests } = yield* makeClient()
692
693 assert.strictEqual(yield* client.Echo({ value: "hi" }), "hi!")
694
695 assert.strictEqual(requests.length, 1)
696 const frame = JSON.parse(requests[0])
697 assert.strictEqual(typeof frame.payload, "string", "the payload hole must carry the codec's output")
698 assert.deepStrictEqual(JSON.parse(frame.payload), { value: "hi" })
699 }))
700
701 it.effect("fills the stream chunk holes with the serialization's codec", () =>
702 Effect.gen(function*() {
703 const { client } = yield* makeClient()
704
705 assert.deepStrictEqual(yield* Stream.runCollect(client.Counts({ to: 3 })), [1, 2, 3])
706 }))
707
708 it.effect("fills the exit hole with the serialization's codec", () =>
709 Effect.gen(function*() {
710 const { client } = yield* makeClient()
711
712 const error = yield* Effect.flip(client.Fail({}))
713 assert.instanceOf(error, EchoError)
714 assert.strictEqual(error.at.getTime(), failedAt.getTime())
715 }))
716
717 it.effect("fills the defect hole with the serialization's codec", () =>
718 Effect.gen(function*() {
719 const { client } = yield* makeClient()
720
721 const exit = yield* Effect.exit(client.Boom({}))
722 if (Exit.isSuccess(exit)) {
723 return assert.fail("expected the handler defect to reach the client")
724 }
725 assert.include(String(exit.cause), "boom")
726 }))
727 })
728})