Преобразование наблюдаемого ECMAScript в наблюдаемое RXJS

Question

У меня есть совместимый со спецификацией ECMAScript Observable, в частности из библиотеки wonka . Я пытаюсь преобразовать этот тип наблюдаемого в наблюдаемый rxjs 6, но безуспешно.

Кажется, это было возможно с rxjs 5. Я пробовал это:

      import { pipe, toObservable, fromArray } from 'wonka';
import { from } from 'rxjs';
...
from(
  pipe(
    fromArray([1, 2, 3]),
    toObservable
  )
);

Я получаю в браузере такую ​​ошибку:

      ERROR TypeError: You provided an invalid object where a stream was expected. 
You can provide an Observable, Promise, Array, or Iterable.

а затем это:

      Argument of type 'observableT<any>' is not assignable to parameter
of type 'ObservableInput<any>'

в диалоговом окне Visual Code.

Как это может быть сделано?

Спасибо J

Answer 1

Догадаться:

      npm i zen-observable
npm i --save-dev @types/zen-observable

      getObservable(): Observable<any> {
  return from(
    pipe(
      fromArray([1, 2, 3]),
      toObservable
    ) as any) as any;
}

Answer 2

Напишите функцию преобразования, которая использует тот факт, что API-интерфейсы подписки идентичны:

      const zenToRx = <T>(zenObservable: Zen.Observable<T>): Rx.Observable<T> =>
  new Rx.Observable(
    observer => zenObservable.subscribe(observer)
  );

Answer 3

По какой-то причине ни один из других ответов по состоянию на 8 июня 2023 г. мне не помог.

В итоге я написал свой собственный небольшой инъекционный класс обслуживания.

Вот:

      // Class: ZenToRxService (injectable service)
// Author: UChin Kim
// Date: 08 Jun 2023.
//
// Notes: The idea here is to "convert" a Zen Observable to an Rx
// Observable, so that you can use nifty Rx features, like pipes and
// utilities that make use of pipes, such as @ngneat/until-destroy.
//
// Strategically, we mirror the Zen Observable's events onto an
// Rx Observable. Easiest way to do this is to use the Rx Subject type, which
// is both an Observable and an event emitter. We create our own Zen
// subscription, where we simply emit the event value using the Subject, which
// is, again, both an event emitter and an Observable.
//
// You can take the resulting Subject<T> and handle it as an Rx Observable<T>.
// Or, you could handle it as an Rx Subject<T>. Whatever floats your boat :).
//
// The advantage of Rx Observables/Subjects over Zen Observables is that you
// can use the Rx pipes, and subsequently, do nifty stuff like use
// @ngneat/until-destroy to automatically clean up subscriptions, without
//  needing to maintain an explicit reference to them.
//
// Note that we automatically unsubscribe our custom Zen subscription when
// the Subject<T> object is unsubscribed or terminated.
//
// You are responsibile  for cleaning up the subscription that you obtain on
// the returned Subject<T> object.
//
// I assume you already know how to inject an Angular service into a component.
//
//
import { Injectable } from '@angular/core';
import { Subject, finalize } from 'rxjs';
import { Observable } from 'zen-observable-ts';

@Injectable({
  providedIn: 'root',
})
export class ZenToRxService {
  convertZenObservable<T>(zenObservable: Observable<T>): Subject<T> {
    const subject = new Subject<T>();
    const zenSubscription = zenObservable.subscribe((e: T) => subject.next(e));

    return subject.pipe(
      finalize(() => {
        zenSubscription.unsubscribe();
        console.log('zen subscription was unsubscribed!');
      })
    ) as Subject<T>;
  }
}