RSDK-7046 Add Stream Ticks to Board (#267)

Author: oliviamiller

src/components/board.ts

@@ -1,2 +1,2 @@
-export { type Board, type Duration, PowerMode } from './board/board';
+export { type Board, type Duration, PowerMode, type Tick } from './board/board';
 export { BoardClient } from './board/client';

src/components/board/board.ts

@@ -10,7 +10,11 @@ interface Status {
 type ValueOf<T> = T[keyof T];
 export const { PowerMode } = pb;
 export type PowerMode = ValueOf<typeof pb.PowerMode>;
-
+export interface Tick {
+  pinName: string;
+  high: boolean;
+  time: number;
+}
 export type Duration = PBDuration.AsObject;
 
 /**
@@ -89,6 +93,17 @@ export interface Board extends Resource {
     digitalInterruptName: string,
     extra?: StructType
   ): Promise<number>;
+  /**
+   * Stream digital interrupt ticks on the board.
+   *
+   * @param interrupts - Names of the interrupts to stream.
+   * @param queue - Array to put the ticks in.
+   */
+  streamTicks(
+    interrupts: string[],
+    queue: Tick[],
+    extra?: StructType
+  ): Promise<void>;
   /**
    * Set power mode of the board.
    *

src/components/board/client.test.ts

@@ -0,0 +1,121 @@
+// @vitest-environment happy-dom
+
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import { BoardClient } from './client';
+import { type Tick } from './board';
+import { EventDispatcher } from '../../events';
+import { type ResponseStream } from '../../gen/robot/v1/robot_pb_service';
+import { RobotClient } from '../../robot';
+vi.mock('../../robot');
+import { StreamTicksResponse } from '../../gen/component/board/v1/board_pb';
+
+import { BoardServiceClient } from '../../gen/component/board/v1/board_pb_service';
+vi.mock('../../gen/component/board/v1/board_pb_service');
+
+let board: BoardClient;
+
+export class TestResponseStream<T> extends EventDispatcher {
+  private stream: ResponseStream<any>;
+
+  constructor(stream: ResponseStream<any>) {
+    super();
+    this.stream = stream;
+  }
+
+  override on(
+    type: string,
+    handler: (message: any) => void
+  ): ResponseStream<T> {
+    super.on(type, handler);
+    return this;
+  }
+
+  cancel(): void {
+    this.listeners = {};
+    this.stream.cancel();
+  }
+}
+
+let tickStream: ResponseStream<StreamTicksResponse>;
+let testTickStream: TestResponseStream<StreamTicksResponse> | undefined;
+
+const tickStreamMock = ():
+  | TestResponseStream<StreamTicksResponse>
+  | undefined => {
+  return testTickStream;
+};
+
+beforeEach(() => {
+  testTickStream = new TestResponseStream(tickStream);
+  RobotClient.prototype.createServiceClient = vi
+    .fn()
+    .mockImplementation(() => new BoardServiceClient('board'));
+
+  BoardServiceClient.prototype.streamTicks = vi
+    .fn()
+    .mockImplementation(tickStreamMock);
+  board = new BoardClient(new RobotClient('host'), 'test-board');
+});
+
+afterEach(() => {
+  testTickStream = undefined;
+});
+
+describe('streamTicks tests', () => {
+  it('streamTicks', () => {
+    const ticks: Tick[] = [];
+    board.streamTicks(['1', '2'], ticks);
+
+    const response1 = new StreamTicksResponse();
+    response1.setPinName('1');
+    response1.setHigh(true);
+    response1.setTime(1000);
+    testTickStream?.emit('data', response1);
+
+    const response2 = new StreamTicksResponse();
+    response2.setPinName('2');
+    response2.setHigh(false);
+    response2.setTime(2000);
+    testTickStream?.emit('data', response2);
+
+    expect(ticks.length).toEqual(2);
+
+    const tick1: Tick = ticks[0]!;
+    expect(tick1.pinName).toEqual('1');
+    expect(tick1.high).toBe(true);
+    expect(tick1.time).toEqual(1000);
+
+    const tick2: Tick = ticks[1]!;
+    expect(tick2.pinName).toEqual('2');
+    expect(tick2.high).toBe(false);
+    expect(tick2.time).toEqual(2000);
+  });
+
+  it('end streamTicks with an error', () => {
+    const error = { code: 1, details: 'test', metadata: undefined };
+
+    const ticks: Tick[] = [];
+    const promise1 = board.streamTicks(['1', '2'], ticks);
+
+    testTickStream?.emit('end', undefined);
+    expect(promise1).rejects.toStrictEqual({
+      message: 'Stream ended without a status code',
+    });
+
+    const promise2 = board.streamTicks(['1', '2'], ticks);
+    testTickStream?.emit('end', error);
+    expect(promise2).rejects.toStrictEqual({
+      code: 1,
+      message: 'test',
+      metadata: undefined,
+    });
+
+    const promise3 = board.streamTicks(['1', '2'], ticks);
+    testTickStream?.emit('status', error);
+    expect(promise3).rejects.toStrictEqual({
+      code: 1,
+      message: 'test',
+      metadata: undefined,
+    });
+  });
+});

src/components/board/client.ts

@@ -7,7 +7,7 @@ import type { Options, StructType } from '../../types';
 
 import pb from '../../gen/component/board/v1/board_pb';
 import { promisify, doCommandFromClient } from '../../utils';
-import type { Board, Duration, PowerMode } from './board';
+import type { Board, Duration, PowerMode, Tick } from './board';
 
 /**
  * A gRPC-web client for the Board component.
@@ -219,6 +219,50 @@ export class BoardClient implements Board {
     return response.getValue();
   }
 
+  async streamTicks(interrupts: string[], queue: Tick[], extra = {}) {
+    const request = new pb.StreamTicksRequest();
+    request.setName(this.name);
+    request.setPinNamesList(interrupts);
+    request.setExtra(Struct.fromJavaScript(extra));
+    this.options.requestLogger?.(request);
+    const stream = this.client.streamTicks(request);
+    stream.on('data', (response) => {
+      const tick: Tick = {
+        pinName: response.getPinName(),
+        high: response.getHigh(),
+        time: response.getTime(),
+      };
+      queue.push(tick);
+    });
+
+    return new Promise<void>((resolve, reject) => {
+      stream.on('status', (status) => {
+        if (status.code !== 0) {
+          const error = {
+            message: status.details,
+            code: status.code,
+            metadata: status.metadata,
+          };
+          reject(error);
+        }
+      });
+      stream.on('end', (end) => {
+        if (end === undefined) {
+          const error = { message: 'Stream ended without a status code' };
+          reject(error);
+        } else if (end.code !== 0) {
+          const error = {
+            message: end.details,
+            code: end.code,
+            metadata: end.metadata,
+          };
+          reject(error);
+        }
+        resolve();
+      });
+    });
+  }
+
   async setPowerMode(
     name: string,
     powerMode: PowerMode,