Merge pull request #85 from kaleido-io/subscription

Add more helpers around subscriptions and batch delivery

Author: nguyer

lib/firefly.ts

@@ -82,7 +82,7 @@ import {
   FireFlyDeployContractRequest,
   FireFlyDeployContractResponse,
   FireFlyWebSocketConnectCallback,
-  FireFlyGetOperationOptions,
+  FireFlyGetWithStatus,
 } from './interfaces';
 import { FireFlyWebSocket, FireFlyWebSocketCallback } from './websocket';
 import HttpBase, { mapConfig } from './http';
@@ -207,6 +207,13 @@ export default class FireFly extends HttpBase {
     return this.getMany<FireFlySubscriptionResponse[]>('/subscriptions', filter, options);
   }
 
+  getSubscription(
+    id: string,
+    options?: FireFlyGetWithStatus,
+  ): Promise<FireFlySubscriptionResponse | undefined> {
+    return this.getOne<FireFlySubscriptionResponse>(`/subscriptions/${id}`, options);
+  }
+
   replaceSubscription(
     sub: FireFlySubscriptionRequest,
     options?: FireFlyReplaceOptions,
@@ -223,7 +230,7 @@ export default class FireFly extends HttpBase {
   }
 
   findData(
-    filter?: FireFlyDataFilter,
+    filter?: FireFlyDataFilter | URLSearchParams,
     options?: FireFlyGetOptions,
   ): Promise<FireFlyDataResponse[]> {
     return this.getMany<FireFlyDataResponse[]>(`/data`, filter, options);
@@ -572,7 +579,7 @@ export default class FireFly extends HttpBase {
 
   getOperation(
     id: string,
-    options?: FireFlyGetOperationOptions,
+    options?: FireFlyGetWithStatus,
   ): Promise<FireFlyOperationResponse | undefined> {
     const params = { fetchstatus: options?.fetchstatus };
     return this.getOne<FireFlyOperationResponse>(`/operations/${id}`, options, params);

lib/http.ts

@@ -9,6 +9,7 @@ import {
   FireFlyUpdateOptions,
   FireFlyDeleteOptions,
   FireFlyIdempotencyError,
+  FireFlyGetWithStatus,
 } from './interfaces';
 
 function isSuccess(status: number) {
@@ -18,6 +19,7 @@ function isSuccess(status: number) {
 export function mapConfig(
   options:
     | FireFlyGetOptions
+    | FireFlyGetWithStatus
     | FireFlyUpdateOptions
     | FireFlyReplaceOptions
     | FireFlyCreateOptions
@@ -29,6 +31,7 @@ export function mapConfig(
     ...options?.requestConfig,
     params,
   };
+
   if (options !== undefined) {
     if ('confirm' in options) {
       config.params = {
@@ -42,6 +45,12 @@ export function mapConfig(
         publish: options.publish,
       };
     }
+    if ('fetchstatus' in options) {
+      config.params = {
+        ...config.params,
+        fetchstatus: options.fetchstatus,
+      };
+    }
   }
   return config;
 }

lib/interfaces.ts

@@ -37,6 +37,10 @@ export interface FireFlyCreateOptions extends FireFlyBaseHttpOptions {
   publish?: boolean;
 }
 
+export interface FireFlyGetWithStatus extends FireFlyGetOptions {
+  fetchstatus?: string;
+}
+
 export interface FireFlyOptionsInput {
   host: string;
   namespace?: string;
@@ -129,7 +133,7 @@ export type FireFlySubscriptionRequest =
 
 export type FireFlySubscriptionResponse = Required<
   operations['getSubscriptionByID']['responses']['200']['content']['application/json']
->;
+> & { status?: any };
 export type FireFlyEventResponse = Required<
   operations['getEventByID']['responses']['200']['content']['application/json']
 >;
@@ -181,6 +185,17 @@ export interface FireFlyEventDelivery extends Omit<FireFlyEnrichedEvent, 'type'>
   };
 }
 
+export interface FireFlyEventBatchDelivery {
+  type: 'event_batch';
+  id: string;
+  subscription: {
+    id: string;
+    name: string;
+    namespace: string;
+  };
+  events: FireFlyEventDelivery[];
+}
+
 // Datatypes
 
 export type FireFlyDatatypeFilter = operations['getDatatypes']['parameters']['query'];
@@ -282,10 +297,6 @@ export type FireFlyTokenApprovalResponse = typeof approvals[0];
 
 // Operations + Transactions
 
-export interface FireFlyGetOperationOptions extends FireFlyGetOptions {
-  fetchstatus?: string;
-}
-
 export type FireFlyOperationFilter = operations['getOps']['parameters']['query'];
 export type FireFlyTransactionFilter = operations['getTxns']['parameters']['query'];
 

lib/websocket.ts

@@ -5,6 +5,7 @@ import {
   FireFlyEphemeralSubscription,
   FireFlyWebSocketOptions,
   FireFlyEventDelivery,
+  FireFlyEventBatchDelivery,
 } from './interfaces';
 import Logger from './logger';
 
@@ -19,7 +20,10 @@ function buildEphemeralQueryParams(sub: FireFlyEphemeralSubscription) {
 }
 
 export interface FireFlyWebSocketCallback {
-  (socket: FireFlyWebSocket, data: FireFlyEventDelivery): void | Promise<void>;
+  (
+    socket: FireFlyWebSocket,
+    data: FireFlyEventDelivery | FireFlyEventBatchDelivery,
+  ): void | Promise<void>;
 }
 
 export class FireFlyWebSocket {
@@ -166,7 +170,7 @@ export class FireFlyWebSocket {
     }
   }
 
-  ack(event: FireFlyEventDelivery) {
+  ack(event: FireFlyEventDelivery | FireFlyEventBatchDelivery) {
     if (this.socket !== undefined && event.id !== undefined) {
       this.socket.send(
         JSON.stringify({
@@ -179,7 +183,7 @@ export class FireFlyWebSocket {
   }
 
   async close(wait?: boolean): Promise<void> {
-    const closedPromise = new Promise<void>(resolve => {
+    const closedPromise = new Promise<void>((resolve) => {
       this.closed = resolve;
     });
     this.clearPingTimers();