Implement different types of Subjects

[lohiaad]

Hi,

I really liked your work. i just wanted to know how can i implement ReplaySubject.

Thanks

[govorov]

Sorry for so late response. Could you please clarify, what are you trying to accomplish?

[hghammoud]

This is because you are using Subject which is emit and forget strategy. You need to implement Behavioral Subject or ReplaySubject to allow getting last values upon subscription.

I am no expert in Rx as i'm still trying to find my way around the paradigm but this is my adaptation that allows a new subscriber to get the n last values of his even upon subscription. Feel free to make the code cleaner.

`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();
private _eventCacheBus = new ReplaySubject();

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';
    }

    // if(_.isNil(this._eventBus[key])){
    //     this._eventBus[key] = new ReplaySubject<RadioEvent>();
    // }
    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();
    }
}

}
`