How to create an observable from pubnub subscribe

Question

I am struggling to convert the following to observables using rxjs lib.

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");}, 
 });

I would like the message callback to be converted to observable along with the rest of these callbacks.

Any ideas on how to do this?

thank

UPDATE - 4/29/15

Here is what I ended up using to make it work. I must add that I need to subscribe to pubnub only after the user writes and clears the logout.

Please let me know if this is a good approach:

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

Of course, you can create the source using .createas such:

// 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
    }
});

Answer 2

PUBNUB, , , :

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

var source = Rx.Observable.defer(function() {
    return Rx.Observable.create(function (observer) {
        client.subscribe({
            channel: 'admin:posts',
            message: function(message){ observer.onNext(message); },
            connect: function(){console.log("Connected");},
            disconnect: function(){console.log("Disconnected");},
            reconnect: function(){console.log("Reconnected");},
            error: function(err){ observer.onError(err); }
        });

        return Rx.Disposable.create(function() { // dispose
            client.unsubscribe();
        });
    });
})
.publish()
.refCount();