DEV Community: RxJS

Much Fun with RxJS-Autorun

Kostia Palchyk — Tue, 27 Oct 2020 15:12:43 +0000

Hi! 👋

Today I'll show you some tricks using rxjs and rxjs-autorun that we reviewed in the previous post. Let's quickly recap what it does:

you pass it an expression that uses some streams
whenever those streams emit, autorun runs the expression with new values
you get a stream of those evaluations results

Now let's go fast explore what we can do with it!

Mapping a single stream: 🌱 🌿 🌳

The most basic usage: we have a stream t of growing integer values: 0, 1, 2, etc (RxJS timer will create us one). And we want to substitute that integer with a string: "even" or "odd":

const t = timer(0, 1_000);
computed(() => $(t) % 2 ? 'odd' : 'even');
// Result:
// > even > odd > even > odd …

Explanation: $(t) will wait until stream t emits a value. Then with every value emission, computed will re-evaluate its expression, putting the result into the output stream.

Combinations: 📦 + 🎀 = 🎁

By tracking multiple streams inside expressions we can combine emissions from these streams:

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

const a = timer(0, 1_000);
const b = of('🐇');
computed(() => $(a) + ' - ' + $(b))
// > 0 - 🐇 > 1 - 🐇 > 1 - 🐇 …

Similarly to combining into strings, we can combine into Arrays:

computed(() => [ $(a), $(b) ])
// > [ 0, 🐇 ] > [ 1, 🐇 ] …

Objects:

computed(() => ({ a: $(a), b: $(b) }))
// > { a: 0, b: 🐇 } > { a: 1, b: 🐇 } …

or even call functions!

function toString(count, animal) {
  return (count < 3 ? 'some' : 'many') + animal;
}
computed(() => toString( $(a), $(b) ))
// > some 🐇 > some 🐇 … > many 🐇 …

Switching: 🌤 🌧

Now, imagine we have a function getWeather that returns an Observable with the weather forecast for a given city. Using the same ?: from the first example we could create a widget to show weather for two cities:

const t = timer(0, 1_000);
const nice = getWeather('Nice'); // => Observable<'Nice: 🌤'>
const cannes = getWeather('Cannes'); // => Observable<'Cannes: 🌧'>
computed(() => $(t) % 2 ? $(nice) : $(cannes));
// > Cannes: 🌧 > Nice: 🌤 > Cannes: 🌧 > Nice: 🌤 …

Tricks: 🐇 🎩

Now, remember I said that $(…) will wait for the stream to emit its first value before continuing evaluation. Let's do some tricks exploiting this rxjs-autorun feature!

Filtering

computed will wait for every observable to emit in order to produce a value. But what if we pass it a never emitting Observable: NEVER? Correct, this will pause emission until other tracked Observable emits:

const a = timer(0, 1_000);
computed(() => $(a) % 2 ? $(NEVER) : $(a) + ' even' );
// > 0 even > 2 even > 4 even …

Take while

Similarly to NEVER, we can exploit awaiting by passing it an Observable that completes never ever emitting a value: EMPTY. If an expression relies on a stream that we know won't emit a value — then surely the resulting Observable won't emit either! So computed completes:

const a = timer(0, 1_000);
const c = computed(() => $(a) < 5 ? $(a) : $(EMPTY));
// > 0 > 1 > 2 > 3 > 4 > COMPLETE!

NOTE: this feature is in beta-testing now and will be available with the next release

Bonus: 🙂

I showed you two functions: computed to run expressions and $ to track values. But $ has a sibling! A _ function, similarly to $, will read the current value of a stream but it won't trigger expression re-evaluation if that stream updates!

Now, imagine the examples above if you mix $ and _. Got you thinking, eh? Then imagine that all the computed streams can be used in other computed expressions!

🙂 → 😲 → 🤯

Outro: 🛸💨

You'll find the docs and the library here: github.com/kosich/rxjs-autorun

I hope you had fun! If you enjoyed reading — please, indicate that with ❤️ 🦄 📘 buttons — it helps a lot!

Follow me here and on twitter for more RxJS, React, and JS posts!

Thank you for reading this article! Stay reactive and have a nice day 🙂

Cya! 👋

Psst.. need something more to read?

I got you covered:

😉

RxJS Proxy: 3 new features

Kostia Palchyk — Fri, 23 Oct 2020 11:04:57 +0000

amazing header photo by Zach Kadolph

Hi! 👋 In the previous post, I've introduced you to rxjs-proxify that turns your Observable into an Object of Observables:

const o = of({ msg: 'Hello' }, { msg: 'World' });
const p = proxify(o);
p.msg.subscribe(console.log); // > Hello > World

Today I'll show you what new abilities you get with the recent 0.0.10 release: now proxify can be applied not only on Observables but also on Subjects and BehaviorSubjects. Also, we get a statify method that will help you manage the local state!

tl;dr try & install: github.com/kosich/rxjs-proxify

📖 Intro

Left to right: Observable, Subject, and BehaviorSubject. Cute photo by The Lucky Neko.

Observable Proxy (already familiar to you)
subscribe at any depth

const observable = proxify( of({ p: '🐑' }) );
observable.subscribe(console.log); // > { p: 🐑 }
observable.p.subscribe(console.log); // > 🐑

It will be handy if you need to access many sub-properties on your streams:

// framework agnostic model
const { title, body } = proxify( fetchArticle() );
// framework agnostic view
<article>
  <Title value={ title } />
  <Body value={ body } />
</article>

Note that here we're creating two separate Observables, so you'll probably want to .pipe( share() ) inside fetchArticle.

Subject Proxy
subscribe at any depth, push at the root

const subject = proxify(new Subject<{ p: string }>());
subject.subscribe(console.log);
subject.p.subscribe(console.log);
subject.next({ p: '🐥' }); // > { p: 🐥 } // > 🐥

While you have all the benefits of Observable Proxy, you can also update the whole Subject value.

BehaviorSubject Proxy

subscribe at any depth, push at any depth, synchronously read the current state

const behavior = proxify(new BehaviorSubject({ p: '🐖' }));
behavior.p.subscribe(console.log); // > 🐖
behavior.p.next('🐇'); // > 🐇
console.log(behavior.p.value) // > 🐇

In addition to Observable Proxy, you can synchronously read and update the state at any level!

// model
const s = proxify(
  new BehaviorSubject({ title: '', checked: false })
);
// view
<div>
  <input
    value={ s.title }
    onChange={ e => s.title.next(e.target.value) } />
  <input
    type="checkbox"
    checked={ s.checked }
    onChange={ e => s.checked.next(e.target.checked) } />
</div>

State proxy

And we also export a statify function that creates a BehaviorSubject Proxy with a distinctUntilChanged on all its properties:

// create a state
const state = statify({ a: '🐰', z: '🏡' });

// listen to & log root state changes
state.subscribe(console.log); //> { a:🐰 z:🏡 }

// update particular substate
state.a.next('🐇'); //> { a:🐇 z:🏡 }
state.a.next('🐇'); //> same value, no update

// read current values
console.log(state.z.value + state.a.value); //> 🏡🐇

// update root state, still logging
state.next({ a: '🐇', z: '☁️' }) //> { a:🐇 z:☁️ }

// and then…
state.z.next('🌙');   //> { a:🐇  z:🌙 }
state.a.next('🐇👀'); //> { a:🐇👀 z:🌙 }
state.z.next('🛸')    //> { a:🐇👀 z:🛸 }
state.a.next('💨');   //> { a:💨  z:🛸 }

That's it! Just a few words before you go:

🎉 Outro

You can try proxify online.

And you'll find more examples and the installation guide at github.com/kosich/rxjs-proxify. Hope you'll find these little tools useful!

If you enjoyed reading — please, indicate that with ❤️ 🦄 📘 buttons

Follow me here and on twitter for more RxJS, React, and JS posts!

Thank you for reading this article! Stay reactive and have a nice day 🙂

And thanks to @fkrasnowski for discussing with me and polishing this idea in the previous post comments!

RxJS Autorun Intro

Kostia Palchyk — Thu, 15 Oct 2020 12:01:35 +0000

Hey, RxJS streamers! 🙋‍♂️

Today we're going to review a small library that re-evaluates an expression based on updates in streams it uses.

tl;dr: docs and package at github.com/kosich/rxjs-autorun 🔗

Let's boldly explore it!

It's a sweetened countdown

Our first example.

Say, we want to prettify each value on a timer stream. So we'll write such an expression:

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

// timer would emit every second
const a = timer(0, 1_000);
// expression would concat every emission with a cake
const result = computed(() => $(a) + ' 🍰');
// now we should subscribe to the resulting stream
result.subscribe(x => console.log(x));
// > 0 🍰
// > 1 🍰
// > 2 🍰
// > …

Explanation: computed takes a function that uses some streams and re-evaluates it when those streams update. It returns an observable that you can manipulate further. And $(a) indicates that a is a stream and its updates should be listened to.

So technically this expression is equivalent to a.pipe( map(x => x + '🍰') )

But let's keep discovering what else this tiny lib can do:

Infinite monkey theorem needs infinite bananas

Here we'll combine a timer that would represent a queue of our little 🐒 fellas with a stream of fetched bananas 🍌:

import { timer, of } from 'rxjs';
import { delay } from 'rxjs/operators';
import { computed, $ } from 'rxjs-autorun';

const a = timer(0, 1_000); // get some monkeys
const b = of('🍌').pipe(delay(2_000)); // fetch some treats
const result = computed(() => '🐒 #' + $(a) + ' gets ' + $(b)); // mix
result.subscribe(x => console.log(x)); // listen to result stream
// ...2s gap...
// > 🐒 #1 gets 🍌
// > 🐒 #2 gets 🍌
// > 🐒 #3 gets 🍌
// > …

BTW, did you know that banana is technically a berry?

Not hard at all, is it?

This expression is similar to combineLatest(a, b).pipe( map(([x,y]) => x + y) ).

Let's review another multiple streams example:

Who's up to some pizza?

The last trick we're gonna learn today is ability to read latest values without tracking their updates:

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

const a = new Subject(); // neighbours
const b = new Subject(); // food
computed(() => $(a) + ' likes ' + _(b))
  .subscribe(x => console.log(x));
a.next('🐈'); // nothing: b is still empty
b.next('🥛'); // > 🐈 likes 🥛
a.next('🐭'); // > 🐭 likes 🥛
b.next('🍕'); // nothing: _(b) doesn't trigger re-runs
a.next('🙋‍♂️'); // 🙋‍♂️ likes 🍕

Explanation: _ function indicates that we need to take one value from a stream, but we don't want to recalculate our expression when this particular stream emits. So if an expression uses $(a) and _(b) — it would react only to a updates.

This also means that computed(() => _(a)) expression would emit one value and immediately complete.

Okay, one really last thing before we wrap-up:

Transformation

This time, try to guess what it is:

import { timer, of } from 'rxjs';
import { computed, $, _ } from 'rxjs-autorun';

const a = timer(0, 1_000);
const b = of('💧');
const c = of('❄');
const result = computed(() => $(a) % 2 ? _(b) : _(c));
result.subscribe(x => console.log(x));

Indeed, this is capricious weather 🙂

Actually, this expression is somewhat similar to switchMap

Outro 😳

All the examples you can try here.

And there's more to the library, go explore it yourself!

In the following articles we'll review how to filter emissions and how to manage subscriptions inside rxjs-autorun expressions. Not to miss those and other RxJS posts — follow me here and on twitter!

If you enjoyed reading — please, indicate that with ❤️ 🦄 📘 buttons — it helps a lot!

