Merge in pre connect-es rpc/js (#365)

Author: edaniels

package-lock.json

@@ -46,7 +46,6 @@
   },
   "homepage": "https://github.com/viamrobotics/viam-typescript-sdk#readme",
   "dependencies": {
-    "@viamrobotics/rpc": "^0.2.6",
     "exponential-backoff": "^3.1.1"
   },
   "devDependencies": {

package.json

@@ -1,5 +1,5 @@
 import { grpc } from '@improbable-eng/grpc-web';
-import { dialDirect } from '@viamrobotics/rpc';
+import { dialDirect } from '../rpc';
 
 import {
   AuthenticateRequest,

src/app/viam-transport.ts

@@ -1,6 +1,5 @@
 /* eslint-disable max-classes-per-file */
 import { grpc } from '@improbable-eng/grpc-web';
-import { dialDirect, dialWebRTC, type DialOptions } from '@viamrobotics/rpc';
 import { backOff } from 'exponential-backoff';
 import { Duration } from 'google-protobuf/google/protobuf/duration_pb';
 import { isCredential, type Credentials } from '../app/viam-transport';
@@ -32,6 +31,7 @@ import { SensorsServiceClient } from '../gen/service/sensors/v1/sensors_pb_servi
 import { SLAMServiceClient } from '../gen/service/slam/v1/slam_pb_service';
 import { VisionServiceClient } from '../gen/service/vision/v1/vision_pb_service';
 import { ViamResponseStream } from '../responses';
+import { dialDirect, dialWebRTC, type DialOptions } from '../rpc';
 import { MetadataTransport, encodeResourceName, promisify } from '../utils';
 import GRPCConnectionManager from './grpc-connection-manager';
 import type { Robot, RobotStatusStream } from './robot';

src/robot/client.ts

@@ -1,10 +1,10 @@
 // @vitest-environment happy-dom
 
-import { beforeEach, describe, expect, it, vi } from 'vitest';
-import { ConnectionClosedError } from '@viamrobotics/rpc';
-import { FakeTransportBuilder } from '@improbable-eng/grpc-web-fake-transport';
 import { grpc } from '@improbable-eng/grpc-web';
+import { FakeTransportBuilder } from '@improbable-eng/grpc-web-fake-transport';
+import { beforeEach, describe, expect, it, vi } from 'vitest';
 import { RobotServiceClient } from '../gen/robot/v1/robot_pb_service';
+import { ConnectionClosedError } from '../rpc';
 vi.mock('../gen/robot/v1/robot_pb_service');
 
 import SessionManager from './session-manager';

src/robot/session-manager.test.ts

@@ -1,10 +1,10 @@
-import { ConnectionClosedError } from '@viamrobotics/rpc';
 import { grpc } from '@improbable-eng/grpc-web';
+import robotApi from '../gen/robot/v1/robot_pb';
 import {
   RobotServiceClient,
   type ServiceError,
 } from '../gen/robot/v1/robot_pb_service';
-import robotApi from '../gen/robot/v1/robot_pb';
+import { ConnectionClosedError } from '../rpc';
 import SessionTransport from './session-transport';
 
 const timeoutBlob = new Blob(

src/robot/session-manager.ts

@@ -1,5 +1,5 @@
-import { GRPCError } from '@viamrobotics/rpc';
 import { grpc } from '@improbable-eng/grpc-web';
+import { GRPCError } from '../rpc';
 import type SessionManager from './session-manager';
 
 export default class SessionTransport implements grpc.Transport {

src/robot/session-transport.ts

@@ -0,0 +1,81 @@
+import type { ProtobufMessage } from '@improbable-eng/grpc-web/dist/typings/message';
+import { ConnectionClosedError } from './connection-closed-error';
+
+export class BaseChannel {
+  public readonly ready: Promise<unknown>;
+
+  private readonly peerConn: RTCPeerConnection;
+  private readonly dataChannel: RTCDataChannel;
+  private pResolve: ((value: unknown) => void) | undefined;
+  private pReject: ((reason?: unknown) => void) | undefined;
+
+  private closed = false;
+  private closedReason: Error | undefined;
+
+  protected maxDataChannelSize = 65_535;
+
+  constructor(peerConn: RTCPeerConnection, dataChannel: RTCDataChannel) {
+    this.peerConn = peerConn;
+    this.dataChannel = dataChannel;
+
+    this.ready = new Promise<unknown>((resolve, reject) => {
+      this.pResolve = resolve;
+      this.pReject = reject;
+    });
+
+    dataChannel.addEventListener('open', () => this.onChannelOpen());
+    dataChannel.addEventListener('close', () => this.onChannelClose());
+    dataChannel.addEventListener('error', (ev) => {
+      this.onChannelError(ev);
+    });
+
+    peerConn.addEventListener('iceconnectionstatechange', () => {
+      const state = peerConn.iceConnectionState;
+      if (
+        !(state === 'failed' || state === 'disconnected' || state === 'closed')
+      ) {
+        return;
+      }
+      this.pReject?.(new Error(`ICE connection failed with state: ${state}`));
+    });
+  }
+
+  public close() {
+    this.closeWithReason(undefined);
+  }
+
+  public isClosed() {
+    return this.closed;
+  }
+
+  public isClosedReason() {
+    return this.closedReason;
+  }
+
+  protected closeWithReason(err?: Error) {
+    if (this.closed) {
+      return;
+    }
+    this.closed = true;
+    this.closedReason = err;
+    this.pReject?.(err);
+    this.peerConn.close();
+  }
+
+  private onChannelOpen() {
+    this.pResolve?.(undefined);
+  }
+
+  private onChannelClose() {
+    this.closeWithReason(new ConnectionClosedError('data channel closed'));
+  }
+
+  private onChannelError(ev: any) {
+    console.error('channel error', ev);
+    this.closeWithReason(new Error(ev));
+  }
+
+  protected write(msg: ProtobufMessage) {
+    this.dataChannel.send(msg.serializeBinary());
+  }
+}

src/rpc/base-channel.ts

@@ -0,0 +1,62 @@
+import type { grpc } from '@improbable-eng/grpc-web';
+import type { PacketMessage, Stream } from '../gen/proto/rpc/webrtc/v1/grpc_pb';
+
+// MaxMessageSize (2^25) is the maximum size a gRPC message can be.
+const MaxMessageSize = 33_554_432;
+
+export class BaseStream {
+  protected readonly stream: Stream;
+  private readonly onDone: (id: number) => void;
+  protected readonly opts: grpc.TransportOptions;
+  protected closed = false;
+  private readonly packetBuf: Uint8Array[] = [];
+  private packetBufSize = 0;
+  private err: Error | undefined;
+
+  constructor(
+    stream: Stream,
+    onDone: (id: number) => void,
+    opts: grpc.TransportOptions
+  ) {
+    this.stream = stream;
+    this.onDone = onDone;
+    this.opts = opts;
+  }
+
+  public closeWithRecvError(err?: Error) {
+    if (this.closed) {
+      return;
+    }
+    this.closed = true;
+    this.err = err;
+    this.onDone(this.stream.getId());
+    // pretty sure passing the error does nothing.
+    this.opts.onEnd(this.err);
+  }
+
+  protected processPacketMessage(msg: PacketMessage): Uint8Array | undefined {
+    const data = msg.getData_asU8();
+    if (data.length + this.packetBufSize > MaxMessageSize) {
+      this.packetBuf.length = 0;
+      this.packetBufSize = 0;
+      console.error(
+        `message size larger than max ${MaxMessageSize}; discarding`
+      );
+      return undefined;
+    }
+    this.packetBuf.push(data);
+    this.packetBufSize += data.length;
+    if (msg.getEom()) {
+      const pktData = new Uint8Array(this.packetBufSize);
+      let position = 0;
+      for (const partialData of this.packetBuf) {
+        pktData.set(partialData, position);
+        position += partialData.length;
+      }
+      this.packetBuf.length = 0;
+      this.packetBufSize = 0;
+      return pktData;
+    }
+    return undefined;
+  }
+}