How to use the rx-lite.Observable.create function in rx-lite

generatePromiseOrObservable = (json: Object): LivePromise => {
        const reqId = json.req_id.toString();

        if (this.useRx) {
            const obs = Observable.create(observer => {
                // if call is an subscription, store it in uncompleteStreamObs
                // ticks is a special case that's a stream without subscribe keyword
                if (json.subscribe || json.ticks) {
                    this.uncompleteStreamObs[reqId] = observer;
                } else {
                    this.uncompleteOneTimeObs[reqId] = observer;
                }

                return () => {
                    delete this.uncompleteOneTimeObs[reqId];
                    delete this.uncompleteStreamObs[reqId];
                };
            });

            const published = obs.publish();

__wrapAsObservable(
    fn: (options: TOptions) => TResult,
    options: TOptions
  ): Observable {
    return Observable.create(observer => {
      observer.onNext(fn(options));
      var subscription = this.subscribe(() => observer.onNext(fn(options)));
      return () => subscription.remove();
    }).distinctUntilChanged(
      /*keySelector*/ null,
      (a, b) => a === b,
    );
  }
}

WebSocket = websocket;
        }

        this.bufferedSends = [];
        this.bufferedExecutes = [];
        this.unresolvedPromises = {};

        this.events = new LiveEvents();
        this.apiState = new ApiState();

        if (useRx) {
            this.uncompleteOneTimeObs = {};
            this.uncompleteStreamObs = {};
        }

        this.authStream = Observable.create(observer => {
            this.onAuth = observer;

            return (): void => {
                this.onAuth = null;
            };
        });

        this.bindCallsAndStateMutators();

        this.connect(connection);
    }