Multicasting in RxJS: share, shareReplay, connectable and Sharing an HTTP Request

Version note: The examples in this post were run with RxJS 7.8.2, Angular 22.2, TypeScript 6.0 and Node.js 24 (the Angular project uses TypeScript 6.0, the range Angular 22.2 supports); timed examples run in virtual time on RxJS's TestScheduler, the Angular example in headless Chromium driven by Playwright 1.65.

Predict First

The catalog page of Bookstore, the online bookstore used as the shared example across this site's series, needs the list of categories in three places: the menu, the filter and the footer. To avoid three requests we add share() to the request. The menu and the filter subscribe immediately, the footer one second later:

import { share, timer } from "rxjs";
import { fromFetch } from "rxjs/fetch";

// GET /api/categories once, shared by every subscriber... or is it?
const categories$ = fromFetch("http://localhost:5180/api/categories", {
  selector: (response) => response.json() as Promise<string[]>,
}).pipe(share());

categories$.subscribe((c) => console.log("menu:", c.join(", ")));
categories$.subscribe((c) => console.log("filter:", c.join(", ")));
// the footer is rendered one second later
timer(1000).subscribe(() => categories$.subscribe((c) => console.log("footer:", c.join(", "))));

With the mock API running (node api/server.ts), how many GET /api/categories lines does the server log show: one, two or three?

In the first post of this series we saw that every subscribe on a cold Observable runs its producer again, and that two subscribes to the same HTTP Observable send two requests. In the previous post a Subject shared values that our own code produced. This post connects the two: we take an existing cold Observable and share it with share, shareReplay, connectable and connect, measure it in a real Angular 22 project, and look at the two classic multicasting bugs: a source that never stops and a failed response that is cached forever.

Unicast and Multicast: Two async Pipes, Two Requests

First the answer:

menu: science-fiction, classics, fantasy
filter: science-fiction, classics, fantasy
footer: science-fiction, classics, fantasy

[api] listening on http://localhost:5180
[api] #1 GET /api/categories
[api] #1 200
[api] #2 GET /api/categories
[api] #2 200

Two requests. The menu and the filter shared one, but the footer started a second one. share() forgets everything when the source completes: by the time the footer subscribed, the first request had finished, and a new subscriber after completion starts the source again. share shares a running execution, not a result. We will fix that in a moment; first let us see the same problem in Angular, where it usually appears without anybody writing subscribe.

An Observable is unicast by default: each subscriber gets its own execution of the producer. A Subject is multicast: one execution, many subscribers. The multicasting operators put a Subject between the source and the subscribers. Angular's HttpClient returns a cold, unicast Observable; this is easy to say from the documentation, so we measured it. bookstore-web's catalog page (in the bookstore-web/ folder of the snapshot) shows the categories twice, as links and in a select box, and its first version read them directly from HttpClient:

    <nav class="categories">
      @for (category of categories$ | async; track category) {
        <a href="#">{{ category }}</a>
      }
    </nav>
    <label>
      Category
      <select>
        @for (category of categories$ | async; track category) {
          <option>{{ category }}</option>
        }
      </select>
    </label>

protected readonly categories$ = this.#http.get<string[]>(`${apiUrl}/categories`);

We opened the catalog in headless Chromium, went to the cart page and came back; these are the requests the page sent:

open catalog
  request /api/books
  request /api/categories
  request /api/categories
  nav: science-fiction, classics, fantasy | select: 3 options
go to cart
back to catalog
  request /api/books
  request /api/categories
  request /api/categories

Two async pipes, two subscriptions, two requests for the same list, and two more when the user comes back to the page. Nothing in the template looks like a subscription, which is why this bug is so common. The same happens if a component reads the same HttpClient Observable with async in the template and subscribe in the code, or with two toSignal calls.

share and shareReplay Are Not the Same Thing

