Migrate to connect-es

etc/buf.web.gen.yaml

@@ -1,15 +1,8 @@
 version: v1
+managed:
+  enabled: true
 plugins:
-  - name: js
+  - plugin: buf.build/connectrpc/es:v1.5.0
     out: dist/js
-    opt:
-      - import_style=commonjs
-  - name: grpc-web
+  - plugin: buf.build/bufbuild/es:v1.10.0
     out: dist/js
-    opt:
-      - import_style=commonjs
-      - mode=grpcwebtext
-  - name: ts
-    out: dist/js
-    opt:
-      - service=grpc-web

rpc/js/package.json

@@ -3,11 +3,11 @@
   "version": "0.2.6",
   "license": "Apache-2.0",
   "dependencies": {
-    "@improbable-eng/grpc-web": "^0.13.0",
-    "google-protobuf": "^3.14.0"
+    "@bufbuild/protobuf": "^1.10.0",
+    "@connectrpc/connect": "^1.5.0",
+    "@connectrpc/connect-web": "^1.5.0"
   },
   "devDependencies": {
-    "@types/google-protobuf": "^3.7.4",
     "@typescript-eslint/eslint-plugin": "^6.10.0",
     "@typescript-eslint/parser": "^6.10.0",
     "@viamrobotics/eslint-config": "^0.2.6",

rpc/js/src/BaseChannel.ts

@@ -1,4 +1,4 @@
-import type { ProtobufMessage } from '@improbable-eng/grpc-web/dist/typings/message';
+import { Message } from '@bufbuild/protobuf';
 import { ConnectionClosedError } from './errors';
 
 export class BaseChannel {
@@ -75,7 +75,7 @@ export class BaseChannel {
     this.closeWithReason(new Error(ev));
   }
 
-  protected write(msg: ProtobufMessage) {
-    this.dataChannel.send(msg.serializeBinary());
+  protected write(msg: Message) {
+    this.dataChannel.send(msg.toBinary());
   }
 }

rpc/js/src/BaseStream.ts

@@ -1,41 +1,30 @@
-import type { grpc } from '@improbable-eng/grpc-web';
 import type { PacketMessage, Stream } from './gen/proto/rpc/webrtc/v1/grpc_pb';
 
 // MaxMessageSize is the maximum size a gRPC message can be.
 let MaxMessageSize = 1 << 25;
 
 export class BaseStream {
-  protected readonly stream: Stream;
-  private readonly onDone: (id: number) => void;
-  protected readonly opts: grpc.TransportOptions;
+  protected readonly grpcStream: Stream;
+  private readonly onDone: (id: bigint) => void;
   protected closed: boolean = false;
   private readonly packetBuf: Array<Uint8Array> = [];
   private packetBufSize = 0;
-  private err: Error | undefined;
 
-  constructor(
-    stream: Stream,
-    onDone: (id: number) => void,
-    opts: grpc.TransportOptions
-  ) {
-    this.stream = stream;
+  constructor(grpcStream: Stream, onDone: (id: bigint) => void) {
+    this.grpcStream = grpcStream;
     this.onDone = onDone;
-    this.opts = opts;
   }
 
-  public closeWithRecvError(err?: Error) {
+  public closeWithRecvError() {
     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);
+    this.onDone(this.grpcStream.id);
   }
 
   protected processPacketMessage(msg: PacketMessage): Uint8Array | undefined {
-    const data = msg.getData_asU8();
+    const { data } = msg;
     if (data.length + this.packetBufSize > MaxMessageSize) {
       this.packetBuf.length = 0;
       this.packetBufSize = 0;
@@ -46,7 +35,7 @@ export class BaseStream {
     }
     this.packetBuf.push(data);
     this.packetBufSize += data.length;
-    if (msg.getEom()) {
+    if (msg.eom) {
       const data = new Uint8Array(this.packetBufSize);
       let position = 0;
       for (let i = 0; i < this.packetBuf.length; i++) {