kosich/rxjs-autorun

Re-evaluate an expression whenever Observable in it emits

🔭 my projects:

• NoCode
  My WIP nocode startup

• RecksJS
  <div>{ ajax.getJSON(URL) }</div>
  JSX + RxJS Framework

• <$> elements for React
  <$div>{ ajax.getJSON(URL) }</$div>
  React elements with RxJS content / params

• ThinkRx.io
  ---o--o-oo-o----
  Instant time-accurate RxJS marble diagrams

• Framd.cc
  🙂 😁 😆
  Funny Emoji animations

📖 my articles:

• "Rx+JSX experiment: Recks framework" 🐶
  Serie of posts reviewing Recks features — Observables inside JSX
  <div>{ timer(0, 100) }</div>

• "Fetching Data in React with RxJS and <$> fragment"
  React with RxJS.fromFetch and <$> JSX helper

• "Text to GIF animation — React Pet Project Devlog"
  React, RxJS, Canvas & Text-to-Speech

• "RegExp syntax for Observables: Never Been Easier!"
  Use regexp-like syntax to query observables. EXPERIMENT

• "Error handling in RxJS"
  Learn how to deal with errors and retry failed requests

• "Pausable Observables in RxJS"
  Create a push back force in your streams

• "RxJS debounce vs throttle vs audit vs sample"
  Difference You Should Know

rxjs-autorun's Issues

Support for async expressions?

I think the scope of applicability of this tool would greatly (and modularly) expand with support of async expressions:

• It removes the need for creating and then handling higher-order observables when single async operations are intended (which to me seems to be a dominating use case of higher-order observables, basically mapping some observable such as user input to a request).

  const a = fromUserInput()
  
  const fetched = computed(async () => {
    const res = await fetch(`https://my.api/${$(a)}`)
    const json = await res.json()
    
    return json
  })

• Combined with a cancelling behaviour, it would also provide means for most common use cases of flow control:

  const a = fromUserInput()
  
  const fetched = cpmputed(async () => {
    const q = $(a)
  
    await sleep(200) // --> this debounces for 200ms
    const res = await fetch(`https://my.api/${$(a)}`)
    const json = await res.json()
  
    return json    
  })

Some notes / open questions:

• This wouldn't work with global tracking functions.
• How would subscription management work? Like what if a source is visited during a cancelled run, after being unsubscribed?

A way to suppress/skip an emission

I was about to discuss if we need a way to skip (suppress, filter) an emission based on some logic:

import { compute, $, _, SKIP } from 'rxjs-autorun';

const even$ = compute(() => $(o) % 2 == 0 ? _(o) : SKIP);

SKIP being some special object that we'll filter out.

But then again I realized that the magic is already there:

import { compute, $, _, SKIP } from 'rxjs-autorun';
import { NEVER } from 'rxjs';

const even$ = compute(() => $(o) % 2 == 0 ? _(o) : _(NEVER));

Is not that beautiful?
Think we gotta document this trick too 🙂

cc @Jopie64

Support multiple independent subscriptions of the same runner

Hello,

Thanks again for this amazing work!

I tried out it's behavior when the same observable returned from run() is subscribed multiple times. I expect it to then also subscribe the upstream observables multiple times. (It should not share the subscription I think. If that is desirable, one could always use shareReplay() or something.) But currently that is not what's happening.

I created some tests in this commit: 5da93e0. Most of them fail. I also tried to make a quick fix, but failed to do so yet...

Do you agree about the new tests I made? If not, what do you think should happen when multiple subscribers subscribe the same runner?

I hope you'll look into this as well.

Greetings,
Johan

Completing with EMPTY observable

this is a continuation of discussion in #19

When an observable in an expression completes without emitting any value — it often makes whole expression impossible to emit. Here are some examples:

// source streams
const a = timer(0, 1_000);
const b = EMPTY; // imagine b comes from an external API