Both operators are configurations of the same mechanism. share takes four options: connector (which Subject to put in the middle, a plain Subject by default), and three reset rules that say when to forget the shared execution: resetOnError, resetOnComplete and resetOnRefCountZero (when the last subscriber leaves). In RxJS 7.8.2 shareReplay(config) is literally share with a ReplaySubject connector, resetOnError: true, resetOnComplete: false and resetOnRefCountZero set to your refCount option:

The same request with the same three subscribers (0, 0 and 1000 ms) shows what this means; the last scenario uses share's options to build a cache that expires after 60 seconds:

import { ReplaySubject, share, shareReplay, timer, type Observable } from "rxjs";
import { log, runInVirtualTime } from "../../tools/virtual-time.ts";
import { fakeRequest } from "../04/fake-api.ts";

const getCategories = () => fakeRequest("GET /categories", 200, () => ["science-fiction", "classics", "fantasy"]);

// three parts of the page subscribe at 0 ms, 0 ms and 1000 ms (and a fourth at 90 s)
function subscribeLikeAPage(categories$: Observable<string[]>, times = [0, 0, 1000]) {
  times.forEach((ms, i) => timer(ms).subscribe(() => categories$.subscribe((c) => log(`subscriber ${i + 1}: ${c.length} categories`))));
}

runInVirtualTime("no sharing", () => subscribeLikeAPage(getCategories()));
runInVirtualTime("share()", () => subscribeLikeAPage(getCategories().pipe(share())));
runInVirtualTime("shareReplay(1)", () => subscribeLikeAPage(getCategories().pipe(shareReplay(1))));
runInVirtualTime("share as a 60-second cache", () =>
  subscribeLikeAPage(
    getCategories().pipe(
      share({
        connector: () => new ReplaySubject(1),
        resetOnComplete: () => timer(60_000), // forget the answer 60 s after it arrived
        resetOnError: true,
        resetOnRefCountZero: false,
      }),
    ),
    [0, 0, 1000, 90_000],
  ),
  120_000, // let the virtual clock run past 90 s
);

--- no sharing
    0 ms  -> GET /categories
    0 ms  -> GET /categories
  200 ms  <- GET /categories
  200 ms  subscriber 1: 3 categories
  200 ms  <- GET /categories
  200 ms  subscriber 2: 3 categories
 1000 ms  -> GET /categories
 1200 ms  <- GET /categories
 1200 ms  subscriber 3: 3 categories
--- share()
    0 ms  -> GET /categories
  200 ms  <- GET /categories
  200 ms  subscriber 1: 3 categories
  200 ms  subscriber 2: 3 categories
 1000 ms  -> GET /categories
 1200 ms  <- GET /categories
 1200 ms  subscriber 3: 3 categories
--- shareReplay(1)
    0 ms  -> GET /categories
  200 ms  <- GET /categories
  200 ms  subscriber 1: 3 categories
  200 ms  subscriber 2: 3 categories
 1000 ms  subscriber 3: 3 categories
--- share as a 60-second cache
    0 ms  -> GET /categories
  200 ms  <- GET /categories
  200 ms  subscriber 1: 3 categories
  200 ms  subscriber 2: 3 categories
 1000 ms  subscriber 3: 3 categories
90000 ms  -> GET /categories
90200 ms  <- GET /categories
90200 ms  subscriber 4: 3 categories

Without sharing, three subscribers made three requests. share() merged the two simultaneous subscribers into one request, but the late one started a new request, exactly as in the predict example. shareReplay(1) did not reset on completion, so the subscriber at 1000 ms got the stored answer from the ReplaySubject without a request. The last scenario is a cache with a lifetime: resetOnComplete: () => timer(60_000) waits 60 seconds after the answer and then resets, so the subscriber at 1000 ms was served from memory and the one at 90 seconds triggered a fresh request. resetOnError: true makes sure a failed request is not kept, and resetOnRefCountZero: false lets a request in progress finish even if every subscriber leaves. The reset options accept either a boolean or a function that returns an Observable; the reset happens when that Observable emits.

