Использование одного субъекта для распространения совершенно разных потоков событий через - PullRequest
Новая
       24

Использование одного субъекта для распространения совершенно разных потоков событий через

0 голосов
/ 06 марта 2012

Мне нужно прокси все потоки событий через одну тему.

Я придумал этот код: 

var mySubject,
    getObservable;

getObservable = function (subject, eventName) {
    return subject
        .asObservable()
        .filter(function (x) {
            return x.EventName === eventName;
        })
        .flatMap(function (x) {
            if (x.Type === 'onNext') {
                return Rx.Observable.return(x.Data);
            }

            if (x.Type === 'onError') {
                return Rx.Observable.throw(x.Data);
            }

            return Rx.Observable.empty();
        });
};

mySubject = new Rx.Subject();

getObservable(mySubject, 'foo')
    .subscribe(function(x){ 
        console.log('foo onNext ' + x); 
    }, function(x){ 
        console.log('foo onError ' + x); 
    }, function(){ 
        console.log('foo onComplete');
    });

getObservable(mySubject, 'bar')
    .subscribe(function(x){ 
        console.log('bar onNext ' + x); 
    }, function(x){ 
        console.log('bar onError ' + x); 
    }, function(){ 
        console.log('bar onComplete');
    });

mySubject.onNext({Type: 'onNext', EventName: 'foo', Data: 5});
mySubject.onNext({Type: 'onCompleted', EventName: 'foo'});

mySubject.onNext({Type: 'onNext', EventName: 'bar', Data: 5});
mySubject.onNext({Type: 'onError', EventName: 'bar', Data: 'Error message'});

Получил вывод: 

foo onNext 5

bar onNext 5
bar onError Error message

Ожидаемый результат: 

foo onNext 5  
foo onCompleted

bar onNext 5
bar onError Error message

Для события bar, которое работает как талисман: onNext будет распространяться, и как только возникает ошибка, вызывается функция onError и поток событий заканчивается. Тем не менее, я не могу заставить его работать на onComplete.

Когда появляется полное уведомление, я вижу, что вызывается Rx.Observable.empty(), но это не вызывает вызов обработчика onComplete для подписчиков. Вместо этого он вызывает свой обработчик onNext.

Ответы [ 2 ]

1 голос
/ 06 марта 2012

Функция getObservable возвращает наблюдаемое, которое подписывается на событие eventName, отправленное через subject. 

let getObservable = function (subject, eventName) {
    return Rx.Observable.create(function (observer) {
        subject
            .asObservable()
            .filter(function(x) {
                return x.EventName === eventName;
            })
            .map(function(x) {
                if (x.Type === 'onNext') {
                    observer.onNext(x.Data);
                }

                if (x.Type === 'onError') {
                    observer.onError(x.Data);
                }

                if (x.Type === 'onCompleted') {
                    observer.onCompleted();
                }

                return x;
            })
            .subscribe();
    });
};

Это рабочий пример с использованием данных из исходного вопроса: 

var mySubject,
    getObservable;

getObservable = function (subject, eventName) {
    return Rx.Observable.create(function (observer) {
        subject
            .asObservable()
            .filter(function(x) {
                return x.EventName === eventName;
            })
            .map(function(x) {
                if (x.Type === 'onNext') {
                    observer.onNext(x.Data);
                }

                if (x.Type === 'onError') {
                    observer.onError(x.Data);
                }
                
                if (x.Type === 'onCompleted') {
                    observer.onCompleted();
                }
                
                return x;
            })
            .subscribe();
    });
};

mySubject = new Rx.Subject();

getObservable(mySubject, 'foo')
    .subscribe(function(x){ 
        console.log('SomethingHappened onNext ' + x); 
    }, function(x){ 
        console.log('SomethingHappened onError ' + x); 
    }, function(){ 
        console.log('SomethingHappened onComplete');
    });


getObservable(mySubject, 'bar')
    .subscribe(function(x){ 
        console.log('DataUpdated onNext ' + x); 
    }, function(x){ 
        console.log('DataUpdated onError ' + x); 
    }, function(){ 
        console.log('DataUpdated onComplete');
    });

mySubject.onNext({Type: 'onNext', EventName: 'foo', Data: 5});
mySubject.onNext({Type: 'onCompleted', EventName: 'foo'});

mySubject.onNext({Type: 'onNext', EventName: 'bar', Data: 5});
mySubject.onNext({Type: 'onError', EventName: 'bar', Data: 'Error message'});
<script src='https://rawgit.com/Reactive-Extensions/RxJS/master/dist/rx.all.js'></script>
0 голосов
/ 06 марта 2012

в .NET Observable.SelectMany использует Observable.Merge для объединения потоков в одну составную наблюдаемую.IMHO Observable.Merge завершается только тогда, когда завершена любая из объединенных наблюдаемых.

например http://theburningmonk.com/2010/02/rx-framework-iobservable-merge/

Это может быть причиной проблемы.

...