Angular2/4 broadcast service using RxJS 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&lt;RadioEvent&gt;();
    private _eventCacheBus = new ReplaySubject&lt;RadioEvent&gt;();

    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 &lt; max; i++) {
            var cK = kArr[i];
            var cW = wArr[i];
            // '**' match all gragments
            if (cW == ww &amp;&amp; (typeof cK !== 'undefined')) {
                return true;
            }
            // test if fragments match
            if (!partMatch(cW, cK)) {
                return false;
            }
        }
        return true;
    };

    cast&lt;T&gt;(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&lt;T&gt;(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&lt;T&gt;(count);
            obs.subscribe(value =&gt; {
                subject.next(value);
            })
            return Observable.merge(normalobs, subject).distinct();
        }
    }
}
First of all, I recommend you to ensure your code is correct (see my comment on your question).

Here are a few suggestions about improving your code.


Please do not use non-informative names like w, ww, sep, kArg, cW, CK, and even the "well-known" i, j, k, m, n. They are absolutely horrible. They make decent code be an unreadable puzzle. They hide the intent and make me think where I should not be.
Please always make sure you use Type when a) the compiler loses (or is unable to infer) the type information; b) you can narrow down the type. That includes the function return type. Typing is not panacea but it helps prevent an entire class of bugs.
Do not use generic methods smth&lt;T&gt;(...) unless you use the type T information anywhere in the function.
Minimize the distance between variable declaration, assignment, and first usage. Better declare and assign in the same place. Ideally, find a way to not even declare a variable.
In error messages be more specific about what went wrong.




  const star = '*';
  const twoStars = '**';

  export class RadioService {
    private separator = ':';
    private _eventBus = new Subject&lt;RadioEvent&gt;();
    private _eventCacheBus = new ReplaySubject&lt;RadioEvent&gt;();

    keyMatch(keyList: string, wildcardList: string): boolean {
      const isMatch = (wildcard, key) =&gt; (wildcard === star) || (wildcard === key);

      const allKeys = keyList.split(this.separator);
      const allWildcards = wildcardList.split(this.separator);

      const keyWildcardPairs = allKeys
        .reduce((accum, current, index) =&gt; {
          accum.push([current, allWildcards[index]]);
          return accum;
        }, &lt;string[][]&gt;[]);

      return keyWildcardPairs.some(pair =&gt; {
        const key = pair[0];
        const wildcard = pair[1];
        // '**' match all gragments
        if (wildcard == twoStars &amp;&amp; (typeof key !== 'undefined')) {
          return true;
        }
        // test if fragments match
        if (!isMatch(wildcard, key)) {
          return true;
        }
        return false;
      });
    };

    cast(key: string, data?: any): void {
      if (typeof key !== 'string' || key.length &lt;= 0) {
        throw `Bad key '${JSON.stringify(key)}'. Please provide a non-empty string.`;
      }

      this._eventBus.next({ key: key, data: data });
      this._eventCacheBus.next({ key: key, data: data });
    };


    on(key: string, count?: number): Observable&lt;any&gt; {
      const _this = this;

      const normalObservable = this._eventBus
        .filter(event =&gt; _this.keyMatch(event.key, key))
        .map(event =&gt; event.data);

      const withoutReplay = _.isNil(count);
      if (withoutReplay)
        return normalObservable;

      const subject = new ReplaySubject&lt;number&gt;(count);
      this._eventCacheBus
        .filter(event =&gt; _this.keyMatch(event.key, key))
        .map(event =&gt; event.data)
        .subscribe(value =&gt; subject.next(value));

      return Observable
        .merge(normalObservable, subject)
        .distinct();
    }
  }