Add pub/sub

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@magic8bot/smq",
-  "version": "1.0.2",
+  "version": "1.1.0",
   "description": "Cryptocurrency trading bot",
   "bugs": "https://github.com/magic8bot/smq/issues",
   "license": "MIT",
@@ -19,6 +19,7 @@
   },
   "dependencies": {
     "dotenv": "^8.2.0",
+    "redis": "^3.0.2",
     "rsmq": "^0.12.2"
   },
   "devDependencies": {

src/channels.ts

@@ -3,3 +3,8 @@ export enum Channel {
   SyncTrades = 'SyncTrades',
   Strategy = 'Strategy',
 }
+
+export enum Event {
+  XCH_TRADE = 'XCH_TRADE',
+  XCH_TRADE_PREROLL = 'XCH_TRADE_PREROLL',
+}

src/smq.ts

@@ -1,12 +1,14 @@
-import { Channel } from './channels'
+import { Channel, Event } from './channels'
 import RedisSMQ from 'rsmq'
+import redis from 'redis'
 
 const config = {
   host: process.env.REDIS_HOST,
   port: Number(process.env.REDIS_PORT),
 }
 
-const rsmq = new RedisSMQ(config)
+const client = redis.createClient(config)
+const rsmq = new RedisSMQ({ client })
 
 export class Smq {
   constructor() {
@@ -25,11 +27,38 @@ export class Smq {
     return rsmq.sendMessageAsync({ qname, message: JSON.stringify(message) })
   }
 
-  receiveMessage(qname: Channel) {
-    return rsmq.receiveMessageAsync({ qname })
+  async receiveMessage<T>(qname: Channel): Promise<T | false> {
+    const { message } = (await rsmq.receiveMessageAsync({ qname })) as RedisSMQ.QueueMessage
+
+    return this.decode(message)
   }
 
   deleteMessage(qname: Channel, id: string) {
     return rsmq.deleteMessageAsync({ qname, id })
   }
+
+  subscribe<T>(event: Event, cb: (data: T | false) => null) {
+    const handler = (_, message: string) => {
+      const payload = this.decode<T>(message)
+      cb(payload)
+    }
+
+    client.subscribe(event, handler)
+
+    return () => client.unsubscribe(event, handler)
+  }
+
+  publish(event: Event, message: Record<string, any>) {
+    const payload = JSON.stringify(message)
+
+    client.publish(event, payload)
+  }
+
+  private decode<T>(message: string): T | false {
+    try {
+      return JSON.parse(message)
+    } catch {
+      return false
+    }
+  }
 }