Thank you for reading this article! Stay reactive and have a nice day 🙂

Also I want to thank @fkrasnowski for lengthy discussions of this idea, @ryansolid for giving it a try and Johan for collaborating with me on this! 🙏

Psst.. need something more to read?

I got you covered:

😉

Turn a Stream of Objects into an Object of Streams

Kostia Palchyk — Fri, 18 Sep 2020 14:40:56 +0000

Hi, fellow RxJS streamer! 👋

Today I want to share a JS/TS package that allows you to access props of objects on Observables:

source$.subscribe(o => console.log(o?.a?.b?.c))
// turn ↑ into ↓
source$.a.b.c.subscribe(console.log)

tl;dr: github.com/kosich/rxjs-proxify

A simple use case: read the msg property of each value on the stream

import { proxify } from "rxjs-proxify";
import { of } from "rxjs";

const source = of({ msg: 'Hello' }, { msg: 'World' });
const stream = proxify(source);
stream.msg.subscribe(console.log); // 'Hello', 'World'

☝️ proxify will create a Proxy for given Observable

The package has good TypeScript support, so all props are intelli-sensed by cats, dogs, and IDEs:

import { of } from 'rxjs';
import { proxify } from 'rxjs-proxify';

const source = of({ a:1, b:1 }, { a:2, b:2 });
const stream = proxify(source);
stream. // <- will suggest .a .b .pipe .subscribe etc

👀 I can see your intentions

t's also possible to call methods on values (even those using this keyword), e.g.:

import { proxify } from "rxjs-proxify";
import { of } from "rxjs";

const source = of({ msg: () => 'Hello' }, { msg: () => 'World' });
const stream = proxify(source);
// calls msg() fn on each value of the stream
stream.msg().subscribe(console.log); // 'Hello', 'World'

🤯 pure magic, I tell you

And you are still free to apply RxJS operators at any depth:

import { proxify } from "rxjs-proxify";
import { of } from "rxjs";
import { scan } from "rxjs/operators";

const source = of({ msg: 'Hello' }, { msg: 'World' });
const stream = proxify(source);
stream.msg.pipe(scan((a, c)=> a + c)).subscribe(console.log); // 'HelloWorld'

Just like regular Observables!

The package uses Proxies under the hood, recursively applying it to sub-properties and method results, so the chain can be indefinitely deep. And you can apply .subscribe or .pipe at any time!

🎹 Try it

You can install it via npm i rxjs-proxify

Or test it online: stackblitz.com/edit/rxjs-proxify-repl

📖 Repository

The source code and more examples are available on github repo:

github.com/kosich/rxjs-proxify

Outro

Thank you for reading this article! Stay reactive and have a nice day 🙂

If you enjoyed reading — please, indicate that with ❤️ 🦄 📘 buttons — it helps a lot!

Soon I'll post a more detailed review of the lib and how it works

Follow me here and on twitter for more RxJS, React, and JS posts:

// Detect dark theme var iframe = document.getElementById('tweet-1302684300366798849-749'); if (document.body.className.includes('dark-theme')) { iframe.src = "https://platform.twitter.com/embed/Tweet.html?id=1302684300366798849&theme=dark" }

🗣 Would love to hear your thoughts!

Psst.. need something more to read?

I got you covered:

😉

Cya 👋

Fetching Data in React with RxJS and <$> fragment

Kostia Palchyk — Tue, 04 Aug 2020 08:49:33 +0000

We often need to fetch data in our components. Here's an example using useState hook and fetch API to get and display some data:



function App(){
  const [data, setData] = useState(null);

  // useEffect for fetching data on mount
  useEffect(() => {
    fetch('//...')
    .then(response => response.json())
    .then(data => setData(data));
  }, []);

  return <div>Data: { data }</div>
}

here and further I'll drop checking response.ok for brevity sake

Looks alright?

Well, this approach lacks few important features:

cancelling fetching on component unmount (e.g if user leaves current page)
handling errors
displaying loading indicator

To handle all these issues nicely we'll use RxJS!

RxJS is a very mighty tool to manage and coordinate async events (like fetching and UI events). Learning it will pay you back 10 fold!

Please, don't get freaked out now, I'll walk you through adding and using it 🙂

tl;dr: resulting app playground and <$> fragment library

Let's start with updating our App to use RxJS!

🔋 Power Up

First we'll switch to RxJS' fromFetch — it's a wrapper around native fetch:



function App(){
  const [data, setData] = useState(null);

  useEffect(() => {
    fromFetch('//...')
      .subscribe(response =>
        response.json().then(data => setData(data))
      );
  }, []);

  return <div>Data: { data }</div>
}

.subscribe method is an analogue for .then in Promises — it will receive value updates from the RxJS stream (currently it will handle only one update, but there'll be more)

Also .subscribe returns an object with which we can cancel the "subscription". This will help us solve our first issue: cancelling fetching on component unmount.



function App(){
  const [data, setData] = useState(null);

  useEffect(() => {
    const subscription = fromFetch('//...')
      .subscribe(response =>
        response.json().then(data => setData(data))
      );

    // this function will be called on component unmount
    // it will terminate the fetching
    return () => subscription.unsubscribe();
  }, []);

  return <div>Data: { data }</div>
}

See React's useEffect#cleaning-up-an-effect docs section for details

Hurray: 1 done, 2 left!

Let's do a small cleanup before we go further:

🔧 Refactoring and <$> fragment

As you can see, we're using response.json() async operation inside our subscribe function — this is a bad practice for a number of reasons: this stream would not be reusable and cancellation wont work if we're already on stage of response.json() parsing.

We'll use a mergeMap RxJS operator to fix this:



function App(){
  const [data, setData] = useState(null);

  useEffect(() => {
    const subscription = fromFetch('//...')
      .pipe(
        // mergeMap is an operator to do another async task
        mergeMap(response => response.json())
      )
      .subscribe(data => setData(data));

    return () => subscription.unsubscribe();
  }, []);

  return <div>Data: { data }</div>
}

UPD: @benlesh made a good point that one can use RxJS' ajax.getJSON instead of fetch wrapper, and skip the mergeMap. E.g.: ajax.getJSON(url).subscribe(/* etc. */). I will keep the fromFetch approach for educational and laziness reasons 🙂

We've separated response.json() operation from results handling. And with our subscribe handler only responsible for displaying data — we can now use <$> fragment!

<$> — is a small (1Kb) package to display RxJS values in our React components.

It will subscribe to provided stream for us and display updates in place. And also unsubscribe on component unmount, so we won't need to worry about that too!



function App(){
  // we need useMemo to ensure stream$ persist
  // between App re-renders
  const stream$ = useMemo(() =>
    fromFetch('//...')
      .pipe(
        mergeMap(response => response.json())
      )
  , []);

  return <div>Data: <$>{ stream$ }</$></div>
}

Note that we've dropped useState and .subscribe: <$> does all that!

So, we're ready to add more operators to continue solving our tasks. Let's add a loading indicator!

⏳ Loading indicator



function App(){
  const stream$ = useMemo(() =>
    fromFetch('//...')
      .pipe(
        mergeMap(response => response.json()),
        // immediately show a loading text
        startWith('loading...')
      )
  , []);

  return <div>Data: <$>{ stream$ }</$></div>
}

startWith will prepend async data stream with provided value. In our case it looks somewhat like this:



start -o---------------------------o- end

       ^ show 'loading'            ^ receive and display
       | immediately               | response later

Awesome: 2 done, 1 left!

We'll handle errors next:

⚠️ Error handling

Another operator catchError will let us handle error from fetching:



function App(){
  const stream$ = useMemo(() =>
    fromFetch('//...')
      .pipe(
        mergeMap(response => response.json()),
        catchError(() => of('ERROR')),
        startWith('loading...')
      )
  , []);

  return <div>Data: <$>{ stream$ }</$></div>
}

Now if fetching fails — we'll display 'ERROR' text.

If you want to dig deeper, I wrote a detailed article on managing errors: "Error handling in RxJS or how not to fail with Observables" — suppressing, strategic fallbacks, retries simple and with exponential delays — it's all there.

3 done, 0 left!

Let's finalize with moving some divs around:

🖼 Better UI

Most likely we'd like to show properly highlighted error and styled (maybe even animated) loading indicator. To do that — we'll simply move our JSX right into the stream:



function App(){
  const stream$ = useMemo(() =>
    fromFetch('//...')
      .pipe(
        mergeMap(response => response.json()),
        // now we'll map not only to text
        // but to JSX
        map(data => <div className="data">Data: { data }</div>),
        catchError(() => of(<div className="err">ERROR</div>)),
        startWith(<div className="loading">loading...</div>)
      )
  , []);

  return <$>{ stream$ }</$>
}

Note that now we can fully customize view for each state!

🍰 Bonus: anti-flickering

Sometimes if the response comes too quickly we'll see the loading indicator flash for a split second. This is generally undesirable since we've worked long on our loading indicator animation and want to ensure user sees it through 🙂

To fix this we'll split out fetching Observable creation and join the fetching with a 500ms delay:



function App(){
  const stream$ = useMemo(() =>
    customFetch('//...').pipe(
        map(data => <div className="data">Data: { data }</div>),
        catchError(() => of(<div className="err">ERROR</div>)),
        startWith(<div className="loading">loading...</div>)
      )
  , []);

  return <$>{ stream$ }</$>
}

function customFetch(URL) {
  // wait for both fetch and a 500ms timer to finish
  return zip(
    fromFetch(URL).pipe( mergeMap(r => r.json()) ),
    timer(500) // set a timer for 500ms
  ).pipe(
    // then take only the first value (fetch result)
    map(([data]) => data)
  )
}

Now our loved user will see the loading animation for at least 500ms!

4 done, 🍰 left!

A few final words:

🎉 Outro

Here's our resulting app if you want to play around with it.

To start using RxJS in your React components just do:



npm i rxjs react-rxjs-elements

And then drop a stream inside <$>:



import { timer } from 'rxjs';
import { $ } from 'react-rxjs-elements';

function App() {
  return <$>{ timer(0, 1000) } ms</$>
}

That's it, I hope you've learnt something new!

Thank you for reading this article! Stay reactive and have a nice day 🙂

If you enjoyed reading — please, indicate that with ❤️ 🦄 📘 buttons

Follow me on twitter for more React, RxJS, and JS posts:

// Detect dark theme var iframe = document.getElementById('tweet-1285580447175643138-374'); if (document.body.className.includes('dark-theme')) { iframe.src = "https://platform.twitter.com/embed/Tweet.html?id=1285580447175643138&theme=dark" }

The End

Thanks to @niklas_wortmann and @sharlatta for reviewing!

Observables, Reactive Programming, and Regret

Ben Lesh — Mon, 29 Jun 2020 21:16:20 +0000

As of this writing, I've been working on the RxJS project for almost 6 years, I think. When I started out, I really had no idea what I was getting into (And I wouldn't have been able to ship those first versions without Paul Taylor and others, for sure). I can remember looking at the number of weekly downloads on npm, and being able to figure out how exactly how many of them were mine. Fast forward to today, RxJS and observables have become extremely popular... Loved by many, hated by a few, and my fear is, misunderstood by most.

Observables vs Reactive Programming vs RxJS

A big problem I see nowadays is how observables have now been intimately linked, for better or worse, to RxJS. Looking back, I sort of wish we had published the Observable primitive as a separate package, and the operators in another package.

When this effort started, I was naively optimistic that Observable would land in the ECMAScript standard, and RxJS would just "become a collection of helper functions" as I think I put it. But years passed, and the TC39 proposal stalled. And the way the world came to know Observable was through RxJS.

