RFC: Support `from(Subscribable)`

[benlesh]

Idea

Support converting objects that have a subscribe() method to Observables.

class SomeThingy<T> {
   subscribe(observer: Observer<T>): { unsubscribe(): void } {
     /// stuff here.
    return {
       unsubscribe() {  }
    };
   }
}

const someThingy = new SomeThingy();

const anObservableOfSomeThingy$ = from(someThingy);

History

Originally, we opted not to do this, because RxJS 4 and under had observers that required methods that were different. Like { onNext, onError, onComplete }, where we had { next, error, complete }. Now that new usership of RxJS 4 and under is down, and mixed usership of RxJS 7 and RxJS 4 and under is almost non-existent, I think we can consider doing this.

Issues

1. There are lots of APIs in the world that have a subscribe function. Many of them take just a single function instead of an observer object. Can we support that too? or only Observer arguments?
2. We may at some point move to using AbortSignal as the second argument to subscribe, do we want to account for that now? We'd at least have to leave room.
3. Do we support returns of void or only if they return an Unsubscribable?
4. Should it just be a "from" helper? fromSubscribable?

Other things

This isn't a "for sure" plan. Thus far we haven't had much need or many requests for this feature. I just think it's time to consider it.

[benlesh]

Core Team Discussion: There's not really a clear use case. Closing for now.

[kajetansw]

@benlesh Hello, I seem to have found a use-case for it 😅 I really wished to use Observables in SolidJS where you have an observable method designed to convert Solid's signals to Observables.

Example of that:

import { observable } from "solid-js";
import { from } from "rxjs";

const [s, set] = createSignal(0);

// Creating Solid's observable from Solid signal
const solidObs$ = observable(s);

// Creating to RxJS observable from Solid signal
const rxObs$ = from(observable(s));

solidObs$.subscribe((v) => console.log(v));
rxObs$.subscribe((v) => console.log(v));

But there are some TS type errors when trying to do so, you can see it in action here: https://stackblitz.com/edit/solid-vite-ogtpkn?file=src/index.tsx

It seems like RxJS's ObservableInput type is incompatible with the Observable's definition in SolidJS.

Observable's definition in SolidJS is the one that is in shape of "objects that have a subscribe() method", just like described in this RFC.

The question is: Is it enough of a use-case to RFC?

---

I see that you already considered Subscribable to be added to a possible type for ObservableInput, like:

export type ObservableInput<T> =
  | Observable<T>
  | InteropObservable<T>
  | AsyncIterable<T>
  | PromiseLike<T>
  | ArrayLike<T>
  | Iterable<T>
  | ReadableStreamLike<T>
  | Subscribable<T>;

As it has the desirable shape already:

export interface Subscribable<T> {
  subscribe(observer: Partial<Observer<T>>): Unsubscribable;
}

And then the Observable be created with sth like:

export function fromSubscribable<T>(obj: any) {
  return new Observable((subscriber: Subscriber<T>) => {
    const obs = obj;
    if (isFunction(obs.subscribe)) {
      return obs.subscribe(subscriber);
    }
    // Should be caught by observable subscribe function error handling.
    throw new TypeError('Provided object does not correctly implement "subscribe" method');
  });
}

I find it very similar case to how the InteropObservable is already handled.

What do you think about this?

I'm sorry to bring this back from the dead and thanks in advance!

[samal-rasmussen]

I want to expose a "subscribable" interface from this tiny websocket rpc library of mine. But the full Observable interface is a bit fat to implement, and I definitely don't want no peer dependency, as I'm trying hard to keep things tiny and simple. The Subscribable interface looks perfect, but now I find that unfortunately from doesn't support it.

I see cartant has a relevant article here: https://ncjamieson.com/how-to-use-interop-observables/

But his solution is also a tad bit large for my goal of being very minimal.

For now I'm using this function instead of from.

export function subscribable_to_observable<T>(subscribable: Subscribable<T>): Observable<T> {
	return new Observable<T>((observer) => {
		return subscribable.subscribe(observer);
	});
}

Leaving the error message here so it is hopyfully at bit more searchable:

import { type Subscribable, from } from "rxjs";

const subscribable = {} as any as Subscribable<unknown>;
const osb = from(subscribable);

TS2345: Argument of type 'Subscribable<unknown>' is not assignable to parameter of type 'ObservableInput<any>'.
  Type 'Subscribable<unknown>' is missing the following properties from type 'Observable<any>': source, operator, lift, forEach, and 2 more.