Добавление завершения к наблюдаемому, которое никогда не завершается

Question

Я пытаюсь добавить дополнение к плагину Cordova с помощью оболочки Ioni c. Так что я поймаю завершение в подписчике. Итак, у меня есть SomeSerive, который является ioni c оболочкой для кордонового плагина:

 startListen(){
            return this.someService.startListen().pipe(
                switchMap((response) => {
                    return new Observable(subscriber => {
                        if (response.index !== undefined) {
                            subscriber.next(1);
                        }
                        if (response.errorTitle === Message.NOT_FOUND) {
                            subscriber.error(Message.NOT_FOUND);
                        }
                        if (response.errorTitle === Message.CANCELLED) {
                            console.log('blah blah'); <---### IS PRINTED
                            subscriber.complete();
                        }
                    });
                }),
            );
        }

И я называю это некоторым компонентом:

 startListen(){
            this.someFacade.startListen().subscribe(
                (x) => console.log('next', x),
                (x) => console.log('error', x),
                () => console.log('complete'), <---# NOT PRINNTED
            );

Проблема в том, что console.log ('complete') никогда не запускается.

Что не так?

Answer 1

takeWhile оператор делает то, что вы хотите:

this.someService.startListen().pipe(
  takeWhile(response => response.errorTitle !== Message.CANCELLED),
  switchMap(response => response.index !== undefined
    ? of(response.index)
    : throwError(response.errorTitle)
  )

Answer 2

Нашел проблему. Мне нужно закрыть первую наблюдаемую тоже. Итак, мое решение:

const completed = new Subject();
        return this.someService.startListen().pipe(
            switchMap((response) => {
                return new Observable(subscriber => {
                    if (response.index !== undefined) {
                        subscriber.next(response.index);
                    }
                    if (response.errorTitle === Message.NOT_FOUND) {
                        subscriber.next(Message.NOT_FOUND);
                    }
                    if (response.errorTitle === Message.CANCELLED) {
                        subscriber.complete();
                        completed.next();
                        completed.unsubscribe();
                    }
                });
            }),
            takeUntil(completed)
        );