Observables are not RxJS. Observables do not require "operators". They are a primitive. The "dual" of the Iterable. A simple push-based type. Nothing more.

Reactive programming isn't necessarily observables. Reactive programming is a paradigm or a practice. It can be done with functions, Promises, etc. In essence, if you can compartmentalize your code into functions that will "react" to incoming events without knowing anything about the source, congrats, you're "reactive".

RxJS is a library of functions built around observables, not the other way around. Observables can, and do, exist in the wild without RxJS. They show up in other libraries, often times in slightly different shapes, but the overall concept is the same. Facebook's Relay has an internal Observable implementation that is eerily similar to RxJS's implementation. In fact, I've lost count of the number of times I've seen an abstraction that amounts to an interface that accepts a callback to handle multiple values, an error, or a completion, and returns or otherwise uses some sort of cancellation semantic.

Regrets

1. The huge API

RxJS 5 inherited its HUGE API surface area from RxJS 4 and under. RxJS 4 and under, in turn, inherited it's API from RxNET, many, many years ago. So much of the API that some might deem "unnecessary" exists because "it always has been, and always must be". RxJS 5 might have been our only chance in the history of the library to truly ween that down. Which we did a bit, but probably not enough. The large API surface leads to confusion and loathing in the community. All of which is understandable, IMO.

2. RxJS out-shined Observable

Observables never had a chance to shine on their own. The real win, IMO, to RxJS is the Observable type itself. Not the operators. Those are just fluff that allow you to do some cool things. Having a lazy type with guarantees like Observable is actually a bigger deal.

With Observable you're guaranteed:

Once it's complete, errored, or unsubscribed, you will get no more messages
Registered teardown WILL occur. If you complete, error, or unsubscribe, you are guaranteed to clean up resources.
A unified API that can represent a wide variety of things: Events, multiple values, single values, user interactions, streaming data, synchronous values, asynchronous values, etc. etc.

There are other great advantages to its design. But IMO, those are the biggest.

RxJS and all of its operators are inseparable in some people's heads from observables. And that's a real shame. Observable is a simple thing. A very simple type. RxJS is complicated with it's huge API and odd names.

3. We never really outlined where RxJS would best serve people

Disclaimer: These are MY opinions on RxJS/Observable use, and not really the RxJS core team's. Feel free to use RxJS or whatever library in whatever way you see fit. If it works, you can maintain it, and you can test it, IMO it's good code. The end.

To put it simply, once people get into RxJS, it's an exciting technology. It suddenly gets used for everything. It's fair to say this mentality exists in tech for a lot of libraries and frameworks. But I think with RxJS it becomes insidious to the detriment of the RxJS community.

Examples:

You have a button that, when clicked, fetches the latest data and displays it. Do you need full-on RxJS? No, probably not. "But what about cancellation???" .. You wanted an observable. Not operators. You can use RxJS here for the Observable implementation, but I would caution against jumping into concatMap et al. Especially if your team isn't used to RxJS. But that doesn't mean you shouldn't use Observable. In fact, you probably should.
You have streaming data over a web socket, and you need to split it into a couple of different streams and update two parts of your UI. Yes! This is what RxJS is for. You're a filter operator away from a solid use case.
You have complex async coordination and/or race conditions, even with APIs that return promises? Honestly, you might want to use RxJS here as well, because of guarantees provided by Observable, and useful operators like concatMap that can guarantee ordering, etc, and have complete interop with async/await and Promise.

4. We never taught people how to write readable code with RxJS

We handed people powerful tools and let them go at it. No guidance or experienced wisdom provided with how to effectively use the library so that you didn't drive your coworkers crazy. This is sort of like getting a power tool set with no manuals. How do you maintain it? How do you resolve issues? Where do you store the tools? etc.

The result of this is people write code they don't understand when they revisit it. Most amazingly, some engineers, who are usually a rational bunch, then declare RxJS to be "unreadable", as in, no matter what they did, they could never make the code readable. Seems defeatist to me. Like anything else, good practices and strategies around reading and organizing rxjs code can be learned and taught. But I know that I personally haven't done enough to spread this know-how.

Consequences

For the most part, I think the response to RxJS has been overwhelmingly positive. The community has organized a conference. I've seen a lot of discussion about it across many communities (beyond just Angular). And usage has been steadily growing.

But on the back swing, there is a trail of destruction to the reputation of RxJS and Observable that has been wrought by misunderstandings about Observable and RxJS, and misuse of the library in general, IMO. There have been well-known tech personalities who have called out "wishing RxJS didn't exist". And my fear is that sort of thinking, if it spreads, will spell doom for the Observable type itself. That would be the biggest shame to this, honestly.

The Observable itself is huge win. It's a primitive that, like I said above, shows up in many forms in many places, and I think it deserves a spot in the language as much as Iterable and Promise. People having a distaste for RxJS's API and/or abuse and misuse is completely understandable, IMO.

There are parts of RxJS I don't like, and here I am unable to pivot the library quickly because it's so popular we'd simply break too many people. But the parts I like the most, the Observable itself, and the guarantees it provides, are in jeopardy of being thrown out with the bath water by some folks. And that's tragic, IMO.

The road forward

For my part, I plan on trying to continue to champion promoting understanding of the when/where/why of RxJS and Observable. And I want to do better to disambiguate Observable from RxJS. I also want to work very hard to simplify the RxJS API: Tighten the API, remove what does not need to be there, improve documentation and readability, add more guidance for folks on how to make their code more maintainable, etc.

Don't get me wrong, I have other regrets with regards to RxJS as it stands, but I'm confident we can remedy all of those things over time. My deepest concern is that there are a huge numbers of people that still don't understand the Observable primitive and its benefits, because they associate it with RxJS and are standoffish about getting involved there because of the learning curve.

RxJS debounce vs throttle vs audit vs sample — Difference You Should Know

Kostia Palchyk — Fri, 19 Jun 2020 14:33:01 +0000

In this article, we'll review how these operators work and how they differ.

For simplicity, we'll compare their *Time counterparts: auditTime, debounceTime, throttleTime, sampleTime — they work in the same way, just in defined time windows.

These operators are handy when you want to lower the load on the consumption function side. For example, you don't want to react to every user mousemove or every input.change event. In their own way, they will let you filter input stream based on time intervals. So let's see how they differ!

We'll start with investigating this marble diagram:

You can play with this marble diagram here

Explanation: here we apply all four operators to the source stream. The source starts emitting at time:0 and completes at time:1000. In the applied operators the colors are preserved from the source, but the value is substituted with the actual time of emission in resulting stream.

As you can see:

debounceTime will emit a value from the source stream only if a given time has passed without source producing more values
throttleTime will start a timer when the source emits. It can be set to emit the first and/or the last value in the given time window. Then it repeats this procedure
auditTime behaves in a similar way to the trailing throttleTime, but note that it won't emit a value from the last time window if the source has completed
sampleTime simply emits a value from the source in a given time window if the source actually emitted

I believe in learning by playing, so head to this marble diagram REPL to try changing code yourself: thinkrx.io/rxjs/debounceTime-vs-throttleTime-vs-auditTime-vs-sampleTime

Outro

That's it, I hope you had fun! If you enjoyed reading — please, indicate that with ❤️ 🦄 📘 buttons — it helps a lot!

Follow me here and on twitter for more RxJS, React, and JS posts!

Thank you for reading this article! Stay reactive and have a nice day 🙂

Cya! 👋

Psst.. need something more to read?

I got you covered:

😉

“RxJS Cheatsheet” VS Code extension

Dzhavat Ushev — Mon, 18 May 2020 14:21:43 +0000

This post was originally published on my blog.

So, I’m back with another VS Code extension! 🚀

I’m really excited to share “RxJS Cheatsheet”. 🎉

Here’s a demo:

Learning RxJS isn’t easy. There are Observables, operators, Subjects, subscriptions, marble diagrams, etc, which can be quite hard to understand when first starting out. Thankfully, there are many resources for learning RxJS. They are all great. Use them. This extension is yet another resource. It’s available right in your code editor and its purpose is to give you quick information about RxJS related stuff.

While it’s really fun working on the extension, I think it’s wise to test it with real users as early as possible. Therefore I’ve decided to publish a very minimal version in order to get feedback from the community.

In this first version, you’ll see only a handful of Observable-creation functions and operators. I’ve tried to include the most frequently used. Each item contains information about the import path and a marble diagram. There are other things worth including but I’ve kept those for later.

After installing the extension, you can open it by running the “Open RxJS Cheatsheet” command.

If you work with RxJS give it a try and let me know what you think. All ideas and suggestions for improvements are welcome! 🙂

I hope you will find it useful!

New in RxJS v7: concatWith operator

Dzhavat Ushev — Sat, 11 Jan 2020 16:25:24 +0000

This post was originally published on my blog.

RxJS v7-alpha.1 was released a couple of weeks ago and while it’s exciting to see v7 slowing coming along, this release came with only one new feature. It’s an operator called concatWith.

In this short post I’m going to show you how this operator works.

Unfortunately the only available documentation about concatWith right now is in the source code. Here’s how it works:

Emits all of the values from the source observable, then, once it completes, subscribes to each observable source provided, one at a time, emitting all of their values, and not subscribing to the next one until it completes.

Let’s translate this to a simple example (StackBlitz):

import { of } from 'rxjs';
import { concatWith } from 'rxjs/operators';

const source = of('RxJS', 'is').pipe(concatWith(of('awesome!')));

source.subscribe(x => console.log(x));

// Output:
// RxJS
// is
// awesome!

If you’ve used the concat operator before this might look somehow familiar to you. This is not a coincidence. concatWith is actually not a completely new operator. It’s only meant to replace the concat operator which is currently marked as deprecated and will be removed in v8.

There’s one subtle difference between the two, though. concatWith only accepts inputs of type ObservableInput, whereas concat can also take a scheduler.

If you want to schedule the observable provided to concatWith, you need pass it to the scheduled function and also specify a scheduler. (StackBlitz)

Hope you learned something new. Happy RxJS-ing!

Thanks to Jan-Niklas Wortmann and Lars Gyrup Brink Nielsen for the review.

Getting started with contributing to RxJS docs

Dzhavat Ushev — Fri, 03 Jan 2020 18:06:23 +0000

This post was originally published on my blog.

I use RxJS a lot during my work. One great place to learn more about it is, of course, the official docs. I visit them multiple times a day to look up things. This includes stuff like operators, Observable creation functions, descriptions, marble diagrams, signatures, examples and so on. The examples in particular are quite useful because they help me understand how something is working in practice.

By looking at all that content, I sometimes notice small things that can be improved. It might be a missing dot, a typo, a duplicated word, an outdated example, an extra import, you name it.

I call these “small opportunities”. They give me a reason to get involved with the project and give something back to the community. Sending a fix takes just a few minutes but it improves the state of the docs for everyone. And I certainly feel happy when my pull request (PR) gets merged!

Are you motivated to contribute to RxJS, but feel overwhelmed by it and don’t know where to start? The docs are a good place. Have you come across something recently that caught your attention? Maybe that’s your “small opportunity”! Did you consider sending a fix? If not immediately then later, when you have a few spare minutes. It doesn’t take long.

To help you with that, here’s a quick guide on how to get started.

Running the docs locally

Fork the project and clone it to your machine.
Open a PowerShell/Terminal window and navigate to rxjs/docs_app folder.
Run npm run setup to install the dependencies and generate the docs.
Run npm run start to start a development server.

At this point, you can open the docs by navigating to http://localhost:4200 in your browser. Nice!