Sharing the Category List Across the App

Categories rarely change, so one request per application session is enough. The list moves into a service, and the shared Observable is created once, in the service's field:

import { HttpClient } from '@angular/common/http';
import { inject, Service } from '@angular/core';
import { shareReplay } from 'rxjs';
import { apiUrl } from '../api';

@Service()
export class CategoryService {
  // one request for the whole app; every later subscriber gets the cached list
  readonly categories$ = inject(HttpClient)
    .get<string[]>(`${apiUrl}/categories`)
    .pipe(shareReplay({ bufferSize: 1, refCount: false }));
}

The catalog page now reads categories$ = inject(CategoryService).categories$; the template with its two async pipes did not change. The same navigation as before:

open catalog
  request /api/books
  request /api/categories
  nav: science-fiction, classics, fantasy | select: 3 options
go to cart
back to catalog
  request /api/books

One categories request for the whole visit, also after coming back from the cart. The book list is still requested again, because the page creates it in its own field with toSignal; whether a list should be cached is a decision per endpoint, not a default. Where the shareReplay is placed matters as much as the operator: shareReplay inside a method (getCategories() { return http.get(…).pipe(shareReplay(1)); }) creates a new shared Observable on every call and shares nothing. The shared instance must be stored in a field or a variable that outlives its subscribers.

refCount: A Source That Never Stops

For an HTTP request refCount hardly matters, because the source completes by itself and shareReplay never resets on completion. It matters for sources that do not complete. The book detail page shows Dune's live stock, shared through shareReplay; at 2.5 seconds the user leaves the page:

import { interval, map, shareReplay, tap, timer } from "rxjs";
import { log, runInVirtualTime } from "../../tools/virtual-time.ts";

// a live stock feed for Dune that never completes
const liveStock = () =>
  interval(1000).pipe(
    map((i) => 12 - i),
    tap((stock) => log(`  feed produced ${stock}`)),
  );

for (const refCount of [false, true]) {
  runInVirtualTime(`shareReplay({ bufferSize: 1, refCount: ${refCount} })`, () => {
    const stock$ = liveStock().pipe(shareReplay({ bufferSize: 1, refCount }));
    const detailPage = stock$.subscribe((s) => log(`detail page: ${s}`));
    timer(2500).subscribe(() => {
      detailPage.unsubscribe();
      log("user leaves the detail page");
    });
  }, 5000);
}

--- shareReplay({ bufferSize: 1, refCount: false })
 1000 ms    feed produced 12
 1000 ms  detail page: 12
 2000 ms    feed produced 11
 2000 ms  detail page: 11
 2500 ms  user leaves the detail page
 3000 ms    feed produced 10
 4000 ms    feed produced 9
 5000 ms    feed produced 8
--- shareReplay({ bufferSize: 1, refCount: true })
 1000 ms    feed produced 12
 1000 ms  detail page: 12
 2000 ms    feed produced 11
 2000 ms  detail page: 11
 2500 ms  user leaves the detail page

With refCount: false, which is what shareReplay(1) means, the feed kept producing after the last subscriber left: 10, 9, 8, and it would go on as long as the application runs (the example stops the virtual clock at 5 seconds). The ReplaySubject keeps a subscription to the source, the source keeps its timer or socket, and whatever the pipe captured stays in memory. With refCount: true the feed stopped together with the last subscriber. The rule: for a source that never completes (interval, WebSocket, a store selector, valueChanges), write shareReplay({ bufferSize: 1, refCount: true }). refCount: false is only right when the source completes, or when you deliberately want it to run for the whole session.

Caching a Failed Response

A cached request has a second trap: what gets cached when the request fails. Here the first call to the categories endpoint fails with 503, every later call succeeds:

import { catchError, defer, Observable, of, shareReplay, timer } from "rxjs";
import { log, runInVirtualTime } from "../../tools/virtual-time.ts";
import { fakeRequest } from "../04/fake-api.ts";

