Merge pull request #4 from Famcache/feature/pubsub

Feature: Implemented Famcache Publish/Subscribe protocol

CHANGELOG.md

@@ -1,13 +1,11 @@
 ## [1.4.1](https://github.com/Famcache/node-famcache/compare/v1.4.0...v1.4.1) (2024-05-29)
 
-
 ### Bug Fixes
 
-* **Delete:** Fixed Delete command ([b591f9e](https://github.com/Famcache/node-famcache/commit/b591f9e250ec80202264dd3f676783ff96dac9d4))
+- **Delete:** Fixed Delete command ([b591f9e](https://github.com/Famcache/node-famcache/commit/b591f9e250ec80202264dd3f676783ff96dac9d4))
 
 # [1.4.0](https://github.com/Famcache/node-famcache/compare/v1.3.0...v1.4.0) (2024-05-29)
 
-
 ### Features
 
-* **npm:** Mark package as public ([e64a9a2](https://github.com/Famcache/node-famcache/commit/e64a9a21c62a2e737a38cba7b5a909479fbc9cce))
+- **npm:** Mark package as public ([e64a9a2](https://github.com/Famcache/node-famcache/commit/e64a9a21c62a2e737a38cba7b5a909479fbc9cce))

README.md

@@ -1,102 +1,151 @@
-# Node-famcache
-
-Node-famcache is a Node.js client for Famcache, a caching server written in Go. This client allows you to interact with the Famcache server from your Node.js applications, providing an easy-to-use interface for caching operations.
-
-## Table of Contents
-
-- [Installation](#installation)
-- [Usage](#usage)
-  - [Connecting to the Server](#connecting-to-the-server)
-  - [Basic Operations](#basic-operations)
-    - [Set a Value](#set-a-value)
-    - [Get a Value](#get-a-value)
-    - [Delete a Value](#delete-a-value)
-- [API Reference](#api-reference)
-- [Contributing](#contributing)
-- [License](#license)
-
-## Installation
-
-To install Node-famcache, use npm:
-
-```sh
-npm install @famcache/famcache
-```
-
-
-
-## Usage
-### Connecting to the Server
-
-First, import the module and create a client instance:
-
-```ts
-import Famcache from '@famcache/famcache';
-
-const client = new FamcacheClient({
-    host: 'localhost',
-    port: 3577
-});
-```
-
-### Basic Operations
-
-#### Set a Value
-To store a value in the cache:
-
-```ts
-await client.set('key', 'value', 30000);
-```
-
-#### Get a Value
-To retrieve a value from the cache:
-
-```ts
-const value = await client.get('key');
-```
-
-#### Delete a Value
-To delete a value from the cache:
-
-```ts
-await client.del('key');
-```
-
-## API Reference
-
-### `FamcacheClient`
-
-#### `new FamcacheClient(options)`
-
-Creates a new client instance.
-
-- **options** (object):
-  - **host** (string): The host of the Famcache server.
-  - **port** (number): The port of the Famcache server.
-
-#### `client.set(key, value, ttl?)`
-
-Sets a value in the cache.
-
-- **key** (string): The key under which the value will be stored.
-- **value** (string): The value to store.
-- **ttl** (number): Time to leave (optional)
-
-#### `client.get(key)`
-
-Gets a value from the cache.
-
-- **key** (string): The key of the value to retrieve.
-
-#### `client.delete(key, callback)`
-
-Deletes a value from the cache.
-
-- **key** (string): The key of the value to delete.
-
-
-# Contributing
-Contributions are welcome! Please open an issue or submit a pull request on GitHub.
-
-# License
-Node-famcache is licensed under the MIT License. See the [LICENSE](./LICENSE) file for more details.
+# Node-famcache
+
+Node-famcache is a Node.js client for Famcache, a caching server written in Go. This client allows you to interact with the Famcache server from your Node.js applications, providing an easy-to-use interface for caching operations.
+
+## Table of Contents
+
+- [Installation](#installation)
+- [Usage](#usage)
+  - [Connecting to the Server](#connecting-to-the-server)
+  - [Basic Operations](#basic-operations)
+    - [Set a Value](#set-a-value)
+    - [Get a Value](#get-a-value)
+    - [Delete a Value](#delete-a-value)
+    - [Publish a topic](#publish-a-topic)
+    - [Subscribe to the topic](#subscribe-to-the-topic)
+    - [Unsubscribe from the topic](#unsubscribe-from-the-topic)
+- [API Reference](#api-reference)
+- [Contributing](#contributing)
+- [License](#license)
+
+## Installation
+
+To install Node-famcache, use npm:
+
+```sh
+npm install @famcache/famcache
+```
+
+## Usage
+
+### Connecting to the Server
+
+First, import the module and create a client instance:
+
+```ts
+import Famcache from '@famcache/famcache';
+
+const client = new FamcacheClient({
+  host: 'localhost',
+  port: 3577,
+});
+```
+
+### Basic Operations
+
+#### Set a Value
+
+To store a value in the cache:
+
+```ts
+await client.set('key', 'value', 30000);
+```
+
+#### Get a Value
+
+To retrieve a value from the cache:
+
+```ts
+const value = await client.get('key');
+```
+
+#### Delete a Value
+
+To delete a value from the cache:
+
+```ts
+await client.del('key');
+```
+
+
+#### Publish a topic
+
+To publish data to the topic:
+```ts
+client.publish('topic', 'data');
+```
+
+#### Subscribe to the topic
+
+To subscribe to the topic:
+```ts
+client.subscribe('topic', (data) => {
+  // ...
+});
+```
+
+#### Unsubscribe from the topic
+
+To unsubscribe from the topic:
+```ts
+client.unsubscribe('topic');
+```
+
+## API Reference
+
+### `FamcacheClient`
+
+#### `new FamcacheClient(options)`
+
+Creates a new client instance.
+
+- **options** (object):
+  - **host** (string): The host of the Famcache server.
+  - **port** (number): The port of the Famcache server.
+
+#### `client.set(key, value, ttl?)`
+
+Sets a value in the cache.
+
+- **key** (string): The key under which the value will be stored.
+- **value** (string): The value to store.
+- **ttl** (number): Time to leave (optional)
+
+#### `client.get(key)`
+
+Gets a value from the cache.
+
+- **key** (string): The key of the value to retrieve.
+
+#### `client.delete(key, callback)`
+
+Deletes a value from the cache.
+
+#### `client.publish(topic, data)`
+
+Publishes data to the topic
+
+- **topic** (string): Topic name
+- **data** (string): Payload that will be send to the subscribers
+
+#### `client.subscribe(topic, callback)`
+
+Subscribes to the topic
+
+- **topic** (string): Topic name
+- **callback** (Function): Callback function that will be invoked when message will be received for this topic
+
+
+#### `client.unsubscribe(topic)`
+
+Unsubscribes from the topic
+
+- **topic** (string): Topic name
+
+# Contributing
+
+Contributions are welcome! Please open an issue or submit a pull request on GitHub.
+
+# License
+
+Node-famcache is licensed under the MIT License. See the [LICENSE](./LICENSE) file for more details.

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');
   });
+

release.config.cjs

@@ -8,7 +8,7 @@ module.exports = {
     '@semantic-release/changelog',
     '@semantic-release/release-notes-generator',
     '@semantic-release/npm',
-    "@semantic-release/git",
+    '@semantic-release/git',
     '@semantic-release/github',
   ],
 };

src/famcache.ts

@@ -1,17 +1,28 @@
 import { Socket } from 'net';
 import { randomUUID } from 'crypto';
 import type { ConnectionParams } from './params';
-import type { QueueResolver } from './types';
-import { CacheQuery, get, set, del } 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 +34,20 @@ 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);
@@ -87,6 +112,32 @@ class Famcache {
       this.queue.set(queryId, { resolve: () => resolve(), reject });
     });
   }
+
+  publish(topic: string, data: string) {
+    const queryId = this.genId();
+
+    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;