Now on to finding the thing you wanted to improve...

Where to make your change?

The RxJS docs are a combination of two things:

The API docs which are generated directly from source code. The source code lives under rxjs/src. If you want to change something related to stuff like of, from, switchMap, etc., look here. The actual documentation is in the form of a comment, containing JSDoc tags and some content.
General content like guides, tutorials, marketing materials, etc. This content is spread across different Markdown files each representing a separate page. The files live under rxjs/docs_app/content. Pages like overview, installation, migration, deprecations, etc., are part of it.

Note: There’s also a rxjs/docs folder. This one contains the old documentation. Don’t bother changing anything in there.

Making your change

Now that you know how to run the docs locally and how to identify where to make your change, it’s time to open the file and take action.

This should be straightforward since your goal is to fix something very specific. So go ahead and put that dot, close that bracket, fix that misspelled word, update that example, rephrase that sentence, add that extra line. You can do it!

Previewing your change

Depending on your change, you might want to see how it looks in the docs. Here’s what I usually do:

Stop the dev server, if it’s running.
Run npm run docs to re-generate the docs.
Run npm run start to start the dev server again.
Open http://localhost:4200 in the browser.

If this sounds annoying, there’s an npm run serve-and-sync command that watches the docs for changes, auto-generates them and runs the dev server all at the same time. Unfortunately this command doesn’t work for me that’s why I use the steps above.

Preparing to commit

Once you’re done with your changes, it’s a good idea to run a lint check to make sure the build will compile. This is especially important for changes related to the API docs.

Do that by running npm run lint from the root rxjs folder. If you don’t get any errors, you’re good to go.

Making a Pull Request (PR)

The lint passes and docs look just about right?

Now create a new branch and commit your changes. Push the branch to your fork. Finally, go to either your fork or RxJS’ repo on GitHub and follow the notification to open a PR.

Congrats! 🎉

Thanks for taking the time to contribute! You’re making the docs a better place not only for yourself but for everyone else as well. The docs are getting better because of contributions like yours.

You don’t have to fix nasty bugs in order to contribute with something meaningful! So go ahead make that small fix you’ve been thinking about. It’s worth it! And it’s appreciated too!

Research on Reactive-Ephemeral-State in component-oriented frameworks

Michael Hladky — Fri, 20 Dec 2019 18:12:38 +0000

Studies are done with Angular as an example for a component-oriented framework and RxJS is used as an example for a reactive programming library

As this is too much text I'm afraid the important things at the end will get lost,
I put it here and quote one of Richard Feynman's rules he stuck to when teaching:

Give credit where it's due
Richard Feynman

@ngrx_io - that listened to my questions and gave me useful feedback
@yjaaidi and @niklas_wortmann - may be the only 2 persons on earth that read ALL THAT
@juristr - that pretended he will use it in his projects
@mvmusatov - that sent a PR and fixed my messy demo :D

In most of the component-oriented applications, there is the need to structure container components.
Even after a well thought refactoring into more display components and grouping logic into responsibilities it's always hard to handle.

The data structure you manage inside these components is only here for their very component. Not for any other components.
This data structure appears with the component and disappears when the component is removed.

This is a good example of an ephemeral state.

If you have a well thought and structured approach on how to manage ephemeral state such components get a breeze to write.
You could master a fully reactive architecture in a scalable and maintainable way.

This article provides you with some fundamental information about my findings in reactive ephemeral state management.
It is applicable to every framework that is component-oriented and has some life cycle hooks for creation and destruction.

The below examples are done with Angular as a framework as it has DI built-in which comes in handy here.
As a Reactive programming library with cold Observables by default, I picked RxJS as it is well supported.

Table of Content

TL;DR
Methodology
Layers of State
- What is the ephemeral state?
  - Global vs Local Accessibility of Data Structures
  - Static vs Dynamic Lifetime of Data Structures
  - Global vs Local Processed Sources
  - Recap Ephemeral State
Problems to Solve on a Low-Level
- Timing
- Subscription Handling
- Sharing State and State Derivations
  - Uni and multi-casting with RxJS
  - Sharing Work
  - Sharing Instances
- The Late Subscriber Problem
  - Cold Composition
- Subscription-Less Interaction with Component StateManagement
  - Subscription-Less Handling of Side-Effects
- Recap Problems
Basic Usage
- Service Design
- Service Usage
Summary

TL;DR

If you are into reactive programming you will learn about some cool topics.

unicast vs. multicast
hot vs. cold
subscription-less components
higher-order operators like mergeAll

If you are also into ephemeral state management you can learn how to detect it:

We defined 3 rules of thumb to detect ephemeral state

No horizontal sharing of state
The lifetime of the state is dynamic
It processes local relevant events

Your understanding of the fundamental implementation will get a good boost!
Examples on how to introduce different architecture patterns in your component structure are only the tip of the iceberg:

Initiation and coupling state to e.g. a component
Interaction
derivation of state

are just some of the nitty-gritty details included!

Here the Important resources:

Recording (🎥 Live Demo at 24:47):
Repository For Examples (💾 Final Example):

BioPhoton / research-reactive-ephemeral-state-in-component-oriented-frontend-frameworks

Research on reactive-ephemeral-state in component oriented frontend frameworks - Demonstrated with Angular and RxJS
Sourcecode: 📦 ngx-rx/rxjs-state
NPM Package: 📦 ngx-rx-state

Methodology

If you go back in history you will find almost all our nowadays "cutting edge problems" already solved.
When I realized the first time that life is a "constant evolutionary repetition" I started to change my strategy on solving problems.

Before I almost always started to implement a half-backed cool idea which I was certain is most up to date with technologies.

After I made many mistakes (without them I would not be here today) and got some incredibly helpful insights, I now started to change my way of solving problems completely.
Let me quote The Gang Of Four to give you the first glimpse of my fundamental changes in how I approach problems:

So here is what the gang of four says about Object-Oriented-Software-Design-Patterns:

If you stick to the paradigms of object-oriented programming,
the design patterns appear naturally.

With that in mind, I did not work on a solution for ephemeral state management,
but tried to look at all the different problems that we will face when managing any local state in general, with the hope, in the end, a solid solution will appear naturally.

Let's start with the first chapter and some general information to get you on track.

Layers of State

Of course, there are WAY more, but in this article, I will introduce 3 layers of state:

(Persistent) Server State
Persistent Client State (Global State)
Ephemeral Client State (Local State)

Persistent Server State is the data in your database. It is provided to the consumer over a data API like REST, GraphQL, Websocket, etc.
This is very different from Meta State, which is information related to the status of a resource that provides us a state. E.g. Loading, Error, Success, etc.

For persistent and ephemeral client states I will try to use to more simpler wording.
I will use Global State for persistent client state and Local State of the ephemeral client state.
Both live on the client, but they demand a completely different way of treatment.

In this article, I want to focus on the ephemeral state.

What is the ephemeral state?

The ephemeral state is just one of many names for data structures
that needed to be managed on the client under special conditions.
Other synonyms are a transient state, UI state, local state, component state, short term data, etc...

It is the data structure that expresses the state
of an isolated unit like for example a component in your application.

As the word "isolated" is a bit vague, let me get a little bit more concrete.

It's the state that lives in your components, pipes, directives and some of the services that are created
and destroyed over time. The state is not shared between siblings and not populated to global services.

Global vs Local Accessibility of Data Structures

The term global state is well known in modern web development.
It is the state we share globally in our app e.g. a @ngRx/store or the good old window object ;)

This is in this article called persistent state.

As we can see one global source distributes state to the whole app.

If we compare this to a local state we see that this data structure is provided and managed only in a certain time-frame of your app.
For example in a component or directive.

This is our first rule of thumb to detect local state:

No horizontal sharing of the state e.g. with sibling components or upwards.

Static vs Dynamic Lifetime of Data Structures

In Angular global state is nearly always shared over global singleton services.
Their lifetime starts even before the root component. And ends after every child component.
The state's lifetime is ~equal to the Apps lifetime or the browser windows lifetime.

This is called a static lifetime.

If we compare this to the lifetime of other building blocks of Angular we can see their lifetime is way more dynamic.

State in this building blocks is tied to the lifetime of their owners, their hosts and if shared this state is shared then only with children.

The best example of a dynamic lifetime is data that gets rendered over the async pipe.

The lifetime depends on the evaluation of the template expression, a potential *ngIf that wraps the expression or e.g a directive.

For our second rule of thumb we detected for the local state is:

The lifetime is dynamic e.g. bound to the lifetime of a component or an async pipe

Global vs Local Processed Sources

Where our global state service nearly always processes remote sources:

REST API's HTTP, HTTP2
Web Sockets
Browser URL
Browser Plugins
Global Static Data
The window object

And the logic is located in the more abstract layers of our architecture.

Code dedicated to the local state would nearly always focus on the process of the following sources:

Data from @InputBindings
UI Events
Component level Side-Effects
Parsing global state to local

The third rule of thumb to detect local state is:

It processes mostly local sources e.g. sort/filter change

Recap Ephemeral State

We defined 3 rules of thumb to detect ephemeral/local state

No horizontal sharing of state

The lifetime of the state is dynamic

It processes local relevant events

Some real-life example that matches the above-defined rules are:

sorting state of a list
form errors
state of an admin panel (filter, open/close, ...)
any dynamic appearing data
accumulations from @Input data
extended global state or derived global state for a container component

You rarely share this data with sibling components, it only shares data-structures only locally and focuses mostly on local sources.
In other words, there is no need to use many of the concepts of global state management libraries e.g. actions.

Still, we need a way to manage these data structures.

Problems to Solve on a Low-Level

As a first and foundational decision fact, we have to know we work with a push-based architecture. This has several advantages but more important defines the problems we will run into when implementing a solution.

As we defined the way how we want to distribute our data let me list a set of problems we need to solve.

Timing

As a lot of problems I ran into in applications are related to timing issues,
this section is here to give a quick overview of all the different things to consider.

Shouldn't reactive code be by design in a way that timing of certain things becomes irrelevant?

I mean, not that there is no time in observables, or that it does not matter when we subscribe to something,
but when we compose observables we should not care about when any of our state sources exactly emit a value...

In a perfect reactive setup, we don't need to care about those problems.
However, as Angular is an object-orientated framework we often have to deal with different problems
related to life-cycles of components and services, router-events and many more things.

In RxJS timing is given by the following:

For hot observables the time of creation
For cold observables the time of subscription
For emitted values the scheduling process

In Angular timing is given by the following:

For global services the creation as well as the application lifetime
For components the creation, several life-cycle hooks as well as the component lifetime
For local services the creation of the component as well as the components lifetime
For pipes or directives in the template also the components lifetime

All timing relates things in Angular are in an object-oriented style, very similar to hot observables.
Subscription handling can be done declaratively over completion operators.
The scheduling process can be controlled both over imperative or over operators and can influence the execution context of the next error or complete callback.

We see that there are two different concepts combined that have completely different ways of dealing with timing.
Angular already solved parts of this friction points but some of
them are still left and we have to find the right spots to put our glue code and fix the problem.

This chart shows a minimal Angular app with the different building units ant their timing:

In this example it marks:

the global store lifetime
the component store lifetime
the async pipe lifetime

As we can see It makes a big difference where we place observables and where we subscribe to them.
It also shows where we need hot observables and where we need to replay values.

Subscription Handling

Let's discuss where subscriptions should take place and for which reason they are made.

Subscriptions are here to receive values from any source, cold or hot.

In most cases, we want to render incoming values to the DOM.

