1/**
2 * RPC schema markers and interruption annotations.
3 *
4 * This module contains the small pieces of schema metadata that the RPC
5 * declaration, client, server, cluster, and reactivity layers share. It marks
6 * streamed responses and annotates interruptions that came from a remote client
7 * closing or cancelling a request.
8 *
9 * @stability unstable
10 * @since 4.0.0
11 */
12import * as Cause from "../Cause.ts"
13import * as Context from "../Context.ts"
14import { constUndefined } from "../Function.ts"
15import * as Option from "../Option.ts"
16import * as Predicate from "../Predicate.ts"
17import * as Schema from "../Schema.ts"
18import type * as SchemaAST from "../SchemaAST.ts"
19import * as Stream_ from "../Stream.ts"
20
21const StreamSchemaTypeId = "~effect/rpc/RpcSchema/StreamSchema"
22
23/**
24 * Returns `true` when a schema is an RPC stream schema created by
25 * `RpcSchema.Stream`.
26 *
27 * @stability unstable
28 * @category guards
29 * @since 4.0.0
30 */
31export function isStreamSchema(schema: Schema.Constraint): schema is Stream<Schema.Top, Schema.Top> {
32 return Predicate.hasProperty(schema, StreamSchemaTypeId)
33}
34
35/**
36 * @internal
37 */
38export function getStreamSchemas(schema: Schema.Constraint): Option.Option<{
39 readonly success: Schema.Top
40 readonly error: Schema.Top
41}> {
42 return isStreamSchema(schema) ?
43 Option.some({
44 success: schema.success,
45 error: schema.error
46 }) :
47 Option.none()
48}
49
50/**
51 * A schema marker for RPC streaming responses, storing the success element
52 * schema and stream error schema used for encoding and decoding stream chunks.
53 *
54 * @stability unstable
55 * @category models
56 * @since 4.0.0
57 */
58export interface Stream<A extends Schema.Constraint, E extends Schema.Constraint> extends
59 Schema.BottomLazy<
60 SchemaAST.Declaration,
61 Stream<A, E>
62 >
63{
64 readonly "Type": Stream_.Stream<A["Type"], E["Type"]>
65 readonly "Encoded": Stream_.Stream<A["Encoded"], E["Encoded"]>
66 readonly "DecodingServices": A["DecodingServices"] | E["DecodingServices"]
67 readonly "EncodingServices": A["EncodingServices"] | E["EncodingServices"]
68 readonly "Rebuild": Stream<A, E>
69 readonly "~type.make.in": Stream_.Stream<A["Type"], E["Type"]>
70 readonly "~type.make": Stream_.Stream<A["Type"], E["Type"]>
71 readonly "Iso": Stream_.Stream<A["Type"], E["Type"]>
72 readonly [StreamSchemaTypeId]: typeof StreamSchemaTypeId
73 readonly success: A
74 readonly error: E
75}
76
77const schema = Schema.declare(Stream_.isStream)
78
79/**
80 * Creates an RPC stream schema from a stream element success schema and stream
81 * error schema.
82 *
83 * @stability unstable
84 * @category constructors
85 * @since 4.0.0
86 */
87export function Stream<A extends Schema.Constraint, E extends Schema.Constraint>(success: A, error: E): Stream<A, E> {
88 return Schema.make(schema.ast, { [StreamSchemaTypeId]: StreamSchemaTypeId, success, error })
89}
90
91/**
92 * Annotation that marks interruptions that originate from an RPC client
93 * abort.
94 *
95 * @stability unstable
96 * @category services
97 * @since 4.0.0
98 */
99export class ClientAbort extends Context.Service<ClientAbort, true>()("effect/rpc/RpcSchema/ClientAbort") {
100 static annotation = this.context(true).pipe(
101 Context.add(Cause.StackTrace, {
102 name: "ClientAbort",
103 stack: constUndefined,
104 parent: undefined
105 })
106 )
107}