Support setting subscribe request data

dist/centrifuge.d.ts

@@ -210,6 +210,7 @@ declare namespace Centrifuge {
 
     export interface SubscribeOptions {
         since?: StreamPosition;
+        data?: any;
     }
 
     export interface StreamPosition {

src/centrifuge.js

@@ -1136,6 +1136,10 @@ export class Centrifuge extends EventEmitter {
       }
     };
 
+    if (sub._subscribeData) {
+      msg.params.data = sub._subscribeData;
+    }
+
     // If channel name does not start with privateChannelPrefix - then we
     // can just send subscription message to Centrifuge. If channel name
     // starts with privateChannelPrefix - then this is a private channel
@@ -1902,16 +1906,13 @@ export class Centrifuge extends EventEmitter {
     if (currentSub !== null) {
       currentSub._setEvents(events);
       if (currentSub._isUnsubscribed()) {
-        currentSub.subscribe();
+        currentSub.subscribe(opts);
       }
       return currentSub;
     }
     const sub = new Subscription(this, channel, events);
     this._subs[channel] = sub;
-    if (opts && opts.since) {
-      this._setSubscribeSince(sub, opts.since);
-    }
-    sub.subscribe();
+    sub.subscribe(opts);
     return sub;
   };
 }

src/client.proto.json

@@ -349,6 +349,10 @@
                     "offset": {
                       "type": "uint64",
                       "id": 7
+                    },
+                    "data": {
+                      "type": "bytes",
+                      "id": 8
                     }
                   }
                 },

src/index.js

@@ -1,2 +1,2 @@
-import {Centrifuge} from './centrifuge.js';
+import { Centrifuge } from './centrifuge.js';
 export default Centrifuge;

src/index_protobuf.js

@@ -1,2 +1,2 @@
-import {CentrifugeProtobuf} from './protobuf.js';
+import { CentrifugeProtobuf } from './protobuf.js';
 export default CentrifugeProtobuf;

src/subscription.js

@@ -27,6 +27,7 @@ export default class Subscription extends EventEmitter {
     this._initializePromise();
     this._promises = {};
     this._promiseId = 0;
+    this._subscribeData = null;
     this.on('error', function (errContext) {
       this._centrifuge._debug('subscription error', errContext);
     });
@@ -51,7 +52,7 @@ export default class Subscription extends EventEmitter {
         this._ready = true;
         reject(err);
       };
-    }).then(function () {}, function () {});
+    }).then(function () { }, function () { });
   };
 
   _setNeedRecover(enabled) {
@@ -198,6 +199,10 @@ export default class Subscription extends EventEmitter {
     return subscribeErrorContext;
   };
 
+  _setSubscribeData(data) {
+    this._subscribeData = data;
+  }
+
   ready(callback, errback) {
     if (this._ready) {
       if (this._isSuccess()) {
@@ -212,10 +217,13 @@ export default class Subscription extends EventEmitter {
     if (this._status === _STATE_SUCCESS) {
       return;
     }
-    this._noResubscribe = false;
     if (opts && opts.since) {
       this._centrifuge._setSubscribeSince(this, opts.since);
     }
+    if (opts && opts.data) {
+      this._setSubscribeData(opts.data);
+    }
+    this._noResubscribe = false;
     this._centrifuge._subscribe(this);
   };
 
@@ -232,7 +240,7 @@ export default class Subscription extends EventEmitter {
     }
     let subPromise = new Promise((res, rej) => {
       const timeout = setTimeout(function () {
-        rej({'code': 0, 'message': 'timeout'});
+        rej({ 'code': 0, 'message': 'timeout' });
       }, this._centrifuge._config.timeout);
       this._promises[this._nextPromiseId()] = {
         timeout: timeout,