Added websocket support

package.json

@@ -21,7 +21,7 @@
     "lint": "eslint . --ext .ts",
     "copy-content-files": "copyfiles -u 1 src/**/*.d.ts src/**/*.proto src/**/*.js build/",
     "compile": "tsc --build",
-    "prebuild": "yarn clean && yarn format && yarn lint && echo Using TypeScript && tsc --version",
+    "prebuild": "yarn clean && yarn lint && echo Using TypeScript && tsc --version",
     "build": "yarn compile && yarn copy-content-files",
     "codegen": "buf generate",
     "dev": "ts-node ./src/dev.ts",
@@ -36,7 +36,8 @@
     "google-protobuf": "^3.20.0",
     "lodash": "^4.17.21",
     "node-cache": "^5.1.2",
-    "rxjs": "^7.5.5"
+    "rxjs": "^7.5.5",
+    "websocket-ts": "^1.1.1"
   },
   "devDependencies": {
     "@types/dotenv": "^8.2.0",
@@ -45,6 +46,7 @@
     "@types/jest": "^27.0.2",
     "@types/js-yaml": "^4.0.1",
     "@types/lodash": "^4.14.168",
+    "@types/node": "16.18.14",
     "@typescript-eslint/eslint-plugin": "^5.8.1",
     "@typescript-eslint/parser": "^5.8.1",
     "copyfiles": "^2.4.1",

src/client/getConsumerClientFactory.js

@@ -0,0 +1,22 @@
+export function getConsumerClientFactory() {
+  const factories = {};
+
+  // todo: can we have both clients on node.js ?
+  // conditionally require respective clients.
+  if (isNodejs()) {
+    // eslint-disable-next-line @typescript-eslint/no-var-requires
+    const grpcClient = require("../streamdbConsumerClient/grpcClient");
+    factories["grpc"] = (endpoint) => new grpcClient.StreamDBConsumerGrpcClient(endpoint);
+  } else {
+    // eslint-disable-next-line @typescript-eslint/no-var-requires
+    const httpClient = require("../streamdbConsumerClient/httpClient");
+    factories["http"] = (endpoint) => new httpClient.StreamDBConsumerHttpClient(endpoint);
+  }
+
+  return factories;
+}
+
+function isNodejs() {
+  return typeof process === "object" &&
+    typeof require === "function";
+}

src/client/proximaStreamClient.ts

@@ -1,4 +1,5 @@
-import { StreamDBConsumerClient } from "../streamdb/consumerClient";
+import { StreamDBConsumerClient } from "../streamdbConsumerClient/consumerClient";
+import { getConsumerClientFactory } from "./getConsumerClientFactory";
 import { PausableStream, SimplePauseController } from "../lib/pausableStream";
 import { Offset, StreamEndpoint, StreamEvent } from "../model";
 import { StreamRegistryClient } from "./streamRegistryClient";
@@ -35,7 +36,7 @@ export class ProximaStreamClient {
         `Can't fetch events for stream ${stream} ${offset.toString()}: no endpoints found`
       );
 
-    const client = this.getStreamConsumerClient(endpoint.uri);
+    const client = this.getStreamConsumerClient(endpoint);
     const events = await client.getEvents(stream, offset, count, direction);
 
     // store last event's offset endpoint to the cache
@@ -79,7 +80,7 @@ export class ProximaStreamClient {
         }
 
         //new StreamDBConsumerClient(endpoint)
-        const client = this.getStreamConsumerClient(endpoint.uri);
+        const client = this.getStreamConsumerClient(endpoint);
         currentEventStream = client.getEventsStream(
           stream,
           currentOffset,
@@ -130,14 +131,30 @@ export class ProximaStreamClient {
     return endpoint;
   }
 
-  private getStreamConsumerClient(endpoint: string) {
+  private getStreamConsumerClient(endpoint: StreamEndpoint) {
+    const createClient = () => {
+      if (endpoint.uri && consumerClientFactory["grpc"]) {
+        return consumerClientFactory.grpc(endpoint.uri);
+      }
+
+      if (endpoint.httpUri && consumerClientFactory["http"]) {
+        return consumerClientFactory.http(endpoint.httpUri);
+      }
+
+      throw new Error(`can't create client for ${JSON.stringify(endpoint)}`);
+    };
+
     return (
-      this.clients[endpoint] ??
-      (this.clients[endpoint] = new StreamDBConsumerClient(endpoint))
+      this.clients[endpoint.uri] ??
+      (this.clients[endpoint.uri] = createClient())
     );
   }
 }
 
 export interface StreamClientOptions {
   registry?: StreamRegistry;
 }
+
+const consumerClientFactory: Partial<
+  Record<"http" | "grpc", (endpoint: string) => StreamDBConsumerClient>
+> = getConsumerClientFactory();

src/client/singleStreamDbRegistry.ts

@@ -2,7 +2,7 @@ import { StreamRegistry } from "./streamRegistry";
 import { Offset, StreamEndpoint, StreamStats } from "../model";
 
 export class SingleStreamDbRegistry implements StreamRegistry {
-  public constructor(private readonly streamDbUrl: string) {}
+  public constructor(private readonly streamDbUrl: string, private readonly streamDbHttpUrl?: string) {}
 
   public async getStreamEndpoints(
     // eslint-disable-next-line @typescript-eslint/no-unused-vars
@@ -14,7 +14,8 @@ export class SingleStreamDbRegistry implements StreamRegistry {
     return [
       new StreamEndpoint(
         this.streamDbUrl,
-        new StreamStats(Offset.zero, undefined, undefined, undefined)
+        new StreamStats(Offset.zero, undefined, undefined, undefined),
+        this.streamDbHttpUrl,
       ),
     ];
   }

src/client/streamRegistryClient.ts

@@ -142,6 +142,7 @@ function parseStreamEndpoint(endpoint: any): StreamEndpoint {
 
   return {
     uri: endpoint.uri,
+    httpUri: endpoint.httpUri,
     stats: {
       start: Offset.fromString(stats.start),
       end: Offset.fromString(stats.end),

src/dev.ts

@@ -13,15 +13,13 @@ async function main() {
   const client = new ProximaStreamClient();
 
   const streamRegistry = new StreamRegistryClient();
-  const endpoints = await streamRegistry.getStreamEndpoints("abc", Offset.zero);
-  console.log("ENDPOINTS", endpoints);
+  const endpoints = await streamRegistry.getStreamEndpoints("proxima.exchange-rates.0_1", Offset.zero);
   const allStreams = await streamRegistry.getStreams();
-  console.dir(allStreams, { depth: 10 });
 
   let currentOffset = Offset.zero;
-  for (let i = 0; i < 10; i++) {
+  for (let i = 0; i < 2; i++) {
     const events = await client.fetchEvents(
-      "proxima.eth-main.blocks.1_0",
+      "proxima.exchange-rates.0_1",
       currentOffset,
       100,
       "next"
@@ -34,7 +32,7 @@ async function main() {
   console.log("starting the stream");
 
   const stream = await client.streamEvents(
-    "proxima.eth-main.blocks.1_0",
+    "proxima.exchange-rates.0_1",
     Offset.zero
   );
   const reader = BufferedStreamReader.fromStream(stream, 10000);
@@ -45,8 +43,8 @@ async function main() {
 
     console.log(
       `doing some stuff with a batch of ${batch.length}. ${batch[
-        batch.length - 1
-      ].offset.toString()}`
+      batch.length - 1
+        ].offset.toString()}`
     );
     // simulate processing
     await sleep(100);

src/model/stream.ts

@@ -47,7 +47,8 @@ export class StreamStats {
 export class StreamEndpoint {
   public constructor(
     public readonly uri: string,
-    public readonly stats: StreamStats
+    public readonly stats: StreamStats,
+    public readonly httpUri?: string,
   ) {}
 }
 

src/streamdbConsumerClient/consumerClient.ts

@@ -0,0 +1,17 @@
+import { StreamEvent, Offset } from "../model";
+import { PausableStream, StreamController } from "../lib/pausableStream";
+
+export interface StreamDBConsumerClient {
+  getEvents(
+    stream: string,
+    offset: Offset,
+    count: number,
+    direction: "next" | "last"
+  ): Promise<StreamEvent[]>;
+
+  getEventsStream(
+    stream: string,
+    offset: Offset,
+    controller?: StreamController
+  ): PausableStream<StreamEvent>;
+}

src/streamdb/consumerClient.ts → src/streamdbConsumerClient/grpcClient.ts

@@ -4,13 +4,14 @@ import {
   Direction,
   StreamStateTransitionsRequest,
 } from "../gen/stream_consumer/v1alpha1/stream_consumer";
-import { offsetToProto, stateTransitionProtoToStreamEvent } from "./converters";
+import { offsetToProto, stateTransitionProtoToStreamEvent } from "../streamdb/converters";
 import * as grpc from "@grpc/grpc-js";
 import { StreamEvent, Offset } from "../model";
 
 import { PausableStream, StreamController } from "../lib/pausableStream";
+import {StreamDBConsumerClient} from "./consumerClient";
 
-export class StreamDBConsumerClient {
+export class StreamDBConsumerGrpcClient implements StreamDBConsumerClient {
   private consumer: StreamConsumerServiceClient;
 
   constructor(uri: string) {

src/streamdbConsumerClient/httpClient.ts

@@ -0,0 +1,74 @@
+import {Offset, StreamEvent} from "../model";
+import {Axios} from "axios";
+import {
+  GetStateTransitionsResponse,
+  StreamStateTransitionsResponse
+} from "../gen/stream_consumer/v1alpha1/stream_consumer";
+import {stateTransitionProtoToStreamEvent} from "../streamdb/converters";
+import {PausableStream, StreamController} from "../lib/pausableStream";
+import {ExponentialBackoff, WebsocketBuilder} from "websocket-ts";
+import {StreamDBConsumerClient} from "./consumerClient";
+
+export class StreamDBConsumerHttpClient implements StreamDBConsumerClient {
+  private readonly client: Axios;
+
+  constructor(private readonly uri: string) {
+    this.client = new Axios({
+      baseURL: "https://" + this.uri,
+      validateStatus: status =>
+        (status >= 200 && status < 300) || status == 404,
+    });
+  }
+
+  public getEvents(
+    stream: string,
+    offset: Offset,
+    count: number,
+    direction: "next" | "last"
+  ): Promise<StreamEvent[]> {
+    return this.client
+      .post(`/api/consumer/${stream}/transitions`,
+        JSON.stringify({
+            offset: offset,
+            count: count,
+            direction: direction.toUpperCase(),
+          }, (key, value) => (typeof value === "bigint" ? value.toString() : value)
+        )
+      ).then(resp =>
+        (JSON.parse(resp.data) as GetStateTransitionsResponse)
+          .stateTransitions
+          .map(stateTransitionProtoToStreamEvent)
+      );
+  }
+
+  public getEventsStream(
+    stream: string,
+    offset: Offset,
+    controller?: StreamController
+  ): PausableStream<StreamEvent> {
+    return PausableStream.create<StreamEvent>((observer, _) => {
+      new WebsocketBuilder(webSocketUriForOffset(this.uri, stream, offset))
+        .onClose(() => observer.complete())
+        .onError((_, ev) => observer.error(ev.type))
+        .onMessage((_, ev) =>
+          (JSON.parse(ev.data)["result"] as StreamStateTransitionsResponse)
+            .stateTransition
+            .map(stateTransitionProtoToStreamEvent)
+            .forEach(transition => observer.next(transition)))
+        .withBackoff(new ExponentialBackoff(100, 7))
+        .build();
+    }, controller);
+  }
+}
+
+function webSocketUriForOffset(endpoint: string, streamId: string, offset: Offset): string {
+  const params = new URLSearchParams({
+    "offset.height": offset.height.toString(),
+    "offset.timestamp.epochMs": offset.timestamp.epochMs.toString(),
+  });
+
+  // creates "&offset.timestamp.parts=parts[0]&offset.timestamp.parts=parts[1]" etc. string
+  const parts = ["", ...offset.timestamp.parts].join("&offset.timestamp.parts=");
+
+  return `wss://${endpoint}/api/consumer/${streamId}/stream?${params.toString()}${parts}`;
+}

tsconfig.json

@@ -12,6 +12,7 @@
     "outDir": "./build",
     "allowSyntheticDefaultImports": true,
     "skipLibCheck": true,
+    "allowJs": true,
     "jsx": "react",
     "strict": true,
     "declaration": true,