For this reason, we use a Pipe or a Directive in the template to trigger
change-detection whenever a value arrives.

The other reason could be to run some background tasks in a Component, Directive or Service,
which should not get rendered. E.g. a request to the server every 30 seconds.

As subscriptions in the Pipe or Directive are handled over their life-cycle
hooks automatically, we only have to discuss the scenarios for side-effects.

Let's take a quick look to the diagram from before:

So what could be a good strategy related to the timing of subscriptions and their termination?

One way to solve it would be to subscribe as early as possible and unsubscribe as late as possible.

On a diagram it would look like that:

@Component({
  ...
})
export class SubscriptionHandlingComponent implements OnDestroy {
    onDestroy$ = new Subject();

    sideEffect$ = timer(0, 1000).pipe(tap(n => serverRequest(n)));

    constructor() {
        this.sideEffect$
            .pipe(takeUntil(this.onDestroy$))
            .subscribe();
    }

    ngOnDestroy(): void {
        this.onDestroy$.next(true);
    }
}

(used RxJS parts: timer, tap, takeUntil, Subject)

We already have a declarative subscription handling.
But this code could get moved somewhere else. We could use
the local service that we most probably will need if we implement
the final implementation for component state handling.

Service

export class SubscriptionHandlingService implements OnDestroy {

    onDestroy$ = new Subject();

    subscribe(o): void {
        o.pipe(takeUntil(this.onDestroy$))
            .subscribe();
    }

    ngOnDestroy(): void {
        this.onDestroy$.next(true);
    }

}

Component

@Component({
    ...
    providers: [SubscriptionHandlingService]
})
export class SubscriptionHandlingComponent {
    sideEffect$ = timer(0, 1000)
        .pipe(tap(console.log));

    constructor(private subHandler: SubscriptionHandlingService) {
        this.subHandler
            .subscribe(this.sideEffect$)
    }
}

In this way, we get rid of thinking about subscriptions in the component at all.

Sharing State and State Derivations

In many cases, we want to subscribe to more than one place to some source and render its data.
Even with such a simple operation as retrieving and displaying the data several things needs to be considered.

The interesting parts here are the data structure and derivation logic.

We will skip the data structure and focus on the things related to RxJS.
The derivation of data.

Uni and multi-casting with RxJS

As we have multiple sources we calculate the data for every subscription separately.
This is given by the default behavior of RxJS. It is uni-cased by default.

Basic uni-cast examples:

const timeStampObservable = new Observable((subscriber) => {
    console.log('create date object');
    const dataObject = new Date();
    subscriber.next(dataObject);
});

timeStampObservable
  .subscribe(d => console.log('date: ', d));
timeStampObservable
  .subscribe(d => console.log('date: ', d));

// Result in console:
// 
// create date object
// date:  Fri Dec 13 2042 00:00:00 
// create date object
// date:  Fri Dec 13 2042 00:00:01

(used RxJS parts: Observable)

If we want to multicast the values we could do something like that:

Basic multi-cast examples:

const timestamp object = new Subject();

timeStampSubject
  .subscribe(d => console.log('date: ', d));
timeStampSubject
  .subscribe(d => console.log('date: ', d));

console.log('create date object');
const dataObject = new Date();
timeStampSubject.next(dataObject);

// Result in console:
// 
// create date object
// date:  Fri Dec 13 2042 00:00:00 
// date:  Fri Dec 13 2042 00:00:00

We see how we can share data, now let's take a look at operations:

Operators uni-cast examples:

const timeStampSubject = new Subject();
const timeStampObservable = timeStampSubject.pipe(
    map((date: Date) => {
        console.log('transformation of ', date);
        return date.getTime();
    })
);

timeStampObservable
    .subscribe(d => console.log('date: ', d));
timeStampObservable
    .subscribe(d => console.log('date: ', d));

console.log('create date object');
const dataObject = new Date();
timeStampSubject.next(dataObject);

// create date object
// transformation of Fri Dec 13 2042 00:00:00
// date:  1576231670737
// transformation of Fri Dec 13 2042 00:00:00
// date:  1576231670737

(used RxJS parts: map)

You remember the subscriber function we saw in the first example with the observable?
Operators internally maintain a similar logic. We apply an operator the inner subscriber functions are chained.
This is the reason we see the log for the transformation 2 times. For every subscriber one time.

There are also operators that help to add multi-casting in operator chains.

Operators multi-cast examples:

const timeStampSubject = new Subject();
const timeStampObservable = timeStampSubject.pipe(
    map((date: Date) => {
        console.log('transformation of ', date);
        return date.getTime();
    }),
    share()
);

timeStampObservable
    .subscribe(d => console.log('date: ', d));
timeStampObservable
    .subscribe(d => console.log('date: ', d));

console.log('create date object');
const dataObject = new Date();
timeStampSubject.next(dataObject);

// create date object
// transformation of Fri Dec 13 2042 00:00:00
// date:  1576231670737
// date:  1576231670737

(used RxJS parts: share)

Sharing Work

With this knowledge let's take a look at some examples:

In our view, we could do some processing for incoming data.
An example could be an array of items from an HTTP call.

@Component({
    template: `
    <list-a [list]="httpResult$ | async"></list-a>
    <list-b [list]="httpResult$ | async"></list-b>
    `,
    ...
})
export class AnyComponent {
    httpResult$ = this.http.get(url)
        .pipe(
          map(this.mapServerToClientObject)
        );
}

the work that is done in mapServerToClient is executed once per subscription.
In our example 2 times. Even if we change the HTML and use ng-container to maintain only one subscription in the template
in the class could be multiple other subscriptions we can't solve in the template.

To save work we need to share the subscription.

export class AnyComponent {
    httpResult$ = this.http.get(url)
        .pipe(
          map(this.mapServerToClientObject),
          shareReplay({refCount: true, bufferSize: 1})
        );
}

(used RxJS parts: shareReplay)

Here we use shareReplay to cache the last value, replay it and share all notifications with multiple subscribers.

Sharing Instances

This is a rare case but important to know if you work fully reactive.

To start this section let's discuss the components implementation details first.
We focus on the component's outputs.

@Component({
    ...
})
export class AnyComponent {
    @Output() compOutput = new EventEmitter()
}

Let's take a closer look at the EventEmitter interface:
EventEmitter<T extends any> extends Subject<T>
And Subject looks like this:
Subject<T> extends Observable<T> implements SubscriptionLike
The important part here is that we can pass everything that holds a subscribe method.

Which means the following would work:

@Component({
    ...
})
export class AnyComponent {
    @Output() compOutput = interval(1000);
}

(used RxJS parts: interval)

An observable for example provides a subscribe method.
Therefore we can directly use it as a value stream for our @Output binding.

Knowing that enables us to take some very nice and elegant shortcuts as well as reducing the amount of code we need to write.

With this in mind let's focus on the original problem, sharing work and references.

In this example, we receive a config object from the parent component
turn it into a reactive form and
emit changes from the form group created out of the config object.

Every time we receive a new value from the input binding

we create a config object out of it
and use the FormBuilder service to create the new form. As output value, we have to provide something that holds a subscribe method. So we could use the form groups value changes to provide the forms changes directly as component output events.

@Component({
    selector: 'sharing-a-reference',
    template: `
        <form *ngIf="(formGroup$ | async) as formGroup" [formGroup]="formGroup">
            <div *ngFor="let c of formGroup.controls | keyvalue">
                <label>{{c.key}}</label>
                <input [formControlName]="c.key"/>
            </div>
        </form>
    `
})
export class SharingAReferenceComponent {
    state$ = new ReplaySubject(1);

    formGroup$: Observable<FormGroup> = this.state$
        .pipe(
            startWith({}),
            map(input => this.getFormGroupFromConfig(input))
        );

    @Input()
    set formGroupModel(value) {
        this.state$.next(value);
    }

    @Output() formValueChange = this.formGroup$
        .pipe(switchMap((fg: FormGroup) => fg.valueChanges));

    constructor(private fb: FormBuilder) {

    }

    getFormGroupFromConfig(modelFromInput) {
        const config = Object.entries(modelFromInput)
            .reduce((c, [name, initialValue]) => ({...c, [name]: [initialValue]}), {});
        return this.fb.group(config);
    }

}

(used RxJS parts: startWith)

If we run the code we will see that the values are not updating in the parent component.

We faced a problem related to the fact that nearly all observables are cold, which means that every subscriber will get its instance of the producer.

You might be even more curious now, as our source that produces the formGroup is a ReplaySubject.
Which are multi-casting values and sharing one producer with multiple subscribers...

What we forgot here is that our formGroup$ observable ends with a map operator,
which turns everything after it again into a uni-cast observable.

So what happened?
We subscribed once in the template over the async pipe to render the form.
And another time in the component internals to emit value changes from the form.

As we now know that the map operator turned everything into a uni-cast observable again,
we realize that we created a new FormGroup instance for every subscription.
The subscription in the template, as well as the subscription, happens internally over @Output().

This can be solved by adding a multicast operator like share or shareReplay at the end of formGroup$.

As we also have late subscribers, the async pipe in the template, we use shareReplay with bufferSize 1 serve them the actual formGroup instance.

   formGroup$: Observable<FormGroup> = this.state$
        .pipe(
            startWith({}),
            map(input => this.getFormGroupFromConfig(input)),
            shareReplay({refCount: true, bufferSize: 1})
        );

shareReplay emits the same value to subscribers.

So the subscription in the template and the subscription in the components internals receive the same instance of FormGroup.

Important to notice here is that shareReplay is cold but multicast.
This means it only subscribes to the source if at least one subscriber is present.
This does not solve the problem of cold composition but it is fine to share specific work or in this case a reference.

Later on, in this article, we will remember this problem to provide a way to share work instances.

The Late Subscriber Problem

In this section, I faced the first time a problem that needed some more thinking.

Incoming values arrive before the subscription has happened.

For example state over @Input() decorators arrive before the view gets rendered and a used pipe could receive the value.

@Component({
  selector: 'app-late-subscriber',
  template: `
    {{state$ | async | json}}
  `
})
export class LateSubscriberComponent {
  state$ = new Subject();

  @Input()
  set state(v) {
    this.state$.next(v);
  }

}

We call this situation a late subscriber problem. In this case, the view is a late subscribe to the values from '@Input()' properties.
There are several situations from our previous explorations that have this problem:

Input Decorators
- transporting values from @Input to AfterViewInit hook
- transporting values from @Input to the view
- transporting values from @Input to the constructor
Component And Directive Life Cycle Hooks
- transporting OnChanges to the view
- getting the state of any life cycle hook later in time (important when hooks are composed)
Local State
- transporting the current local state to the view
- getting the current local state for other compositions

A quick solution here would replay the latest notification to use a ReplaySubject with bufferSize 1.
This would cache the latest emitted value and replay it when the async pipe subscribes.

Primitive Solution

@Component({
  selector: 'app-late-subscriber',
  template: `
    {{state$ | async | json}}
  `
})
export class LateSubscriberComponent {
  state$ = new ReplaySubject(1);

  @Input()
  set state(v) {
    this.state$.next(v);
  }

}

(used RxJS parts: ReplaySubject)

This quick solution has 2 major caveats!

First Caveat:
The downside here is that we can only replay the latest value emitted.

Replaying more values would cause problems for later compositions of this stream,
as a new subscriber would get all past values of the @Input Binding.
And that's not what we want.

More important here is the fact that we push workload to the consumer.
We can not assume everybody adopts that.

If we would make every source replay at least the last value we would have to implement this logic in the following places:

View Input bindings (multiple times)
View events (multiple times)
Other Service Changes (multiple times)
Component Internal interval (multiple times)