// combinations:
computed(() => ({ a: $(a), b: $(b) }))
computed(() => [$(a), $(b)])
computed(() => $(a) || $(b))

Currently we swallow this issue. And it's not that easy to detect this completion if computed doesn't report it in any way! Users would probably need to plug in some combination of tap pushing to Subject that would complete computed via takeUntil.

Though when it comes to conditions, it's not that obvious:

computed(() => $(a) < 5 ? $(a) : $(b)); // GOOD: values > 5 are endless
computed(() => $(a) > 5 ? $(a) : $(b)); // BAD: values > 5 are meaningless

But we don't know which of the two cases we face. So to choose what behavior is more obvious — I'd vote for completion with EMPTY. It's very much expected in simpler cases and in conditional cases user would be able to easily handle it with defaultIfEmpty operator or concat(NEVER), depending on the need.

Completion with EMPTY also has a feature to it: users would be able to intentionally complete their streams:

computed(() => $(a) < 5 ? $(a) : _(EMPTY));

That's like a.pipe( takeWhile(x < 5) ) in an expression!

@Jopie64 as to the case you've mentioned with unsubscription and filtering: IMHO, it makes sense that if user wants to unsubscribe eagerly — they would mark it as weak, just as we intended:

const r = computed(() => {
   switch($(a)) {
      case 0: return $.weak(b); // weak b
      case 1: return $(c); // intentionally normal c
      default: return $(NEVER); // skip emission if not 0 or 1
   });

While default: return $(EMPTY) would mean that user wants to complete the stream.

LMKWYT 🙂

Untracking unused streams

As @Jopie64 describes in #7, in an expression with a condition there might be a case when an Observable becomes unused:

const ms = timer(0, 1);
const s = timer(0, 1000);
run(() => $(s) % 2 : $(ms) : 'even');

Steps of calculation:

1. s=0 that would result to "even"
2. s=1 that would lead to subscription to ms and a thousand of ms values (or ~ thousand)
3. s=2 means that expression emits "even" and ms stream is not needed for the next second
4. s=3 means ms stream is again used and we'll emit a thousand of ms values

Current behavior:
Due to ms being tracked, the expression would be still needlessly recalculated a thousand times. Even though at step # 3 we learned that ms is not used when s=2.

Expected behavior:
I think, when ms is detected unused on step # 3, we should mark it as "untracked" until it is again used. So $ turns under the hood into _ until it's used again.

Proposal to wait for at least one value from untracked observable

Description

Before I forget, here my proposal for the case that an untracked observable doesn't immediately (synchronously) emit.

In the code there is a place where a decision should still be made:

                                // NOTE: what to do if the silenced value is absent?
                                // should we:
                                // - interrupt computation & w scheduling re-run when first value available
                                // - interrupt computation & w/o scheduling re-run
                                // - continue computation w/ undefined as value of _(o)

I'd propose to go for the first. I think the spec would be something like: an observable is always tracked before it emits its first value. Only then it will be tracked when resolved with $.

Rationale

Consider this scenario:

const r = run(() => _(o1) + $(o2));

If we do the second, this resulting expression will complete without emitting when o1 doesn't immediately emit.
If we do the third, then a user should always check untracked observable values for undefined.

In case we do the first, it will wait for a value of o1, then subscribe o2 and start to emit values whenever o2 emits.
This IMHO is the best solution.

Let me know what you think. Could you think of a scenario where the first solution would do something inconsistent?

Alternative APIs

Here we're exploring API suggestions that might be handy or help with handling README#precautions:

1. Expression with explicit dependencies by @voliva :

const a = timer(0, 1000);
const b = timer(0, 1000);
const c = computed((a, b) => a + b,  a, untrack(b)))

This will react to every change and is side-effect safe
It's basically a combineLatest + withLatestFrom + map — most honest way to react to every Observable emission

2. Async/await syntax:

const a = timer(0, 1000);
const b = timer(0, 1000);
const c = computed(async () => await $(a) + await $(b))