// the first call fails with 503, the server is fine afterwards
function categoriesApi(): Observable<string[]> {
  let calls = 0;
  return defer(() =>
    ++calls === 1
      ? new Observable<string[]>((subscriber) => {
          log("-> GET /categories");
          const pending = timer(200).subscribe(() => {
            log("<- GET /categories 503");
            subscriber.error(new Error("503 Service Unavailable"));
          });
          return () => pending.unsubscribe();
        })
      : fakeRequest("GET /categories", 200, () => ["science-fiction", "classics", "fantasy"]),
  );
}

function render(categories$: Observable<string[]>, label: string) {
  categories$.subscribe({
    next: (c) => log(`${label}: [${c.join(", ")}]`),
    error: (err: Error) => log(`${label}: error ${err.message}`),
  });
}

runInVirtualTime("catchError before shareReplay: the fallback is cached", () => {
  const categories$ = categoriesApi().pipe(
    catchError(() => of([])),
    shareReplay(1),
  );
  render(categories$, "menu");
  timer(5000).subscribe(() => render(categories$, "menu after 5 s"));
});

runInVirtualTime("shareReplay alone: the error is not cached", () => {
  const categories$ = categoriesApi().pipe(shareReplay(1));
  render(categories$.pipe(catchError(() => of([]))), "menu");
  timer(5000).subscribe(() => render(categories$.pipe(catchError(() => of([]))), "menu after 5 s"));
});

--- catchError before shareReplay: the fallback is cached
    0 ms  -> GET /categories
  200 ms  <- GET /categories 503
  200 ms  menu: []
 5000 ms  menu after 5 s: []
--- shareReplay alone: the error is not cached
    0 ms  -> GET /categories
  200 ms  <- GET /categories 503
  200 ms  menu: []
 5000 ms  -> GET /categories
 5200 ms  <- GET /categories
 5200 ms  menu after 5 s: [science-fiction, classics, fantasy]

In the first version catchError sits before shareReplay and turns the error into an empty list. For shareReplay that empty list is a successful answer followed by completion, so it is cached: five seconds later the server was healthy again, but the menu showed an empty list without even sending a request, and it would stay empty until the page is reloaded. In the second version the error reaches shareReplay, which resets on error (resetOnError: true), and each subscriber handles the error itself. The subscriber at 5 seconds sent a new request and got the categories. Put the fallback after the sharing, at the consumer; the shared part should either succeed or fail, not store a substitute. The error handling post of this series goes deeper into catchError and retry.

connectable and connect: Controlled Multicasting

share starts the source when the first subscriber arrives. Sometimes you want to decide the moment yourself, for example to make sure that every subscriber is in place before the first value is produced. That is what connectable does; the connect operator solves a related problem inside a single pipe:

import { connect, connectable, defer, filter, interval, map, merge, take, timer } from "rxjs";
import { log, runInVirtualTime } from "../../tools/virtual-time.ts";

// Dune's stock feed: logs every connection, emits 12, 11, 10 one second apart
const stockFeed = defer(() => {
  log("feed connected");
  return interval(1000).pipe(
    map((i) => 12 - i),
    take(3),
  );
});

runInVirtualTime("connectable: subscribe first, connect when ready", () => {
  const stock$ = connectable(stockFeed);
  stock$.subscribe((s) => log(`header badge: ${s}`));
  stock$.subscribe((s) => log(`detail page: ${s}`));
  timer(500).subscribe(() => {
    log("connect()");
    stock$.connect(); // one subscription to the feed, for both subscribers
  });
});

runInVirtualTime("connect operator: share inside one pipe", () => {
  stockFeed
    .pipe(
      connect((shared$) =>
        merge(
          shared$.pipe(map((s) => `stock: ${s}`)),
          shared$.pipe(
            filter((s) => s <= 10),
            map((s) => `low stock warning: ${s}`),
          ),
        ),
      ),
    )
    .subscribe((line) => log(line));
});