It would also force the parts to cache values and increase memory.
Furthermore, it would force the third party to implement this too.

IMHO not scalable.

Another downside is the bundle size of ShareReplay.
But it will be used anyway somewhere in our architecture, so it's a general downside.

Second Caveat:
The second and more tricky caveat is composition is still cold.
we rely on the consumer to initialize state composition.

Cold Composition

Let's quickly clarify hot/cold and uni-case/multi-cast.

First, let's remember what we learned about uni- and multi-casting in the earlier chapter.

Uni-cast
The producer is unique per subscription.
Any creation operator is uni-cast. (publish operators are not yet refactored to creation operators, but they would be the only exception)
interval for example would call setInterval for every subscriber separately.

Multi-cast
The producer is shared over all subscriptions.
Subject for example emits it's value to multiple subscribers without executing some producer logic again.

Cold
The internal logic of the observable is executed only on subscription.
The consumer controls the moment when internal logic is executed by the subscribe is called.
The interval creation operator, for example, will only start it's internal tick if we subscribe to it.
Also, nearly every pipe-able operator will execute only if we have an active subscriber.

An interesting example for a cold operator is share.
Even if is multi-casts it's notifications to multiple subscribers,
it will not emit any notification until at least one subscriber is present.

So it's cold at the beginning but multi-cast after the first subscriber. :)

Hot
The internal logic is executed independently from any consumer.
The Subject for example can emit values without any present consumer.

There is also an operator that turn all the above logic into a hot path.
multicast and every publish operator returns a ConnectableObservable.
If we call connect on it we connect to the source an start to execute the logic and all the operators in between publish and it's source observable.

With this in mind, we can discuss the problem of cold composition in the case of our local state.

As we will have to deal with:

View Interaction ( button click )
Global State Changes ( e.g. HTTP update )
Component State Changes ( triggered internal logic )

Putting all this logic in the component class is a bad idea.
Not only because of the separations of concerns but also because we would have to implement it over and over again.

We need to create the logic that deals with the problems around the composition
in a way that it can be reused and is independent!

So far our sources got subscribed to when the view was ready and we rendered the state.
As the input from the view is a hot producer of values and injected services too we have to decouple
the service that handles component state from other sources.

So what is the problem?!!11

We have hot sources and we have to compose them. As we already learned in the section sharing work and instances nearly every operator returns a cold source.
No matter if it was hot before or not.

If we compose state we have to consider that our scan operator returns a cold observable.

So no matter what we do before, after an operation we get a cold observable, and we have to subscribe to it to trigger the composition.
I call this situation a cold composition.

Some of our sources are cold. This can be solved in tow ways:

a) Make all sources replay at least their last value (push workload to all relevant sources)
b) Make the composition hot as early as possible (push workload to the component related part)

We discuss a) already in the previous section.
This solution pushed workload to all involved parties, and the initiation will still be controlled by the consumer.

What would be the scenario with b)?

We could think of the earliest possible moment to make the composition hot.
From the diagram about lifecycle hooks and the different instances in Angular we know that a service, even if locally provided, is instantiated first, before the component.

If we would put it there we could take over the workload from:

View Input bindings (multiple times)
View events (multiple times)
Component Internal interval (multiple times)
Locally provided services
Global services

We could also get rid of the dependency on their subscription.

Let's see a simple example where we rely on the consumer to start the composition.

Service:

export class SomeService {
    commands$ = new Subject();
    composedState$ = this.commands$
        .pipe(
            tap(v => console.log('compute state ', v)),
            scan((acc, i) => {
                return {sum : acc['sum'] + i['sum']};
            }, {sum: 0}),
           // operator here
           shareReplay({refCount: true, bufferSize: 1})
        );
}

(used RxJS parts: scan)

In this service we could try to solve our problem by using:

share()
shareReplay({refCount: true, bufferSize: 1})
shareReplay({refCount: false, bufferSize: 1})

Component:

@Component({
    selector: 'cold-composition',
    template: `
        <h1>Cold Composition</h1>
        <button (click)="updateState()">update state</button><br/>
        <label><input [(ngModel)]="isOpen" type="checkbox"> Show result</label>
        <div *ngIf="isOpen">
            someService.composedState$: {{someService.composedState$ | async | json}}
        </div>
    `,
    providers: [SomeService]
})
export class ColdCompositionComponent {
    isOpen = false;

    constructor(public someService: SomeService) {

    }

    updateState() {
        this.someService.commands$.next({sum: 1})
    }

}

If we run the code and click the button first and then open the result area we see we missed the values emitted before opening the area.

No matter which of the above ways we try, nothing works.
We always lose values if no subscriber is present.

Even if the source is hot (the subject in the service is defined on instantiation) the composition over scan made the stream cold again.
This means the composed values can be received only if there is at least 1 subscriber.
In our case, the subscriber was the components async pipe in the template.

Let's see how we can implement the above in a way we could run hot composition:

Hot Composition Service:

export class SomeService {
    subscription = new Subscription();

   commands$ = new Subject();
    composedState$ = this.commands$
        .pipe(
            tap(v => console.log('compute state ', v)),
            scan((acc, i) => {
                return {sum : acc['sum'] + i['sum']};
            }, {sum: 0}),
           // operator here
           publishReplay(1)
        );

    constructor() {
        // Composition is hot from here on
        this.serviceSubscription = this.composedState$.connect();
    }

}

(used RxJS parts: publishReplay, Subscription)

We kept the component untouched and only applied changes to the service.

We used the publishReplay operator to make the
source replay the last emitted value by using 1 as bufferSize.

In the service constructor, we called connect to make it hot subscribe to the source.

Subscription-Less Interaction with Component StateManagement

So far we only had focused on independent peace and didn't pay much attention to their interaction.
Let's analyze the way we interact with components and services so far.

Well, known implementations of sate management like @ngrx/store in angular,
which is a global state management library, implemented parts of the consumer-facing API imperatively.
Also, all implementations of REDUX in react did it like that.

The provided method is dispatch which accepts a single value, an action, that gets sent to the store.

Let's look at a simple example:

Imperative Interaction Service

export class StateService {
    ...

    private stateSubject = new Subject<{ [key: string]: any }>();

    // setter like dispatch
    setState(v) {
        this.stateSubject.next(v);
    }

    ...

}

Imperative Interaction Component

@Component({
    selector: 'component',
    template: `
        <p>Imperative Interaction</p>
        <pre>{{state$ | async | json}}</pre>
        <button (click)="updateState()">
            Update State
        </button>
    `,

    providers: [StateService]
})
export class AnyComponent implements OnDestroy {
    state$ = this.stateService.state$;

    constructor(private stateService: StateService) {

    }

    updateState() {
        this.stateService
            .dispatch(({key: value}));
    }

}

Why is this imperative? Imperative programming means working with instances and mutating state.
Whenever you write a setter or getter your code is imperative, It's not compose-able.

If we now think about the dispatch method of @ngrx/store, we realize that it is similar to working with setter.

While in this example it sits inside another un-compose-able thing, the instance method and therefore is ok.
Everything else would resolve in more refactoring.

However, we can not use it to work with compose-able sources.

Let's think about connection RouterState or any other source like ngrx/store to the local state:

Imperative Interaction Component

@Component({
    selector: 'component',
    template: `
        <p>Imperative Interaction</p>
        <pre>{{state$ | async | json}}</pre>
    `,

    providers: [StateService]
})
export class AnyComponent implements OnDestroy {
    subscription = new Subscription();
    state$ = this.stateService.state$;

    constructor(private stateService: StateService,
                private store: Store) {

        this.subscription.add(
            this.store.select(getStateSlice)
                .subscribe(value => this.stateService
                    .setState({key: value})
                )
        );

    }

    ngOnDestroy() {
        this.subscription.unsubscribe();
    }

}

As we can see as soon as we deal with something compose-able setters don't work anymore.
We end up in a very ugly code. We break the reactive flow and we have to take care of subscriptions.

But how can we go more declarative or even reactive?
By providing something compose-able :)

Like an observable itself. :)

By adding a single line of code we can go fully declarative as well as fully subscription-less.

Declarative Interaction Service

export class StateService implements OnDestroy {
    ...
    private stateSubject = new Subject<Observable<{ [key: string]: any }>>();
    state$ = this.stateSubject
        .pipe(
            // process observables of state changes
            mergeAll(),
            ...
        );

    // "connector" takes observable (compose-able)
    connectState(o) {
        this.stateSubject.next(o);
    }
    ...
}

(used RxJS parts: mergeAll)

Declarative Interaction Component

@Component({
    selector: 'component',
    template: `
        <p>Imperative Interaction</p>
        <pre>{{state$ | async | json}}</pre>
    `,

    providers: [StateService]
})
export class AnyComponent {
    state$ = this.stateService.state$;

    constructor(private stateService: StateService,
                private store: Store) {
    this.stateService.connectState(
        this.store.select(getStateSlice)
                .pipe(map(value => ({key: value})))
        ); 

    }
}

Let's take a detailed look at the introduced changes:

In StateService we changed stateSubject = new Subject<{ [key: string]: any }>(); to stateSubject = new Subject<Observable<{ [key: string]: any }>>(); It now accepts Observables instead of state objects.
In StateService we added the mergeAll() operator to our state computation logic.
In StateService we replaced the setState method

setState(v: { [key: string]: any }) {
    this.stateSubject.next(v);
}

that took a single value to connectState

connectState(o: Observable<{ [key: string]: any }>) {
    this.stateSubject.next(o);
}

By providing the whole observable we can handle all related mechanisms
of subscription handling, as well as value processing and emission in the service itself and hide all this away from others.

We now have not only way less and maintainable code but also a "subscription-less component".

This simple change will enable us to do create way more than just subscription-Less components.
But this document is dedicated to the very fundamentals.

As a last additional benefit in this section we can talk a little bit about side-effects:

Subscription-Less Handling of Side-Effects

If we recap the above snippet we can see that we not only introduced a subscription-less way for state management interaction,
but also a very elegant way to handle side-effects over our service.

Let's implement some more lines of code:

Service

export class DeclarativeSideEffectsGoodService implements OnDestroy {
    private effectSubscription = new Subscription();
    private effectSubject = new Subject<Observable<{ [key: string]: number }>>();

    constructor() {
        this.effectSubscription = (this.effectSubject
            .pipe(mergeAll(), publish()) as ConnectableObservable<any>).connect();
    }

    ngOnDestroy(): void {
        this.effectSubscription.unsubscribe();
    }

    connectEffect(o: Observable<any>) {
        this.effectSubject.next(o);
    }

}

Declarative Interaction Component

@Component({
    selector: 'declarative-side-effects',
    template: `
        <p>Declarative SideEffects</p>
    `,
    providers: [StateAndEffectService]
})
export class AnyComponent {
    constructor(private stateService: StateAndEffectService) {
        this.stateService.connectEffect(interval(1000)
            .pipe(tap(_ => ({key: value}))));
    }

}

(used RxJS parts: publish ))

Note that the side-effect is now placed in a tap operator and the whole observable is handed over.

Recap Problems

So far we encountered the following problems:

sharing work and references
subscription handling
late subscriber
cold composition
moving primitive tasks as subscription handling and state composition into another layer
Subscription-less components and declarative interaction

If you may already realize all the above problems naturally collapse into a single piece of code.
:)

Also if you remember from the beginning this is what "the gang of four" quote says about Object-Oriented Design Patterns.
:)

