feat(Messaging): Subscribe to the topic and receive payload in the ca…

…llback

example/client.ts

@@ -1,18 +1,22 @@
 import Famcache from '../src';
 
-const cache = new Famcache({
+const client = new Famcache({
   host: 'localhost',
   port: 3577,
 });
 
-cache
+client
   .connect()
   .then(() => {
     console.log('Connected!');
 
-    cache.set('key', '10', 3000)
+    client.subscribe('topic1', (data) => {
+      console.log('topic1 received data: ', data);
+    });
+
+    client.set('key', '10', 3000)
       .then(() => {
-        return cache.get('key');
+        return client.get('key');
       })
       .then((data) => {
         console.log('Received', data);
@@ -21,3 +25,4 @@ cache
   .catch((e) => {
     console.log('Failed to connect');
   });
+

src/famcache.ts

@@ -1,17 +1,19 @@
 import { Socket } from 'net';
 import { randomUUID } from 'crypto';
 import type { ConnectionParams } from './params';
-import type { QueueResolver } from './types';
-import { CacheQuery, get, set, del, publish, unsubscribe } from './transport';
+import type { QueueResolver, SubscribeCallback } from './types';
+import { CacheQuery, get, set, del, publish, unsubscribe, subscribe, Messaging } from './transport';
 
 class Famcache {
   private socket: Socket;
   private params: ConnectionParams;
   private queue: Map<string, QueueResolver>;
+  private listeners: Map<string, SubscribeCallback[]>
 
   constructor(params: ConnectionParams) {
     this.socket = new Socket();
     this.queue = new Map();
+    this.listeners = new Map();
 
     this.params = params;
   }
@@ -23,6 +25,18 @@ class Famcache {
   private listen() {
     this.socket.on('data', (data) => {
       const payload = data.toString();
+
+      if (Messaging.isMessagingEvent(payload)) {
+        const message = Messaging.fromEvent(payload);
+
+        if (!this.listeners.has(message.topic)) {
+          return;
+        }
+
+        this.listeners.get(message.topic)?.forEach((callback) => callback(message.data));
+        return;
+      }
+
       const query = CacheQuery.fromString(payload);
 
       const resolver = this.queue.get(query.id);
@@ -94,11 +108,26 @@ class Famcache {
     this.socket.write(publish(queryId, topic, data));
   }
 
+  subscribe(topic: string, callback: SubscribeCallback) {
+    const queryId = this.genId();
+
+    this.socket.write(subscribe(queryId, topic))
+
+    const listeners = this.listeners.get(topic);
+
+    if (!listeners) {
+      this.listeners.set(topic, [callback]);
+    } else {
+      listeners.push(callback);
+    }
+  }
+
   unsubscribe(topic: string) {
     const queryId = this.genId();
 
     this.socket.write(unsubscribe(queryId, topic));
   }
+
 }
 
 export default Famcache;

src/transport/index.ts

@@ -1,2 +1,3 @@
 export * from './cache-query';
 export * from './commands';
+export * from './messaging';

src/transport/messaging.ts

@@ -0,0 +1,19 @@
+export class Messaging {
+    static isMessagingEvent(event: string): boolean {
+        return event.startsWith("MESSAGE ");
+    }
+
+    static fromEvent(event: string): Messaging {
+        const [_, topic, data] = event.split(' ');
+
+        return new Messaging(topic, data);
+    }
+
+    public topic: string;
+    public data: string;
+
+    constructor(topic: string, data: string) {
+        this.topic = topic;
+        this.data = data;
+    }
+}

src/types.ts

@@ -4,3 +4,6 @@ export type QueueResolver = {
   resolve: (value: Optional<string>) => void;
   reject: (reason: Optional<string>) => void;
 };
+
+
+export type SubscribeCallback = (data: string) => void;