CODEWITHSUNDEEP
javascriptrxjs
undefined
undefined

Reputation: 6874

RxJS custom operator doesn't call complete

I have the following custom operator:

export function test() {
  return function(source) {
    return new Observable(observer => {

      return source.subscribe({
        next(value) {
          observer.next(value);
        },
        error(error) {
          observer.error(error);
        },
        complete() {
          console.log('completed')
          observer.complete();
        }
      })
    });
  }
}

The problem is that when I subscribe to an observable that completes I didn't get to the complete callback, i.e I don't see the completed log.

interval(1000).pipe(
  test(),
  take(2)
).subscribe();

What am I missing?

Upvotes: 4

Views: 387

Answers (1)

Християн Христов
Християн Христов

Reputation: 3588

nice question, now lets dive in the explanation.

First, let's see the solution

function test(limitter) {
  return function(source) {
    return new Observable(observer => {
      return source.subscribe({
        next(value) {
          observer.next(value);
        },
        error(error) {
          observer.error(error);
        },
        complete() {
          console.log("completed action");
          observer.complete("completed value");
        }
      });
    });
  };
}

interval(1000)
  .pipe(
    take(2),
    test()
  )
  .subscribe(
    x => {
      console.log("result: ", x);
    },
    err => {
      console.log(err);
    },
    end => {
      console.log("Observable has been completed");
    }
  );

So what's the difference, in this snippet the take operator is prior to the custom test() operator, which means that whenever we hit the desired count (in our case 2), the take operator will return completed source which will trigger our complete method inside the follow up subscribers (In our case inside the custom test operator and also inside the subscribe ) after that the source won't emit anything else, because it has already been completed.

You can check out the source => take(), for more information feel free to ask anything if there are some blurry parts.

Upvotes: 2

Related Questions