We can be happy as we did a great job so far.
We focused on understanding the problems, we used the language specific possibilities the right way,
and naturally, we ended up with a solution that is compact, robust and solves all related problems in an elegant way.

Let's see how the local state service looks like.

Basic Usage

Service Design

State Logic

import {ConnectableObservable, merge, noop, Observable, OperatorFunction, Subject, Subscription, UnaryFunction} from 'rxjs';
import {map, mergeAll, pluck, publishReplay, scan, tap} from 'rxjs/operators';

export function stateful<T>() {
  return (o: Observable<T>): Observable<T> => {
    return o.pipe(
      filter(v => v !== undefined),
      distinctUntilChanged(),
      shareReplay(1)
    );
  };
}

function pipeFromArray<T, R>(fns: Array<UnaryFunction<T, R>>): UnaryFunction<T, R> {
  if (!fns) {
    return noop as UnaryFunction<any, any>;
  }

  if (fns.length === 1) {
    return fns[0];
  }

  return function piped(input: T): R {
    return fns.reduce((prev: any, fn: UnaryFunction<T, R>) => fn(prev), input as any);
  };
}

export class State<T> {
  private subscription = new Subscription();
  private stateObservables = new Subject<Observable<Partial<T>>>();
  private effectSubject = new Subject<any>();
  private stateSlices = new Subject<Partial<T>>();

  private state$ = merge(
    this.stateObservables.pipe(mergeAll()),
    this.stateSlices
  ).pipe(
    scan(this.stateAccumulator, {} as T),
    publishReplay(1)
  );

  constructor() {
    this.init();
  }

  private stateAccumulator(acc: T, command: Partial<T>): T {
    const a = (acc as any) as object;
    const c = (command as any) as object;
    return ({...a, ...c} as T);
  }

  init() {
    this.subscription.add((this.state$ as ConnectableObservable<T>).connect());
    this.subscription.add(
      this.effectSubject.pipe(mergeAll())
        .subscribe()
    );
  }


  setState(s: Partial<T>): void {
    this.stateSlices.next(s);
  }

  connectState<A extends keyof T>(str: A, obs: Observable<T[A]>): void;
  connectState<A extends keyof T>(obs: Observable<Partial<T>>): void;
  connectState<A extends keyof T>(strOrObs: any, obs?: any): void {
    if (typeof strOrObs === 'string') {
      this.stateObservables.next(obs.pipe(map(s => ({[strOrObs as A]: s}))) as Observable<T[A]>);
    } else {
      this.stateObservables.next(strOrObs);
    }
  }

  select(): Observable<T>;
  // ========================
  select<A = T>(
    op: OperatorFunction<T, A>
  ): Observable<A>;
  select<A = T, B = A>(
    op1: OperatorFunction<T, A>,
    op2: OperatorFunction<A, B>
  ): Observable<B>;
  select<A = T, B = A, C = B>(
    op1: OperatorFunction<T, A>,
    op2: OperatorFunction<A, B>,
    op3: OperatorFunction<B, C>
  ): Observable<C>;
  select<A = T, B = A, C = B, D = C>(
    op1: OperatorFunction<T, A>,
    op2: OperatorFunction<A, B>,
    op3: OperatorFunction<B, C>,
    op4: OperatorFunction<C, D>,
  ): Observable<D>;
  select<A = T, B = A, C = B, D = C, E = D>(
    op1: OperatorFunction<T, A>,
    op2: OperatorFunction<A, B>,
    op3: OperatorFunction<B, C>,
    op4: OperatorFunction<C, D>,
    op5: OperatorFunction<D, E>,
  ): Observable<E>;
  // ================================
  select<K1 extends keyof T>(k1: K1): Observable<T[K1]>;
  select<K1 extends keyof T,
    K2 extends keyof T[K1]>(k1: K1, k2: K2): Observable<T[K1][K2]>;
  select<K1 extends keyof T,
    K2 extends keyof T[K1],
    K3 extends keyof T[K1][K2]>(k1: K1, k2: K2, k3: K3): Observable<T[K1][K2][K3]>;
  select<K1 extends keyof T,
    K2 extends keyof T[K1],
    K3 extends keyof T[K1][K2],
    K4 extends keyof T[K1][K2][K3]>(k1: K1, k2: K2, k3: K3, k4: K4): Observable<T[K1][K2][K3][K4]>;
  select<K1 extends keyof T,
    K2 extends keyof T[K1],
    K3 extends keyof T[K1][K2],
    K4 extends keyof T[K1][K2][K3],
    K5 extends keyof T[K1][K2][K3][K4]>(k1: K1, k2: K2, k3: K3, k4: K4, k5: K5): Observable<T[K1][K2][K3][K4][K5]>;
  select<K1 extends keyof T,
    K2 extends keyof T[K1],
    K3 extends keyof T[K1][K2],
    K4 extends keyof T[K1][K2][K3],
    K5 extends keyof T[K1][K2][K3][K4],
    K6 extends keyof T[K1][K2][K3][K4][K5]>(k1: K1, k2: K2, k3: K3, k4: K4, k5: K5, k6: K6): Observable<T[K1][K2][K3][K4][K5][K6]>;
  // ===========================
  select(...opOrMapFn: any[]): Observable<any> {
    if (!opOrMapFn || opOrMapFn.length === 0) {
      return this.state$
        .pipe(
          stateful()
        );
    } else if (!this.isStringArray(opOrMapFn)) {
      const path = (opOrMapFn as any) as string[];
      return this.state$.pipe(
        pluck(...path),
        stateful()
      );
    } else if (this.isOperateFnArray(opOrMapFn)) {
      const oprs = opOrMapFn as OperatorFunction<T, any>[];
      return this.state$.pipe(
        pipeFromArray(oprs),
        stateful()
      );
    }

    throw new Error('Wrong params passed' + JSON.stringify(opOrMapFn));
  }

  holdEffect<S>(observableWithSideEffect: Observable<S>): void;
  holdEffect<S>(obsOrObsWithSideEffect: Observable<S>, sideEffectFn?: (arg: S) => void): void {
    if (sideEffectFn) {
      this.effectSubject.next(obsOrObsWithSideEffect.pipe(tap(sideEffectFn)));
    }
    this.effectSubject.next(obsOrObsWithSideEffect);
  }

  teardown(): void {
    this.subscription.unsubscribe();
  }

  private isOperateFnArray(op: any[]): op is OperatorFunction<T, any>[] {
    return op.every((i: any) => typeof i !== 'string');
  }

  private isStringArray(op: any[]): op is string[] {
    return op.every((i: any) => typeof i !== 'string');
  }

}

Service Implementation

Now let us see some minimal examples on how to use the service:

Extending the service

@Component({
    selector: 'component',
    template: `
        <p>Component</p>
        ...
    `
})
export class AnyComponent extends LocalState {

    constructor() {
        this.super();
    }

}

Injecting the service

@Component({
    selector: 'component',
    template: `
        <p>Component</p>
        ...
    `,
    providers: [LocalState]
})
export class AnyComponent {

    constructor(private stateService: LocalState) {
    }

}

Service Usage

Now let's see some basic usage:

Connecting Input-Bindings

@Component({
    selector: 'component',
    template: `
        <p>Component</p>
        ...
    `
})
export class AnyComponent extends LocalState {

    @Input() 
    set value(value) {
        this.setState({slice: value})
    }

    constructor() {
        this.super();
    }

}

Connecting GlobalState

@Component({
    selector: 'component',
    template: `
        <p>Component</p>
        ...
    `
})
export class AnyComponent extends LocalState {

    @Input() 
    set value(value) {
        this.setState({slice: value})
    }

    constructor(private store: Store) {
        this.connectState(
            this.store.select(getStateSlice)
                    .pipe(transformation)
        );
    }

}

Selecting LocalState

@Component({
    selector: 'component',
    template: `
        <p>Component</p>
        <input (input)="input$.next($event)"/>
        {{state$ | async}}
    `
})
export class AnyComponent extends LocalState {
    input$ = new Subject(); 

    state$ = this.select(
        withLatestFrom(input$),
        map(([state, _]) => state.slice)
    );    

}

(used RxJS parts: withLatestFrom)

Handling LocalSideEffects

@Component({
    selector: 'component',
    template: `
        <p>Component</p>
        ...
    `
})
export class AnyComponent extends LocalState {

    constructor(private store: Store) {
        this.connectEffect(interval(10000)
            .pipe(tap(this.store.dispatch(loadDataAction)))
        );
    }

}

This example shows a material design list that is collapsable.
It refreshed data every n seconds of if we click the button.
Also, it displays the fetched items.

Basic Example - Stateful Component*:

@Component({
    selector: 'basic-list',
    template: `
        <mat-expansion-panel
                *ngIf="model$ | async as m"
                (expandedChange)="listExpandedChanges.next($event)"
                [expanded]="m.listExpanded">
            <mat-expansion-panel-header>
                <mat-panel-title>
                    User Name
                </mat-panel-title>
                <mat-panel-description>
                    <span *ngIf="!m.listExpanded">{{m.list.length}} Repositories Updated every: {{m.refreshInterval}}
                        ms</span>
                    <span *ngIf="m.listExpanded">{{m.list.length}}</span>
                </mat-panel-description>
            </mat-expansion-panel-header>

            <button mat-raised-button color="primary"
                    (click)="refreshClicks.next($event)">
                Refresh List
            </button>

            <div *ngIf="m.list.length; else noList">
                <mat-list>
                    <mat-list-item *ngFor="let item of m.list">
                        {{item.name}}
                    </mat-list-item>
                </mat-list>
            </div>

            <ng-template #noList>
                <mat-card>No list given!</mat-card>
            </ng-template>

        </mat-expansion-panel>
    `,
    changeDetection: ChangeDetectionStrategy.OnPush
})
export class DemoBasicsComponent3 extends LocalState<ComponentState> {
    initComponentState = {
      refreshInterval: 10000,
      listExpanded: false,
      list: []
    };
    refreshClicks = new Subject<Event>();
    listExpandedChanges = new Subject<boolean>();

    model$ = this.select();

    @Input()
    set refreshInterval(refreshInterval: number) {
      this.setState({refreshInterval});
    }

    refreshListSideEffect$ = merge(
        this.refreshClicks,
        this.select(
            map(s => s.refreshInterval),
            tap(console.log),
            switchMap(ms => timer(0, ms))
        )
    )
        .pipe(
            tap(_ => this.store.dispatch(fetchRepositoryList({})))
        );

    constructor(private store: Store<any>) {
        super();
        this.setState(this.initComponentState);
        this.connectState(this.listExpandedChanges
            .pipe(map(b => ({listExpanded: b}))));
        this.holdEffect(this.refreshListSideEffect$);
        this.connectState('list',
            this.store.select(selectRepositoryList).pipe(map(this.parseListItems))
        );
    }

    parseListItems(l: RepositoryListItem[]): DemoBasicsItem[] {
        return l.map(({id, name}) => ({id, name}))
    }

}

This shows some fundamental interaction for template global state and ephemeral state.
The next snippet shows how you would implement architecture patterns based on that service. In this case I picked a simple implementation of the MVVM design pattern.

Basic Example - Design Pattern MVVM*:

export interface DemoBasicsBaseModel {
    refreshInterval: number;
    list: DemoBasicsItem[];
    listExpanded: boolean;
}

export interface DemoBasicsView {
    refreshClicks: Subject<Event>;
    listExpandedChanges: Subject<boolean>
    baseModel$: Observable<DemoBasicsBaseModel>;
}

@Injectable()
export class DemoBasicsViewModelService extends LocalState<DemoBasicsBaseModel> implements DemoBasicsView {
    initState: DemoBasicsBaseModel = {
      refreshInterval: 1000,
      listExpanded: true,
      list: []
    }

