refactor: combine channel and exec call in gRPC

proto/index.proto

@@ -6,6 +6,5 @@ import "intermediate.proto";
 
 service Runner {
   rpc load(IRStage) returns (Empty);
-  rpc exec(Empty) returns (Empty);
-  rpc channel(stream ChannelMessage) returns (stream ChannelMessage);
+  rpc exec(stream ChannelMessage) returns (stream ChannelMessage);
 }

runners/nodejs/src/interfaces/callback_channel.ts

@@ -31,15 +31,15 @@ export class CallbackChannel<T> implements Writer<T> {
    * Handle incoming messages of the channel by piping them into the callback.
    * @private
    */
-  private handler = new Promise(async () => {
+  private handler = (async () => {
     for await (const data of this.channel) {
-      await this.onWrite(data)
+      await this.onWrite(data);
     }
 
     if (this.onClose != null) {
-      await this.onClose()
+      await this.onClose();
     }
-  });
+  })();
 
   /**
    * Create a new callback channel with a specific callback.
@@ -56,7 +56,7 @@ export class CallbackChannel<T> implements Writer<T> {
   }
 
   close(): Promise<void> {
-    this.channel.close()
+    this.channel.close();
     return Promise.resolve();
   }
 
@@ -65,6 +65,6 @@ export class CallbackChannel<T> implements Writer<T> {
   }
 
   write(data: T) {
-    this.channel.write(data)
+    this.channel.write(data);
   }
 }

runners/nodejs/src/interfaces/channel.ts

@@ -1,7 +1,6 @@
 import { RunnerError } from "../error";
 import { Reader } from "./reader";
 import { Writer } from "./writer";
-import {Log} from "./log";
 
 /**
  * A channel is a communication mechanism that allows for the transfer of values
@@ -20,7 +19,10 @@ export class Channel<T> implements Reader<T>, Writer<T> {
    * Outstanding reads that are waiting for a value to be written to the channel.
    * @private
    */
-  private readonly reads: Array<{ resolve: (value: T) => void, reject: (reason: any) => void }> = [];
+  private readonly reads: Array<{
+    resolve: (value: T) => void;
+    reject: (reason: unknown) => void;
+  }> = [];
 
   /**
    * Whether the channel has been closed or not.
@@ -73,7 +75,7 @@ export class Channel<T> implements Reader<T>, Writer<T> {
     this.closed = true;
 
     for (const promise of this.reads) {
-      promise.reject(RunnerError.channelError())
+      promise.reject(RunnerError.channelError());
     }
   }
 
@@ -91,10 +93,10 @@ export class Channel<T> implements Reader<T>, Writer<T> {
   async *[Symbol.asyncIterator]() {
     while (true) {
       try {
-        yield await this.read()
+        yield await this.read();
       } catch (e) {
         if (this.closed && this.values.length == 0) {
-          break
+          break;
         }
       }
     }

runners/nodejs/src/interfaces/log.ts

@@ -1,9 +1,8 @@
 import { RunnerError } from "../error";
 
-const TIME_PADDING = 15
-const LEVEL_PADDING = 10
-const FILE_PADDING = 39
-const MESSAGE_PADDING = 87
+const TIME_PADDING = 15;
+const LEVEL_PADDING = 10;
+const FILE_PADDING = 39;
 
 enum LogLevel {
   DEBUG,
@@ -64,12 +63,15 @@ export class Log {
    * @private
    */
   private push(value: { level: LogLevel; message: string }): void {
-    const time = new Date().toISOString().slice(11, 22).padEnd(TIME_PADDING, ' ')
-    const level = LogLevel[value.level].padEnd(LEVEL_PADDING, ' ')
-    const caller = this.getCaller().padEnd(FILE_PADDING, ' ')
-    const message = value.message
+    const time = new Date()
+      .toISOString()
+      .slice(11, 22)
+      .padEnd(TIME_PADDING, " ");
+    const level = LogLevel[value.level].padEnd(LEVEL_PADDING, " ");
+    const caller = this.getCaller().padEnd(FILE_PADDING, " ");
+    const message = value.message;
 
-    console.log(`${time}${level}${caller}${message}`)
+    console.log(`${time}${level}${caller}${message}`);
   }
 
   /**