How to use the rx-lite.Subject function in rx-lite

constructor() {
    const self = this;

    this._inputSubscription = null;
    this.setInput(new Subject());
    this._state = new Subject();
    this._output = new Subject();
    this._activators = [];
    this._paths = [];
    this._reducers = {};
    this._sagas = {};
    this._sagaMiddleware = createSagaMiddleware();
    this._store = createStore(
      (state) => state,
      applyMiddleware(this._sagaMiddleware)
    );

    // Set up internal sagas.

    // Send.
    this._sendSaga = function* sendSaga({ data, sessionId }) {
      yield self._send(data, sessionId);

public emit(name: string, data: Object): void {
    let fnName: string = this._createName(name);

    if (!this.subjects[fnName]) {

      this.subjects[fnName] = new Subject();
    }

    this.subjects[fnName].onNext(data);
  }

}
        region.ttl = region.ttl || 120;

        const registerNeedTypeForCaptain = () => axios.post(`${dav.missionControlURL}/captains/${dav.davId}`, { need_type: needType, region })
          .catch((err) => {
            console.error(err);
          });

        registerNeedTypeForCaptain();

        rx.Observable.interval(90000 /*1.5m*/).subscribe(
          registerNeedTypeForCaptain,
          (err) => console.log('Error: ' + err),
          () => console.log(''));

        const observable = new rx.Subject();
        dav.needTypes[needType] = observable;

        observable.update = async () => {
          await axios.put(`${dav.missionControlURL}/captains/${dav.davId}/needs`, { need_type: needType, region });
        };

        return observable;
      }
    };

forBid: (bidId, contract) => {
        dav.contracts[bidId] = new rx.Subject;
        contract.dav_id = dav.davId;
        axios.post(`${dav.missionControlURL}/contracts/${bidId}`, contract)
          .catch((err) => {
            console.error(err);
          });
        return dav.contracts[bidId];
      }
    };

var searchTerm = keySlash.flatMap(function (md) {
    self.searchMode = true;
    self.searchTerm = '';
    self.render();
    var end$ = new rx.Subject();
    var done$ = rx.Observable.merge(events.line, end$);
    return alphaNumeric.map(function (e) {
      if (e.key.name === 'backspace' && self.searchTerm.length) {
        self.searchTerm = self.searchTerm.slice(0, -1);
      } else if (e.value) {
        self.searchTerm += e.value;
      }
      if (self.searchTerm === '') {
        end$.onNext(true);
      }
      return self.searchTerm;
    })
    .takeUntil(done$)
    .doOnCompleted(function() {
      self.searchMode = false;
      self.render();

build(ipkgFile) {
    this.idrisBuildSubject = new Rx.Subject
    this.idrisBuild(this.compilerOptions, ipkgFile)
    return this.idrisBuildSubject
  }

prepareCommand(cmd) {
    let id = this.getUID()
    let subject = new Rx.Subject
    this.subjects[id] = subject
    this.warnings[id] = []
    this.ideMode(this.compilerOptions).send([cmd, id])
    return subject
  }

public on(name: string, handler: any): void {
    let fnName: string = this._createName(name);

    if (!this.subjects[fnName]) {

      this.subjects[fnName] = new Subject();
    }

    this.subjects[fnName].subscribe(handler);
  }