promisify receiveUnaryMessage server-call

packages/grpc-js/src/server-call.ts

@@ -553,13 +553,8 @@ export class Http2ServerCallStream<
     return metadata;
   }
 
-  receiveUnaryMessage(
+  receiveUnaryMessage(encoding: string): Promise<RequestType | void> {
-    encoding: string,
+    return new Promise((resolve, reject) => {
-    next: (
-      err: Partial<ServerStatusResponse> | null,
-      request?: RequestType
-    ) => void
-  ): void {
       const { stream } = this;
 
       let receivedLength = 0;
@@ -569,18 +564,19 @@ export class Http2ServerCallStream<
       const body: Buffer[] = [];
       const limit = this.maxReceiveMessageSize;
 
-    stream.on('data', onData);
+      this.stream.on('data', onData);
-    stream.on('end', onEnd);
+      this.stream.on('end', onEnd);
-    stream.on('error', onEnd);
+      this.stream.on('error', onEnd);
 
-    function onData(chunk: Buffer) {
+      async function onData(chunk: Buffer) {
         receivedLength += chunk.byteLength;
 
         if (limit !== -1 && receivedLength > limit) {
           stream.removeListener('data', onData);
           stream.removeListener('end', onEnd);
           stream.removeListener('error', onEnd);
-        next({
+
+          reject({
             code: Status.RESOURCE_EXHAUSTED,
             details: `Received message larger than max (${receivedLength} vs. ${limit})`,
           });
@@ -590,18 +586,18 @@ export class Http2ServerCallStream<
         body.push(chunk);
       }
 
-    function onEnd(err?: Error) {
+      async function onEnd(err?: Error) {
         stream.removeListener('data', onData);
         stream.removeListener('end', onEnd);
         stream.removeListener('error', onEnd);
 
         if (err !== undefined) {
-        next({ code: Status.INTERNAL, details: err.message });
+          reject({ code: Status.INTERNAL, details: err.message });
           return;
         }
 
         if (receivedLength === 0) {
-        next({
+          reject({
             code: Status.INTERNAL,
             details: 'received empty unary message',
           });
@@ -619,14 +615,15 @@ export class Http2ServerCallStream<
         );
 
         if (Buffer.isBuffer(decompressedMessage)) {
-        call.safeDeserializeMessage(decompressedMessage, next);
+          call.safeDeserializeMessage(decompressedMessage, resolve, reject);
           return;
         }
 
         decompressedMessage.then(
-        decompressed => call.safeDeserializeMessage(decompressed, next),
+          decompressed =>
+            call.safeDeserializeMessage(decompressed, resolve, reject),
           (err: any) =>
-          next(
+            reject(
               err.code
                 ? err
                 : {
@@ -636,19 +633,20 @@ export class Http2ServerCallStream<
             )
         );
       }
+    });
   }
 
   private safeDeserializeMessage(
     buffer: Buffer,
-    next: (
+    resolve: (
-      err: Partial<ServerStatusResponse> | null,
+      value: void | RequestType | PromiseLike<void | RequestType>
-      request?: RequestType
+    ) => void,
-    ) => void
+    reject: (reason: any) => void
   ) {
     try {
-      next(null, this.deserializeMessage(buffer));
+      resolve(this.deserializeMessage(buffer));
     } catch (err) {
-      next({
+      reject({
         details: getErrorMessage(err),
         code: Status.INTERNAL,
       });

packages/grpc-js/src/server.ts

@@ -96,6 +96,7 @@ function getUnimplementedStatusResponse(
   return {
     code: Status.UNIMPLEMENTED,
     details: `The server does not implement the method ${methodName}`,
+    metadata: new Metadata(),
   };
 }
 
@@ -1176,17 +1177,13 @@ export class Server {
   }
 }
 
-function handleUnary<RequestType, ResponseType>(
+async function handleUnary<RequestType, ResponseType>(
   call: Http2ServerCallStream<RequestType, ResponseType>,
   handler: UnaryHandler<RequestType, ResponseType>,
   metadata: Metadata,
   encoding: string
-): void {
+): Promise<void> {
-  call.receiveUnaryMessage(encoding, (err, request) => {
+  const request = await call.receiveUnaryMessage(encoding);
-    if (err) {
-      call.sendError(err);
-      return;
-    }
 
   if (request === undefined || call.cancelled) {
     return;
@@ -1209,7 +1206,6 @@ function handleUnary<RequestType, ResponseType>(
       call.sendUnaryMessage(err, value, trailer, flags);
     }
   );
-  });
 }
 
 function handleClientStreaming<RequestType, ResponseType>(
@@ -1243,17 +1239,13 @@ function handleClientStreaming<RequestType, ResponseType>(
   handler.func(stream, respond);
 }
 
-function handleServerStreaming<RequestType, ResponseType>(
+async function handleServerStreaming<RequestType, ResponseType>(
   call: Http2ServerCallStream<RequestType, ResponseType>,
   handler: ServerStreamingHandler<RequestType, ResponseType>,
   metadata: Metadata,
   encoding: string
-): void {
+): Promise<void> {
-  call.receiveUnaryMessage(encoding, (err, request) => {
+  const request = await call.receiveUnaryMessage(encoding);
-    if (err) {
-      call.sendError(err);
-      return;
-    }
 
   if (request === undefined || call.cancelled) {
     return;
@@ -1267,7 +1259,6 @@ function handleServerStreaming<RequestType, ResponseType>(
   );
 
   handler.func(stream);
-  });
 }
 
 function handleBidiStreaming<RequestType, ResponseType>(