It might react to every change, and is side-effect safe

3. Provide initial value via tracker by @loreanvictor

Trackers might with initial value (default value?)

const a = interval(1000);
const b = timer(500);
const c = computed(() => $(a, 42) + _(b, 'hi!'));
// > 42 hi!
// > 0 0
// > 1 0
// > …

4. String tag tracker by @voliva & @loreanvictor:

const a = timer(0, 1000);
const b = tagger`${ a } 🦔`;
// > 0 🦔
// > 1 🦔
// > ...

Alternatives:

1. we can have tagger to be equivalent of a tracker with concatenation

2. string expression might be evaluated

--

Suggested APIs can co-exist with the original one, available via different names.

--

If you have an idea — please, add a comment describing it.

Suggestion: Trackers through arguments

This is a cool and fun idea indeed. Thanks for this!

I'd like to suggest an alternative API that maybe would make more explicit what $ and _ do, and make the implementation slightly easier.

Instead of importing $ and _ from the package, what if these two trackers could be passed as arguments to run? So instead of:

import { run, $, _ } from 'rxjs-autorun';

const a = new BehaviorSubject('#');
const b = new BehaviorSubject(1);
const c = run(() => _(a) + $(b));

You could:

import { run } from 'rxjs-autorun';

const a = new BehaviorSubject('#');
const b = new BehaviorSubject(1);
const c = run((track, read) => read(a) + track(b));

This would also let the consumer name $ and _ as they want (in this case, track and read respectively), but they could also go with a smaller t and r.

Why the distinctUntilChanged?

In commit 613224c, two distinctUntilLatest operators were added... I was wondering why that is..?
I have some use cases where it doesn't have any effect because it is used on reference or array types. And it doesn't perform a deep compare by default.
Also I could imagine some use cases where you don't want it to block non-changes. It could be a signal to update something regardless of whether it changed or not.
Also, a user can add it him/herself, even with deep compare, when needed...

Would it be a problem to remove it? Or is there a good reason to leave it there?

Cover multiple sync emissions

Currently, we are skipping multiple sync emissions. This will wait for all 3 values from a and only then would produce a result:

const a = of(1, 2, 3);
const b = of('🐁');
computed(() => $(a) + $(b)).subscribe(…); // > 3🐁

I think its fine for computed, but it seems to be an issue for our shiny new combined (or whatever name we end up with in #32):

combined(() => $(a) + $(b)).subscribe(…); // > 1🐁 > 2🐁 > 3🐁

Though it's not very clear how to handle this with multiple streams emitting synchronously.

combineLatest ignores sync values from the first stream:

const a$ = of(1,2,3);
const b$ = of('a', 'b', 'c');
const result$ = combineLatest(a$, b$);

---

So, I'm not sure if, what, and how it should be done.
Please, share your thoughts!

Allowing dependent observables?

Consider this scenario:

const myPresence$ = run(() => {
   const userId = $(loggedInUser$);
   return $(getPresence(userId));
});

At first sight, to me it looked like this is going to work well. But on second thought... What if the presence receives it's first value? The expression will run again. Hence it will call getPrecense(userId) again which might (probably) return a new observable. That new observable will be observed immediately. So when it emits synchronously it will end up in an infinite loop or yield a stack overflow!

So currently this lib doesn't support dependent observables like this. Question is, do we want to allow this? And if so, how would we do this?
My first thought was, detecting whether the argument of getPresence is still the same, and if it is, don't run the function but use the old value.
But I currently can't think of a way to check that, and also none to block running that function. Maybe require notation $(() => getPresence(userId)) or something? And include dependent arguments like $(userId, () => getPresence(userId))? Looks a bit ugly maybe... Any other ideas? Or just drop this issue?

Investigate tree-shaking compatibility

As reported by @loreanvictor in #1, tree shaking might not be compatible with exporting $ and _ that share context variable with the computed fn.

Needs investigation.

Release (unsubscribe) unused subscriptions

I figured it might be better to discuss this in a new issue :)

