I am no expert in RxJS as I am still learning my way around the new paradigm of Observable/Subject.
I was trying to implement a broadcast service (based on NgRadio) that allows a new subscriber to get the nth last value of events he was listening to and the next emitted values after subscription.
As I stumbled upon issues with ReplaySubject to ask for the last n values... I ended up implementing it as follows. I feel that the code is not that crisp and clean so I would appreciate any improvement from more experienced RxJS developers.
import { BehaviorSubject } from 'rxjs/BehaviorSubject';
import { Observable } from 'rxjs/Observable';
import { ReplaySubject } from 'rxjs/ReplaySubject';
import { Subject } from 'rxjs/Subject';
import * as _ from 'lodash';
export interface RadioEvent {
key: string;
data?: any;
}
export class RadioService {
private separator = ':';
private _eventBus = new Subject<RadioEvent>();
private _eventCacheBus = new ReplaySubject<RadioEvent>();
keyMatch(key, wildcard) {
var w = '*';
var ww = '**';
var partMatch = function (wl, k) {
var match = (wl === w) || (wl === k);
return match;
};
var sep = this.separator;
var kArr = key.split(sep);
var wArr = wildcard.split(sep);
var kLen = kArr.length;
var wLen = wArr.length;
var max = Math.max(kLen, wLen);
for (var i = 0; i < max; i++) {
var cK = kArr[i];
var cW = wArr[i];
// '**' match all gragments
if (cW == ww && (typeof cK !== 'undefined')) {
return true;
}
// test if fragments match
if (!partMatch(cW, cK)) {
return false;
}
}
return true;
};
cast<T>(key: string, data?: any) {
if (typeof key !== 'string' || !key.length) {
throw 'Bad key. Please provide a string';
}
this._eventBus.next({ key: key, data: data });
this._eventCacheBus.next({ key: key, data: data });
};
on<T>(key: string, count?: number) {
var _this = this;
var normalobs = this._eventBus
.filter(function (event: RadioEvent) {
return _this.keyMatch(event.key, key);
}).map(function (event) {
return event.data;
});
if (_.isNil(count)) {
return normalobs;
} else {
let obs = this._eventCacheBus
.filter(function (event: RadioEvent) {
return _this.keyMatch(event.key, key);
}).map(function (event) {
return event.data;
});
let subject = new ReplaySubject<T>(count);
obs.subscribe(value => {
subject.next(value);
})
return Observable.merge(normalobs, subject).distinct();
}
}
}
if (!partMatch(cW, cK)) { return false; }
should probably beif (!partMatch(cW, cK)) { continue; }
or evenif (!partMatch(cW, cK)) { return true; }
instead, while the last instruction inkeyMatch()
should bereturn false
. \$\endgroup\$ – Igor Soloydenko Sep 8 '17 at 17:13