--- connectable: subscribe first, connect when ready
  500 ms  connect()
  500 ms  feed connected
 1500 ms  header badge: 12
 1500 ms  detail page: 12
 2500 ms  header badge: 11
 2500 ms  detail page: 11
 3500 ms  header badge: 10
 3500 ms  detail page: 10
--- connect operator: share inside one pipe
    0 ms  feed connected
 1000 ms  stock: 12
 2000 ms  stock: 11
 3000 ms  stock: 10
 3000 ms  low stock warning: 10

connectable(source) returns an Observable that does nothing on subscribe; the header badge and the detail page attached to it, and only connect() at 500 ms subscribed to the feed, once, for both. Its options are connector (a Subject by default) and resetOnDisconnect. The connect operator gives a function a shared version of the source: both branches inside merge read the same feed, and the log shows a single "feed connected". Without connect, two pipes on stockFeed would mean two connections.

If you maintain older code, you will find publish, publishReplay, multicast and refCount() instead. In RxJS 7 they are deprecated; the type definitions in 7.8.2 say they will be removed in version 8 and point to connectable, connect and share. publishReplay(1), refCount(), for example, becomes shareReplay({ bufferSize: 1, refCount: true }). The RxJS 9 post of this series covers the migration.

In Production

Count requests, not assumptions. The Network tab of the browser with a filter on /api/, or the access log of the API, shows duplicate requests immediately; a page that sends the same GET twice within a few milliseconds almost always has two subscriptions on a cold Observable.

An application-wide shareReplay is a cache, and every cache needs an answer to "how does it get fresh data?". Categories may live for a session; prices and stock may not. Use a share configuration with a timed reset as above, or invalidate explicitly by recreating the shared Observable after a write (for example after an admin adds a category). In Angular, a service that holds a cache should also be cleared when the user logs out, otherwise the next user on the same browser sees the previous user's data.

Finally, a cached value is shared by reference. If one component mutates the array it received (push, sort), every other subscriber sees the change. Treat shared values as immutable: copy before sorting (toSorted), or freeze them in development.

Common Pitfalls

• Using the same HttpClient Observable with two async pipes, or with async and subscribe. Each one sends its own request.
• Expecting share() to cache. It resets when the source completes; a subscriber after that starts the source again. Use shareReplay or a ReplaySubject connector.
• Calling shareReplay inside a method. Every call creates a new shared Observable; store it in a field.
• Using shareReplay(1) on a source that never completes. With refCount: false the source runs forever; write refCount: true.
• Putting catchError before shareReplay. The fallback value is cached and the error never goes away; handle errors after sharing.
• Mutating a shared, cached value. Every subscriber holds the same reference.
• Writing new code with publish, multicast or refCount(). They are deprecated in RxJS 7.

A Small Experiment

★☆☆ In 01-predict.ts, replace share() with shareReplay(1). How many requests does the server log show now? Then move the footer's subscription to 0 ms and back to 1000 ms with share({ resetOnComplete: false }); what changed compared to shareReplay(1), and what did not?

★★☆ Give CategoryService a refresh() method that makes the next subscriber fetch the list again (for example after an admin adds a category), without breaking components that are already subscribed. Hint: keep the shared Observable in a field and replace it, or use a Subject as a reset notifier for share.

★★★ Change 03-sharereplay-refcount.ts so that the user leaves the detail page at 2.5 seconds and comes back at 4 seconds. With refCount: true, what does the page show right after it comes back, and does the feed continue or start over? Then try share({ connector: () => new ReplaySubject(1), resetOnRefCountZero: () => timer(3000) }) and explain the difference.

Two scenarios in this post ended with an error: the failed categories request and the 503 that was cached by mistake. In the next post we handle errors properly: what catchError really replaces, how to keep a search box alive after a failed request, retry with exponential backoff only where it is safe, Angular's HttpErrorResponse and ASP.NET Core's ProblemDetails, and finalize for loading indicators.