Here's a use-case that advocates for unsubscribing unused observables.

Suppose there's an observable userPresence$ which, when observed, starts a resourceful backend subscription on this users presence. There's also a button modelled by another observable that enables showing the user presence, e.g. presenceEnabled$. Out of those two you want to make a new observable that is used to display the current presence state. Following could be a naive implementation:

const displayPresence$ = combineLatest([
    userPresence$, presenceEnabled$
  ]).pipe(
    map(([presence, enabled]) => enabled ? presence : Presence.Disabled));

But this will not release the subscription resource when presence is disabled. So then you might want to write it like this:

const displayPresence$ = presenceEnabled$.pipe(
  switchMap(enabled => enabled ? userPresence$ : of(Presence.Disabled)));

This has the correct behavior. But I think it would be nicer when you could use this lib.

const displayPresence$ = run(() => $(presenceEnabled$) ? $(userPresence$) : Presence.Disabled);

Now it becomes a one liner that is easy to read and does exactly what I want which I think is very cool!
But in this scenario, currently it leaves the userPresence$ subscription open once presenceEnabled$ became true once (even after it becomes false again later). That means it never releases the costly presence subscription resource anymore.

So I would love to propose a feature that does not only 'late subscription', but also 'early unsubscription' :)

Do you think it is OK if I write a PR for this?

Originally posted by @Jopie64 in #3 (comment)

Renaming functions and-or adding aliases

While $, _ function names are short and are unobtrusive, I think we'll need more descriptive names.
Also, run function name doesn't reflect well what it does (as it doesn't immediately run the expression) and is not conventional.

Maybe autorun needs renaming too.

As suggested by @fkrasnowski run could be renamed to something like computed or derived or autopiped

$ and _ as indirectly indicated by @voliva could have additional export aliases, like track (watch, observe, or …) and read (silent, muted, sample, untracked, or …)

Let me know what you think

Document weak, normal, strong kinds of strength

We need to cover in docs behavior in logic branching and our .weak .normal and .strong APIs

Throwing primitive errors in expression emits another kind of error

computed(() => { throw 42; });

Currently:
Emits [TypeError: Cannot convert object to primitive value]

Expected:
Should error with 42

Complete/error tracked expression with inner observables

If underlying observables of tracked expression complete — the resulting observable should complete too, e.g:

const o = of(1);
const r = run(() => $(o));
r.subscribe({
  next: v => console.log(v),
  complete: () => console.log('DONE')
});

Current behavior

> 1

Expected behavior

> 1
> DONE

(same goes with errors)

Suggestion: in-place error throwing

Since autorun API lets users run code in synchronous manner,
it might make sense to allow handling errors right inside the expression:

computed(() => {
  try { return $(a); }
  catch(e) { return e + $(b); }
});

(I've shared this suggestion somewhere, created this issue to discuss further)

Maybe default subscription strength should be "Strong" instead of "Normal"?

Right now, for this simplistic example:

import { computed, $ } from 'rxjs-autorun';
import { interval } from 'rxjs';

const a = interval(1000);
const b = interval(500);

computed(() => $(a) % 2 === 0 ? 'A:' + $(a) : 'B:' + $(b))
.subscribe(console.log);

You get this output:

A:0
B:0
A:2
B:0
A:4
B:0
A:6
B:0
...

Playground

This means even for pretty simplistic cases users would need to take subscription strength (and the fact that their observables might regularly get unsubscribed) into account. So if the intent is to sort of hide that from users and only keep it as an option for advanced usage, I would recommend bumping default subscription strength.

P.S. to be fair the default strong output is also not that super intuitive:

A:0
B:0
A:2
B:2
B:3
B:4
A:4
B:6
B:7
B:8
A:6
B:10
B:11
B:12
...

But I suspect you can reason about it without knowledge of subscription strength. It is also notable that this scenario only happens in conditional expressions.