我正在努力了解如何使用 rxjs 库将以下内容转换为 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。
关于如何做到这一点的任何想法?
谢谢
更新 - 15 年 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);
});
最佳答案
当然,你可以像这样使用 .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
}
});
关于pubnub - 如何从pubnub subscribe创建一个observable,我们在Stack Overflow上找到一个类似的问题:https://stackoverflow.com/questions/29858913/