Replace observables, observers with subjects?

[xhydongda]

import { Subject } from 'rxjs'
import { Observable } from 'rxjs/internal/Observable'

export type ALL = 'ALL'
const ALL: ALL = 'ALL'

interface State {
callChain: Set<keyof Messages | ALL>
subjects:Map<keyof Messages | ALL, Subject>
options: Options
}

export type Options = {
onCycle(chain: (keyof Messages | ALL)[]): void
isDevMode: boolean
}

export class Emitter {

private emitterState: State

constructor(options?: Partial<Options>) {

let DEFAULT_OPTIONS: Options<Messages> = {
  isDevMode: false,
  onCycle(chain) {
    console.error(
      '[typed-rx-emitter] Error: Cyclical dependency detected. '
      + 'This may cause a stack overflow unless you fix it. '
      + chain.join(' -> ')
    )
  }
}

this.emitterState = {
  callChain: new Set,
  subjects: new Map,
  options: {...DEFAULT_OPTIONS, ...options}
}

}

/**

• Emit an event (silently fails if no listeners are hooked up yet)
  */
  emit(key: K, value: Messages[K]): this {
  let { isDevMode, onCycle } = this.emitterState.options
  if (isDevMode) {
  if (this.emitterState.callChain.has(key)) {
  onCycle(Array.from(this.emitterState.callChain).concat(key))
  return this
  } else {
  this.emitterState.callChain.add(key)
  }
  }
  if (this.hasSubject(key)) {
  this.emitSubject(key, value)
  }
  if (this.hasSubject(ALL)) {
  this.emitSubject(ALL, value)
  }
  if (isDevMode) this.emitterState.callChain.clear()
  return this
  }

/**

• Subscribe to an event
  */
  on(key: K): Observable<Messages[K]> {
  return this.getSubject(key)
  }

/**

• Subscribe to all events
  */
  all(): Observable<Messages[keyof Messages]> {
  return this.getSubject(ALL)
  }

///////////////////// privates /////////////////////

private getSubject(key: K | ALL) {
if(!this.emitterState.subjects.has(key)){
this.emitterState.subjects.set(key, new Subject());
}
return this.emitterState.subjects.get(key);
}

private emitSubject(
key: K | ALL,
value: Messages[K]
) {
this.emitterState.subjects.get(key)!.next(value);
}

private hasSubject(key: K | ALL): boolean {
return this.emitterState.subjects.has(key);
}
}