• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1// Copyright 2021 The Pigweed Authors
2//
3// Licensed under the Apache License, Version 2.0 (the "License"); you may not
4// use this file except in compliance with the License. You may obtain a copy of
5// the License at
6//
7//     https://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12// License for the specific language governing permissions and limitations under
13// the License.
14
15import { Status } from 'pigweedjs/pw_status';
16import { Message } from 'google-protobuf';
17
18import {
19  BidirectionalStreamingCall,
20  Call,
21  Callback,
22  ClientStreamingCall,
23  ServerStreamingCall,
24  UnaryCall,
25} from './call';
26import { Channel, Method, MethodType, Service } from './descriptors';
27import { PendingCalls, Rpc } from './rpc_classes';
28
29export function methodStubFactory(
30  rpcs: PendingCalls,
31  channel: Channel,
32  method: Method,
33): MethodStub {
34  switch (method.type) {
35    case MethodType.BIDIRECTIONAL_STREAMING:
36      return new BidirectionalStreamingMethodStub(rpcs, channel, method);
37    case MethodType.CLIENT_STREAMING:
38      return new ClientStreamingMethodStub(rpcs, channel, method);
39    case MethodType.SERVER_STREAMING:
40      return new ServerStreamingMethodStub(rpcs, channel, method);
41    case MethodType.UNARY:
42      return new UnaryMethodStub(rpcs, channel, method);
43  }
44}
45
46export abstract class MethodStub {
47  readonly method: Method;
48  readonly rpcs: PendingCalls;
49  readonly rpc: Rpc;
50  private channel: Channel;
51
52  constructor(rpcs: PendingCalls, channel: Channel, method: Method) {
53    this.method = method;
54    this.rpcs = rpcs;
55    this.channel = channel;
56    this.rpc = new Rpc(channel, method.service, method);
57  }
58
59  get id(): number {
60    return this.method.id;
61  }
62}
63
64export class UnaryMethodStub extends MethodStub {
65  invoke(
66    request: Message,
67    onNext: Callback = () => {
68      // Do nothing.
69    },
70    onCompleted: Callback = () => {
71      // Do nothing.
72    },
73    onError: Callback = () => {
74      // Do nothing.
75    },
76  ): UnaryCall {
77    const call = new UnaryCall(
78      this.rpcs,
79      this.rpc,
80      onNext,
81      onCompleted,
82      onError,
83    );
84    call.invoke(request);
85    return call;
86  }
87
88  open(
89    request: Message,
90    onNext: Callback = () => {
91      // Do nothing.
92    },
93    onCompleted: Callback = () => {
94      // Do nothing.
95    },
96    onError: Callback = () => {
97      // Do nothing.
98    },
99  ): UnaryCall {
100    const call = new UnaryCall(
101      this.rpcs,
102      this.rpc,
103      onNext,
104      onCompleted,
105      onError,
106    );
107    call.invoke(request, true);
108    return call;
109  }
110
111  async call(request: Message, timeout?: number): Promise<[Status, Message]> {
112    return await this.invoke(request).complete(timeout);
113  }
114}
115
116export class ServerStreamingMethodStub extends MethodStub {
117  invoke(
118    request?: Message,
119    onNext: Callback = () => {
120      // Do nothing.
121    },
122    onCompleted: Callback = () => {
123      // Do nothing.
124    },
125    onError: Callback = () => {
126      // Do nothing.
127    },
128  ): ServerStreamingCall {
129    const call = new ServerStreamingCall(
130      this.rpcs,
131      this.rpc,
132      onNext,
133      onCompleted,
134      onError,
135    );
136    call.invoke(request);
137    return call;
138  }
139
140  open(
141    request: Message,
142    onNext: Callback = () => {
143      // Do nothing.
144    },
145    onCompleted: Callback = () => {
146      // Do nothing.
147    },
148    onError: Callback = () => {
149      // Do nothing.
150    },
151  ): UnaryCall {
152    const call = new UnaryCall(
153      this.rpcs,
154      this.rpc,
155      onNext,
156      onCompleted,
157      onError,
158    );
159    call.invoke(request, true);
160    return call;
161  }
162
163  call(request?: Message, timeout?: number): Promise<[Status, Message[]]> {
164    return this.invoke(request).complete(timeout);
165  }
166}
167
168export class ClientStreamingMethodStub extends MethodStub {
169  invoke(
170    onNext: Callback = () => {
171      // Do nothing.
172    },
173    onCompleted: Callback = () => {
174      // Do nothing.
175    },
176    onError: Callback = () => {
177      // Do nothing.
178    },
179  ): ClientStreamingCall {
180    const call = new ClientStreamingCall(
181      this.rpcs,
182      this.rpc,
183      onNext,
184      onCompleted,
185      onError,
186    );
187    call.invoke();
188    return call;
189  }
190
191  open(
192    onNext: Callback = () => {
193      // Do nothing.
194    },
195    onCompleted: Callback = () => {
196      // Do nothing.
197    },
198    onError: Callback = () => {
199      // Do nothing.
200    },
201  ): ClientStreamingCall {
202    const call = new ClientStreamingCall(
203      this.rpcs,
204      this.rpc,
205      onNext,
206      onCompleted,
207      onError,
208    );
209    call.invoke(undefined, true);
210    return call;
211  }
212
213  async call(requests: Array<Message> = [], timeout?: number) {
214    return this.invoke().finishAndWait(requests, timeout);
215  }
216}
217
218export class BidirectionalStreamingMethodStub extends MethodStub {
219  invoke(
220    onNext: Callback = () => {
221      // Do nothing.
222    },
223    onCompleted: Callback = () => {
224      // Do nothing.
225    },
226    onError: Callback = () => {
227      // Do nothing.
228    },
229  ): BidirectionalStreamingCall {
230    const call = new BidirectionalStreamingCall(
231      this.rpcs,
232      this.rpc,
233      onNext,
234      onCompleted,
235      onError,
236    );
237    call.invoke();
238    return call;
239  }
240
241  open(
242    onNext: Callback = () => {
243      // Do nothing.
244    },
245    onCompleted: Callback = () => {
246      // Do nothing.
247    },
248    onError: Callback = () => {
249      // Do nothing.
250    },
251  ): BidirectionalStreamingCall {
252    const call = new BidirectionalStreamingCall(
253      this.rpcs,
254      this.rpc,
255      onNext,
256      onCompleted,
257      onError,
258    );
259    call.invoke(undefined, true);
260    return call;
261  }
262
263  async call(requests: Array<Message> = [], timeout?: number) {
264    return this.invoke().finishAndWait(requests, timeout);
265  }
266}
267