1 |
|
2 |
|
3 |
|
4 |
|
5 |
|
6 |
|
7 |
|
8 |
|
9 |
|
10 |
|
11 |
|
12 |
|
13 |
|
14 |
|
15 |
|
16 |
|
17 |
|
18 | import { EventEmitter } from 'events';
|
19 | import { Duplex, Readable, Writable } from 'stream';
|
20 |
|
21 | import { StatusObject, MessageContext } from './call-stream';
|
22 | import { Status } from './constants';
|
23 | import { EmitterAugmentation1 } from './events';
|
24 | import { Metadata } from './metadata';
|
25 | import { ObjectReadable, ObjectWritable, WriteCallback } from './object-stream';
|
26 | import { InterceptingCallInterface } from './client-interceptors';
|
27 |
|
28 |
|
29 |
|
30 |
|
31 | export type ServiceError = StatusObject & Error;
|
32 |
|
33 |
|
34 |
|
35 |
|
36 | export type SurfaceCall = {
|
37 | call?: InterceptingCallInterface;
|
38 | cancel(): void;
|
39 | getPeer(): string;
|
40 | } & EmitterAugmentation1<'metadata', Metadata> &
|
41 | EmitterAugmentation1<'status', StatusObject> &
|
42 | EventEmitter;
|
43 |
|
44 |
|
45 |
|
46 |
|
47 | export type ClientUnaryCall = SurfaceCall;
|
48 |
|
49 |
|
50 |
|
51 |
|
52 | export type ClientReadableStream<ResponseType> = {
|
53 | deserialize: (chunk: Buffer) => ResponseType;
|
54 | } & SurfaceCall &
|
55 | ObjectReadable<ResponseType>;
|
56 |
|
57 |
|
58 |
|
59 |
|
60 | export type ClientWritableStream<RequestType> = {
|
61 | serialize: (value: RequestType) => Buffer;
|
62 | } & SurfaceCall &
|
63 | ObjectWritable<RequestType>;
|
64 |
|
65 |
|
66 |
|
67 |
|
68 | export type ClientDuplexStream<
|
69 | RequestType,
|
70 | ResponseType
|
71 | > = ClientWritableStream<RequestType> & ClientReadableStream<ResponseType>;
|
72 |
|
73 |
|
74 |
|
75 |
|
76 |
|
77 |
|
78 |
|
79 | export function callErrorFromStatus(status: StatusObject, callerStack: string): ServiceError {
|
80 | const message = `${status.code} ${Status[status.code]}: ${status.details}`;
|
81 | const error = new Error(message);
|
82 | const stack = `${error.stack}\nfor call at\n${callerStack}`;
|
83 | return Object.assign(new Error(message), status, {stack});
|
84 | }
|
85 |
|
86 | export class ClientUnaryCallImpl
|
87 | extends EventEmitter
|
88 | implements ClientUnaryCall {
|
89 | public call?: InterceptingCallInterface;
|
90 | constructor() {
|
91 | super();
|
92 | }
|
93 |
|
94 | cancel(): void {
|
95 | this.call?.cancelWithStatus(Status.CANCELLED, 'Cancelled on client');
|
96 | }
|
97 |
|
98 | getPeer(): string {
|
99 | return this.call?.getPeer() ?? 'unknown';
|
100 | }
|
101 | }
|
102 |
|
103 | export class ClientReadableStreamImpl<ResponseType>
|
104 | extends Readable
|
105 | implements ClientReadableStream<ResponseType> {
|
106 | public call?: InterceptingCallInterface;
|
107 | constructor(readonly deserialize: (chunk: Buffer) => ResponseType) {
|
108 | super({ objectMode: true });
|
109 | }
|
110 |
|
111 | cancel(): void {
|
112 | this.call?.cancelWithStatus(Status.CANCELLED, 'Cancelled on client');
|
113 | }
|
114 |
|
115 | getPeer(): string {
|
116 | return this.call?.getPeer() ?? 'unknown';
|
117 | }
|
118 |
|
119 | _read(_size: number): void {
|
120 | this.call?.startRead();
|
121 | }
|
122 | }
|
123 |
|
124 | export class ClientWritableStreamImpl<RequestType>
|
125 | extends Writable
|
126 | implements ClientWritableStream<RequestType> {
|
127 | public call?: InterceptingCallInterface;
|
128 | constructor(readonly serialize: (value: RequestType) => Buffer) {
|
129 | super({ objectMode: true });
|
130 | }
|
131 |
|
132 | cancel(): void {
|
133 | this.call?.cancelWithStatus(Status.CANCELLED, 'Cancelled on client');
|
134 | }
|
135 |
|
136 | getPeer(): string {
|
137 | return this.call?.getPeer() ?? 'unknown';
|
138 | }
|
139 |
|
140 | _write(chunk: RequestType, encoding: string, cb: WriteCallback) {
|
141 | const context: MessageContext = {
|
142 | callback: cb,
|
143 | };
|
144 | const flags = Number(encoding);
|
145 | if (!Number.isNaN(flags)) {
|
146 | context.flags = flags;
|
147 | }
|
148 | this.call?.sendMessageWithContext(context, chunk);
|
149 | }
|
150 |
|
151 | _final(cb: Function) {
|
152 | this.call?.halfClose();
|
153 | cb();
|
154 | }
|
155 | }
|
156 |
|
157 | export class ClientDuplexStreamImpl<RequestType, ResponseType>
|
158 | extends Duplex
|
159 | implements ClientDuplexStream<RequestType, ResponseType> {
|
160 | public call?: InterceptingCallInterface;
|
161 | constructor(
|
162 | readonly serialize: (value: RequestType) => Buffer,
|
163 | readonly deserialize: (chunk: Buffer) => ResponseType
|
164 | ) {
|
165 | super({ objectMode: true });
|
166 | }
|
167 |
|
168 | cancel(): void {
|
169 | this.call?.cancelWithStatus(Status.CANCELLED, 'Cancelled on client');
|
170 | }
|
171 |
|
172 | getPeer(): string {
|
173 | return this.call?.getPeer() ?? 'unknown';
|
174 | }
|
175 |
|
176 | _read(_size: number): void {
|
177 | this.call?.startRead();
|
178 | }
|
179 |
|
180 | _write(chunk: RequestType, encoding: string, cb: WriteCallback) {
|
181 | const context: MessageContext = {
|
182 | callback: cb,
|
183 | };
|
184 | const flags = Number(encoding);
|
185 | if (!Number.isNaN(flags)) {
|
186 | context.flags = flags;
|
187 | }
|
188 | this.call?.sendMessageWithContext(context, chunk);
|
189 | }
|
190 |
|
191 | _final(cb: Function) {
|
192 | this.call?.halfClose();
|
193 | cb();
|
194 | }
|
195 | }
|