如何从pubnub订阅创建一个observable

Question

我正在努力如何使用rxjs lib将以下内容转换为observable.

var client = PUBNUB.init({
  publish_key: 'pubkey',
  subscribe_key: 'subkey'
});

client.subscribe({
  channel: 'admin:posts',
  message: function(message, env, channel){console.log("Message");},
  connect: function(){console.log("Connected");},
  disconnect: function(){console.log("Disconnected");},
  reconnect: function(){console.log("Reconnected");},
  error: function(){console.log("Network Error");}, 
 });

我希望将消息回调转换为observable以及其余的回调.

关于如何做到这一点的任何想法？

谢谢

更新 - 2015年4月29日

以下是我最终使用它来实现这一目标.我应该补充一点,我需要在用户登录并清理注销后才订阅pubnub.

如果这是一个好方法,请告诉我:

var self = this;

var logins = Rx.Observable.create(function (obs) {
  //Using a session manager in ember.
  self.get('session').on('sessionAuthenticationSucceeded', function(e){
    var data = {
      token:this.content.secure.token,
      email:this.content.secure.email
    }

    obs.onNext(data);
  });
  self.get('session').on('sessionAuthenticationFailed', function(e){obs.onError(e)});
  return function(){
    self.get('session').off('sessionAuthenticationSucceeded', function(e){obs.onNext(e)});
    self.get('session').off('sessionAuthenticationFailed', function(e){obs.onError(e)});
  }
});

var logouts = Rx.Observable.create(function (obs) {
  self.get('session').on('sessionInvalidationSucceeded', function(){obs.onNext()});
  self.get('session').on('sessionInvalidationFailed', function(e){obs.onError(e)});
  return function(){
    self.get('session').off('sessionInvalidationSucceeded', function(e){obs.onNext(e)});
    self.get('session').off('sessionInvalidationFailed', function(e){obs.onError(e)});
  }
});

var dataStream = logins
  .map(function(credentials){
    return PUBNUB.init({
      publish_key: 'pub_key',
      subscribe_key: 'sub_key',
      auth_key: credentials.token,
      uuid: credentials.email
    });
  })
  .scan(function(prev, current){
    prev.unsubscribe({
      channel_group:'admin:data'
    });

    return current;
  })
  .concatMap(function(client){
    return Rx.Observable.create(function (observer) {

      client.subscribe({
        channel_group:'admin:data',
        message:function(message, env, channel){
          observer.onNext({message:message, env:env, channel:channel, error:null});
        },
        error:function(error){
          observer.onNext({error:error})
        }
      });

      return function(){
        client.unsubscribe({channel_group:'admin:data'});
      }
    });
  })
  .takeUntil(logouts)
  .publish()
  .refCount();

  var sub1 = dataStream.subscribe(function(data){
    console.log('sub1', data);
  });

  var sub2 = dataStream.subscribe(function(data){
    console.log('sub2', data);
  });

Answer 1

当然,您可以.create像这样创建一个源:

// your client
var client = PUBNUB.init({
    publish_key: 'pubkey',
    subscribe_key: 'subkey'
});

var source = Rx.Observable.create(function (observer) {
    client.subscribe({
         channel: 'admin:posts',
         message: function(message){ observer.onNext(message); }, // next item
         error: function(err){ observer.onError(message); },
         // other functions based on your logic, might want handling
    });
    return function(){ // dispose
        // nothing to do here, yet, you might want to define completion too
    }
});