KOL-27 | Added support for listening to notifications (#2)

* Added support for listening to notifications

* Cleanup

Author: renra

README.md

@@ -1,8 +1,8 @@
 # Kolme client
 
-This is a library you can use to interact with kolme-powered applications. 
+This is a library you can use to interact with kolme-powered applications. Things you can do:
 
-Things you can do:
+## Broadcast to your kolme app
 
 ```TypeScript
 import { KolmeClient } from 'kolme-client'
@@ -17,3 +17,28 @@ const block = await client.broadcast(privateKey, [{
   }
 }])
 ```
+
+## Listen to notifications on the kolme's websocket endpoint
+
+```TypeScript
+import { KolmeClient } from 'kolme-client'
+
+const client = new KolmeClient('https://yourkolme.app')
+
+const client.subscribeToNotifications(
+  (message) => {  
+    // Do something interesting with the message
+  },
+  (socketState) => {  
+    // Get updates about the socket state - may be useful for React re-rendering for example
+  }
+)
+
+// Stop listening after a minute
+setTimeout(
+  () => {
+    client.unsubscribeFromNotifications()
+  },
+  60 * 1000
+)
+```

src/HttpApi.ts

@@ -2,6 +2,7 @@ import * as t from 'io-ts';
 import { isRight } from 'fp-ts/Either';
 import { PathReporter } from 'io-ts/lib/PathReporter';
 import waitFor from './waitFor';
+import * as websocket from './websocket';
 
 const processResponse = async <T>(response: Response, decoder: t.Decoder<unknown, T>) : Promise<t.Validation<T>> => {
   if(!response.ok) {
@@ -156,6 +157,20 @@ export const waitForTx = async(base: string, txHash: string) : Promise<Block> =>
   throw new Error(`Kolme block with tx ${txHash} did not appear on the side chain after ${(after - before) / 1000} seconds`)
 }
 
+export type SubscribeToNotificationsInputs = {
+  onOpen: () => void
+  onMessage: (message: MessageEvent) => void
+  onClose: () => void
+  onError: () => void
+}
+
+export const subscribeToNotifications = (base: string, inputs: SubscribeToNotificationsInputs) => {
+  return websocket.open({
+    ...inputs,
+    endpoint: `${base}/notifications`,
+  })  
+}
+
 export class WithBase {
   base: string;
 

src/index.ts

@@ -2,6 +2,9 @@ import { digest } from './crypto';
 import * as secp from '@noble/secp256k1';
 import { Buffer } from 'buffer'
 import * as HttpApi from './HttpApi'
+import waitFor from './waitFor';
+
+export const NOTIFICATIONS_RETRY_TIMEOUT = 3000
 
 export const generatePrivateKey = () : Uint8Array => {
   return secp.utils.randomPrivateKey()
@@ -39,11 +42,19 @@ export const broadcast = async (httpBase: string, signingKey: Uint8Array, messag
   return HttpApi.waitForTx(httpBase, result.txhash)
 }
 
+export type WebSocketState = 
+  | { type: 'subscribed', socket: WebSocket }
+  | { type: 'unsubscribed' }
+
 export class KolmeClient {
   httpBase: string
+  notificationsSocket: WebSocketState
+  lastNotificationsErrorAt: number | undefined
+  onNotificationsSocketStateChange: ((state: number) => void) | undefined
 
   constructor(httpBase: string) {
     this.httpBase = httpBase;
+    this.notificationsSocket = { type: 'unsubscribed' };
   }
 
   generatePrivateKey() {
@@ -57,4 +68,80 @@ export class KolmeClient {
   async broadcast(signingKey: Uint8Array, messages: unknown[]) {
     return broadcast(this.httpBase, signingKey, messages)
   }
+
+  onNotificationsSocketOpen() {
+    if(this.onNotificationsSocketStateChange && this.notificationsSocket.type === 'subscribed') {
+      this.onNotificationsSocketStateChange(this.notificationsSocket.socket.readyState)
+    }
+  }
+
+  async onNotificationsSocketClosed(onMessage: (message: MessageEvent) => void) : Promise<boolean> {
+    switch(this.notificationsSocket.type) {
+      case 'subscribed': {
+        if(this.onNotificationsSocketStateChange) {
+          this.onNotificationsSocketStateChange(this.notificationsSocket.socket.readyState)
+        }
+
+        // Make sure we're not re-trying too often
+        if(this.lastNotificationsErrorAt) {
+          const now = Date.now()
+          const diff = now - this.lastNotificationsErrorAt
+
+          if(diff < NOTIFICATIONS_RETRY_TIMEOUT) {
+            await waitFor(NOTIFICATIONS_RETRY_TIMEOUT - diff)
+          }
+        }    
+
+        this.subscribeToNotifications(onMessage)
+        return true
+      }
+      case 'unsubscribed': {
+        return false;
+      }
+    }
+  }
+
+  async subscribeToNotifications(onMessage: (message: MessageEvent) => void, onReadyStateChange? : (state: number) => void) : Promise<void> {
+    const endpoint = this.httpBase.replace('https', 'wss')
+
+    if(onReadyStateChange) {
+      this.onNotificationsSocketStateChange = onReadyStateChange
+    }
+
+    return new Promise((resolve, reject) => {
+      this.notificationsSocket = {
+        type: 'subscribed',
+        socket: HttpApi.subscribeToNotifications(endpoint, {
+          onMessage,
+          onOpen: () => {
+            this.onNotificationsSocketOpen()
+            resolve()
+          },
+          onClose: () => {
+            this.onNotificationsSocketClosed(onMessage)
+          },
+          onError: () => {
+            this.lastNotificationsErrorAt = Date.now()
+            reject()
+          },
+        })
+      }
+    })
+  }
+
+  unsubscribeFromNotifications(): boolean {
+    this.onNotificationsSocketStateChange = undefined
+    
+    switch(this.notificationsSocket.type) {
+      case 'subscribed': {
+        this.notificationsSocket.socket.close()
+        this.notificationsSocket = { type: 'unsubscribed' }
+
+        return true
+      }
+      case 'unsubscribed': {
+        return false
+      }
+    }
+  }
 }

src/websocket.ts

@@ -0,0 +1,16 @@
+export type OpenProps = {
+  endpoint: string;
+  onOpen: () => void
+  onMessage: (message: MessageEvent) => void
+  onClose: () => void
+}
+
+export const open = ({ endpoint, onOpen, onClose, onMessage } : OpenProps) : WebSocket => {
+  const ws = new WebSocket(endpoint);
+
+  ws.onopen = onOpen
+  ws.onmessage = onMessage
+  ws.onclose = onClose
+
+  return ws
+}