feathersjs
diff --git a/‎README.md‎
Lines changed: 45 additions & 35 deletions b/‎README.md‎
Lines changed: 45 additions & 35 deletions
diff --git a/‎lib/adapters/nats.js‎
Lines changed: 46 additions & 16 deletions b/‎lib/adapters/nats.js‎
Lines changed: 46 additions & 16 deletions
@@ -46,9 +46,11 @@ const sync = require('feathers-sync');
 
 const app = feathers();
 
-app.configure(sync({
-  uri: 'redis://localhost:6379'
-}));
+app.configure(
+  sync({
+    uri: 'redis://localhost:6379',
+  })
+);
 app.use('/todos', todoService);
 ```
 
@@ -78,13 +80,13 @@ app.service('messages').hooks({
   after: {
     create(context) {
       // Don't synchronize if more than 1000 items were created at once
-      if(context.result.length > 1000) {
+      if (context.result.length > 1000) {
         context[SYNC] = false;
       }
 
       return context;
-    }
-  }
+    },
+  },
 });
 ```
 
@@ -94,18 +96,24 @@ app.service('messages').hooks({
 
 ```js
 // Configure Redis
-app.configure(sync({
-  uri: 'redis://localhost:6379'
-}));
-
-app.configure(sync.redis({
-  db: redisInstance
-}));
+app.configure(
+  sync({
+    uri: 'redis://localhost:6379',
+  })
+);
+
+app.configure(
+  sync.redis({
+    db: redisInstance,
+  })
+);
 
 // Configure Redis using an existing redisClient
-app.configure(sync.redis({
-  redisClient: redisClient
-}))
+app.configure(
+  sync.redis({
+    redisClient: redisClient,
+  })
+);
 ```
 
 ### Redis
@@ -125,9 +133,9 @@ app.configure(sync.redis({
 
 ### NATS
 
-- `uri` - The connection string (must start with `nats://`)
-- `key` (default: `feathers-sync`) - The name exchange where sync messages will be published
-
+- `uri` - The connection string (example `nats://`)
+- `key` (default: `feathers-sync`) - The name of subject where sync messages will be published
+- `natsConnectionOptions` - NATS [connection options](https://github.com/nats-io/nats.js#Connection-Options)
 
 ## How it works
 
@@ -145,15 +153,17 @@ Event data are serialized and deserialized using `JSON.stringify` and `JSON.pars
 
 ```js
 // BSON can serialize / deserialize `Date` values.
-const bson = require('bson')
-
-app.configure(sync({
-  uri: 'redis://localhost:6379',
-  // Replies will be sent to callbacks as Buffers instead of Strings for bson.deserialize to work.
-  redisOptions: { return_buffers: true },
-  serialize: bson.serialize,
-  deserialize: bson.deserialize,
-}));
+const bson = require('bson');
+
+app.configure(
+  sync({
+    uri: 'redis://localhost:6379',
+    // Replies will be sent to callbacks as Buffers instead of Strings for bson.deserialize to work.
+    redisOptions: { return_buffers: true },
+    serialize: bson.serialize,
+    deserialize: bson.deserialize,
+  })
+);
 ```
 
 > `Redis` and `AMQP` can support binary serialization / deserialization (i.e. `Buffer` data). `NATS` currently does not support custom serialization / deserialization/
@@ -171,14 +181,14 @@ const myMessagingService = {
 
   subscribe(callback) {
     // subscribe to message queue and emit data
-  }
-}
+  },
+};
 
-module.exports = config => {
+module.exports = (config) => {
   // If adapter supports configurable serializer / deserializer (defaults to `JSON.stringfy` / `JSON.parse`)
   const { deserialize, serialize } = config;
 
-  return app => {
+  return (app) => {
     app.configure(core);
     app.sync = {
       type: 'custom',
@@ -187,16 +197,16 @@ module.exports = config => {
         // reject on connection error
       }),
       serialize,
-      deserialize
+      deserialize,
     };
 
     // Sent every time a service
-    app.on('sync-out', data => {
+    app.on('sync-out', (data) => {
       // Publish `data` to the message queue
       myMessagingService.publish(data);
     });
 
-    myMessagingService.subscribe(data => {
+    myMessagingService.subscribe((data) => {
       // Send the synchronization event to the application
       app.emit('sync-in', data);
     });
 
@@ -1,34 +1,64 @@
-const nats = require('nats');
+const { connect, StringCodec } = require('nats');
 const debug = require('debug')('feathers-sync:nats');
 const core = require('../core');
 
-module.exports = config => {
+module.exports = (config) => {
   debug(`Setting up NATS connection ${config.uri}`);
 
-  return app => {
-    const { key, serialize, deserialize } = config;
-    const natsClient = nats.connect({ json: true, url: config.uri, ...config.natsConnectionOptions });
+  return (app) => {
+    const {
+      uri,
+      key: subject,
+      serialize,
+      deserialize,
+      natsConnectionOptions
+    } = config;
+
+    // Setting up nats connection with unique servers list
+    const natsClient = connect({
+      ...natsConnectionOptions,
+      servers: [
+        ...new Set([
+          uri,
+          ...(natsConnectionOptions.servers
+            ? natsConnectionOptions.servers
+            : [])
+        ])
+      ]
+    });
+    const stringCodec = StringCodec();
 
     app.configure(core);
     app.sync = {
       type: 'nats',
       serialize,
       deserialize,
       ready: new Promise((resolve, reject) => {
-        natsClient.once('connect', resolve);
-        natsClient.once('error', reject);
+        natsClient
+          .then((connection) => {
+            const sub = connection.subscribe(subject);
+            // listening events and resolving connection
+            (async () => {
+              for await (const message of sub) {
+                const data = stringCodec.decode(message.data);
+                debug(
+                  `[${sub.getProcessed()}]: ${stringCodec.decode(message.data)}`
+                );
+                app.emit('sync-in', data);
+              }
+              debug('subscription closed');
+            })();
+            resolve(connection);
+          })
+          .catch((error) => reject(error));
       })
     };
-    app.on('sync-out', data => {
-      debug(`Publishing key ${key} to Nats`);
-      natsClient.publish(key, data);
-    });
 
-    natsClient.subscribe(key, (msg) => {
-      if (msg) {
-        debug(`Got ${key} message from Nats`);
-        app.emit('sync-in', msg);
-      }
+    app.on('sync-out', async (data) => {
+      const natsClient = await connect(natsConnectionOptions);
+      debug(`Publishing key ${subject} to NATS`);
+      await natsClient.publish(subject, stringCodec.encode(data));
+      await natsClient.drain();
     });
   };
 };