    baseModel$ = this.select();

    refreshClicks = new Subject<Event>();
    listExpandedChanges = new Subject<boolean>();

    refreshListSideEffect$ = merge(
        this.refreshClicks,
        this.select(map(s => s.refreshInterval))
            .pipe(switchMap(ms => timer(ms)))
    );

    constructor() {
        super();
        this.setState(this.initState);

        this.connectState(this.listExpandedChanges
            .pipe(map(b => ({listExpanded: b})))
        );
    }

}



@Component({
    selector: 'basic-list',
    template: `
<mat-expansion-panel
        *ngIf="vm.baseModel$ | async as bm"
        (expandedChange)="vm.listExpandedChanges.next($event)"
        [expanded]="bm.listExpanded">
    <mat-expansion-panel-header>
        <mat-panel-title>
            User Name
        </mat-panel-title>
        <mat-panel-description>
                          <span *ngIf="!bm.listExpanded">{{bm.list.length}}
                              Repositories Updated every: {{bm.refreshInterval}}
                              ms</span>
            <span *ngIf="bm.listExpanded">{{bm.list.length}}</span>
        </mat-panel-description>
    </mat-expansion-panel-header>

    <button mat-raised-button color="primary"
            (click)="vm.refreshClicks.next($event)">
        Refresh List
    </button>

    <div *ngIf="bm.list.length; else noList">
        <mat-list>
            <mat-list-item *ngFor="let item of bm.list">
                {{item.name}}
            </mat-list-item>
        </mat-list>
    </div>

    <ng-template #noList>
        <mat-card>No list given!</mat-card>
    </ng-template>

</mat-expansion-panel>
            `,
    changeDetection: ChangeDetectionStrategy.OnPush,
    providers: [DemoBasicsViewModelService]
})
export class DemoBasicsComponent4 {

    @Input()
    set refreshInterval(refreshInterval: number) {
            this.vm.setState({refreshInterval});
    }

    constructor(public vm: DemoBasicsViewModelService,
                private store: Store<any>) {
        this.vm.connectState('list',
            this.store.select(selectRepositoryList).pipe(map(this.parseListItems))
        );
        this.vm.holdEffect(this.vm.refreshListSideEffect$
            .pipe(tap(_ => this.store.dispatch(fetchRepositoryList())))
        );
    }

    parseListItems(l: RepositoryListItem[]): DemoBasicsItem[] {
        return l.map(({id, name}) => ({id, name}))
    }

}

The lase example showed how MVVM in implemented based on the reactive state class. What is interesting here is that the template only accesses the ViewModel, nothing else.

But this is part of another document I would suggest. ;p

Summary

How to differ global from the ephemeral state:

No horizontal sharing of state
The lifetime of the state is dynamic, bound to e.g. a component
It processes local relevant events

If we take a look at our operator reference list at the end of this document we can see it was a lot about:

unicast vs. multicast
hot vs. cold

The main outcome here was we should ensure that the moment of computation of states is not controlled by the subscriber.
It should be hot.

We learned how to can have a fully reactive flow with

higher-order operators like mergeAll The combination with our logic bound to a certain life-time enabled us to create
subscription-less components

An example implementation of our learning can be found in the resources.

Based on that we used in a minimal example and also made the first test with some design patterns like MVVM.

Resources

Recording (🎥 Live Demo at 24:47):

🎥 Angular Vienna, Angular, and RxJS - Tackling Ephemeral State Reactively

Slides:

🖼️ Tackling Component State Reactively

Repository For Examples (💾 Final Example):

💾 research-on-reactive-ephemeral-state-in-component-oriented-frontend-frameworks

Sourcecode: 📦 ngx-rx/rxjs-state
NPM Package: 📦 ngx-rx-state

Used RxJS parts:

Common Approaches to Handling Subscriptions

Andrew Evans — Tue, 17 Sep 2019 18:06:02 +0000

When developers first start using RxJS one of the biggest challenges is handling subscriptions.

Subscriptions in RxJS handle the execution flow of a stream. When we create Observables, we "subscribe" to them to start using them. Conversely, when we "unsubscribe" we do the reverse to stop the same execution flow.

Handling this can be a bit tricky. This post is going to cover some common patterns for handling subscriptions in your code.

This post is also going to be framework agnostic in an effort to make these patterns accessible by all.

The examples that are used in this post can be reached in my Stackblitz Project.

I'm going to show the code here, and have an embedded link to my Stackblitz project at the end. I encourage you to run the code examples that I walkthrough to get a better understanding.

Memory Leaks and Your First Unsubscribe

When we do not successfully unsubscribe from an Observable, we create a situation called a "memory leak." This is any time a stream is started (using system resources) and not stopped. If you have enough streams started without an "unsubscribe," you can use up a lot of your system resources and significantly slow down your application...this is not a good thing.

A good example of this would be a simple Observable from the creation operator interval. Consider the following code:

import { interval } from 'rxjs';

const observable = interval(1000);
const subscription = observable.subscribe(() => console.log('Hello!'));

So in this example we are just using the interval operator to create a stream that writes "Hello!" to the console every 1 second. When we call subscribe we are saying that whenever the stream emits a response (in this case every 1 second), we print "Hello!".

This is very simplistic, but the challenge here is that if we do not call unsubscribe, this stream will continue running until you end your session or destroy the associated component etc. This is really easy to miss and important for performance.

To fix this situation, a simple "unsubscribe" is needed. So consider the same code but with the addition of an "unsubscribe" call like so:

import { interval } from 'rxjs';

const observable = interval(1000);
const subscription = observable.subscribe(() => console.log('Hello!'));
setTimeout(() => {
  subscription.unsubscribe();
  console.log('unsubscribed');
}, 1000);

NOTE: We are using a setTimeout here just to allow for a second to pass and the "hello" to be shown in the console.

Now with the "unsubscribe" called, the execution ends correctly and you're successfully managing the stream.

Using the take Operator

So in the previous example the subscription was managed manually with direct calls to subscribe and unsubscribe. This pattern is fine but is also easy to forget to do.

A less error prone approach would be to make use of the take operator. When passed into an Observable, the take operator enables you to end the execution after a set number of emissions from the stream.

Consider the following code:

import { interval } from 'rxjs';
import { take } from 'rxjs/operators';

const intervalCount = interval(1000);
const takeObservable = intervalCount.pipe(take(2));
takeObservable.subscribe(x => console.log(x));

When you run this, you should see the following:

0
1

Now what if you changed that same code to the following:

import { interval } from 'rxjs';
import { take } from 'rxjs/operators';

const intervalCount = interval(1000);
const takeObservable = intervalCount.pipe(take(10));
takeObservable.subscribe(x => console.log(x));

When you run this, you should see the same as before but the count goes from 0 to 9.

So what's happening? The take operator just controls the execution flow so that the number you pass in determines how many times it emits a value before completing. You don't have to worry about a memory leak here because the completion formally stops the execution flow here.

In addition to the take operator there are multiple other examples of ways to do this behavior.

Some include the following:

The important thing about this behavior is just that you're letting RxJS handle the stream for you. This lets you write clean code that is easily maintainable.

Combining Subscriptions

Another common pattern that you run across is when you multiple observables, and want to manage their subscriptions together.

Consider the following code:

import { Subscription, of } from 'rxjs';

// create a subscription object
const subs = new Subscription();

// create observables
const value$ = of(1, 2, 3, 4);
const anotherValue$ = of(true);

// subscribe to observables and add to subscription
subs.add(value$.subscribe(x => console.log(x)));
subs.add(anotherValue$.subscribe(x => console.log(x)));

// calling subs.unsubscribe() will unsubscribe from all sub
subs.unsubscribe();

special thanks to my friend Tim Deschryver for this specific example

In this example you see that we define an instance of a Subscription that we add two observables to. The Subscription class enables you to wrap your subscriptions in one resource. When you're ready to dispose of your application, you can just call a singular unsubscribe and execution in all of the observables wrapped will be properly stopped.

This pattern is particularly useful when you have multiple observables in a component that you want to manage together. It makes the implementation cleaner and easier to maintain.

Combining Subscriptions with tap and merge

In addition to the above example, another common pattern is to make use of the tap operator and static merge function to combine multiple observables.

Consider the following code:

// create observables
const value$ = of(1, 2, 3, 4).pipe(tap(x => console.log(x)));
const anotherValue$ = of(true).pipe(tap(x => console.log(x)));

const subs = merge(value$, anotherValue$).subscribe();

subs.unsubscribe();

The static merge function enables you to combine many observables into a single value. Then when you are ready to stop execution, a single unsubscribe stops execution on the group. This pattern is very clean, and enables RxJS to handle the orchestration of your streams without needing to declare additional operators etc.

Closing Thoughts

So in this post you saw a few patterns for handling subscriptions with RxJS. The really great part is that RxJS is very flexible and can accommodate (almost) any use-case. I hope the examples here have provided you with some basic patterns for your applications. Feel free to leave comments and follow me on Twitter at @AndrewEvans0102!

Here's a stackblitz for the examples above:

DEV Community: RxJS

Much Fun with RxJS-Autorun

Mapping a single stream: 🌱 🌿 🌳

Combinations: 📦 + 🎀 = 🎁

Switching: 🌤 🌧

Tricks: 🐇 🎩

Filtering

Take while

Bonus: 🙂

🙂 → 😲 → 🤯

Outro: 🛸💨

Psst.. need something more to read?

😉

RxJS Proxy: 3 new features

📖 Intro

🎉 Outro

RxJS Autorun Intro

It's a sweetened countdown

Infinite monkey theorem needs infinite bananas

Who's up to some pizza?

Transformation

Outro 😳

Psst.. need something more to read?

😉

Turn a Stream of Objects into an Object of Streams

🎹 Try it

📖 Repository

Outro

Psst.. need something more to read?

😉

Fetching Data in React with RxJS and <$> fragment

🔋 Power Up

🔧 Refactoring and <$> fragment

⏳ Loading indicator

⚠️ Error handling

🖼 Better UI

🍰 Bonus: anti-flickering

🎉 Outro

Observables, Reactive Programming, and Regret

Observables vs Reactive Programming vs RxJS

Regrets

1. The huge API

2. RxJS out-shined Observable

3. We never really outlined where RxJS would best serve people

Examples:

4. We never taught people how to write readable code with RxJS

Consequences

The road forward

RxJS debounce vs throttle vs audit vs sample — Difference You Should Know

Outro

Psst.. need something more to read?

😉

“RxJS Cheatsheet” VS Code extension

New in RxJS v7: concatWith operator

Getting started with contributing to RxJS docs

Running the docs locally

Where to make your change?

Making your change

Previewing your change

Preparing to commit

Making a Pull Request (PR)

Congrats! 🎉

Research on Reactive-Ephemeral-State in component-oriented frameworks

Studies are done with Angular as an example for a component-oriented framework and RxJS is used as an example for a reactive programming library

Table of Content

TL;DR

BioPhoton / research-reactive-ephemeral-state-in-component-oriented-frontend-frameworks

Research on reactive-ephemeral-state in component oriented frontend frameworks - Demonstrated with Angular and RxJS

Methodology

Layers of State

What is the ephemeral state?

Global vs Local Accessibility of Data Structures

Static vs Dynamic Lifetime of Data Structures

Global vs Local Processed Sources

Recap Ephemeral State

Problems to Solve on a Low-Level

Timing

Subscription Handling

Sharing State and State Derivations

Uni and multi-casting with RxJS