используя Connect и публиковать на том же наблюдаемом в rxjs

Question

Я пытаюсь сделать следующее:

1. Взять наблюдаемую text$ (в моем приложении это входной поток), а затем передать это значение от ввода до родительского компонента,родительский компонент затем вызовет API и получит некоторые данные.
2. ПОДОЖДИТЕ, пока родительский объект не отправит обратно наблюдаемую (<string[]>) как @input, родительский компонент отправит обратно компоненту нижеданные, полученные из API.
3. После того, как мы знаем, что у нас есть новые данные, извлеченные из родительского компонента, я хочу выполнить другой канал для той же наблюдаемой (text$), которая является моим входным потоком.

Пока что это то, что я придумал:

import { of, Observable, ConnectableObservable } from 'rxjs';
import { take, map, publish, tap, toArray, distinctUntilChanged, debounceTime, switchMap, repeat } from 'rxjs/operators';


const tt = of('a','b','c','d')
let text$ = of('nct')
const data = tt.pipe(
      map((result: any) => {
        return result
      })
    ) as Observable<string[]>


let dataSnap;
    data.pipe(toArray()).subscribe(x => dataSnap = x)
    let obs1 = text$.pipe(
      distinctUntilChanged(),
      debounceTime(250),
      // tap((term) => this.target = text$),
      tap((term) => console.log("handleTypeahead", term)),
      // tap((term) => this.onTypeahead.emit(term)),
    ) as ConnectableObservable<string>
    let obs2 = obs1.pipe(
      publish(),
      switchMap(term => term === '' ? [] : data)
    ) as ConnectableObservable<string>
      data
        .pipe(
          tap(x => console.log(dataSnap)),
          toArray(),
          ).subscribe(x => {
        if (x == dataSnap){
          repeat(2)
        }
        else {
         obs2.connect()
        }
          })

Или здесь, на стеке: https://stackblitz.com/edit/typescript-p3yoqa В настоящее время моя проблема, похоже, связана с connect, гдеЯ получаю эту ошибку: Property 'connect' does not exist on type 'Observable<string>'.Я попробовал предложенные решения, но, похоже, реализовал их неправильно.Кроме того, я не уверен, что моя логика правильна, поскольку я новичок в RXJS.

Answer 1

Observable - это не ConnectableObservable, это одно.Другое дело, что если то, что pipe возвращает, будет ConnectableObservable, код будет работать так же без фактического приведения типов.Очевидно, что компилятор будет жаловаться, но его легко отключить (как вы это сделали в любом случае).Кастинг не влияет на время выполнения и, конечно, НЕ на тип объекта.

Вывод:

1. Кастинг не имеет значения во время выполнения
2. Whetever pipe возвращается, этоНЕ ConnectableObservalbe

Я посмотрел источники rxjs на github и обнаружил, что, например,

https://github.com/ReactiveX/rxjs/blob/master/src/internal/util/pipe.ts

EDIT

pipe оператор возвращает UnaryFunction, который расширяет Observable, но не ConnectableObservable.

Это не правда, что UnaryFunction расширяет Observable.Я понятия не имею, как я это придумал (просматривая на github ....)