Your methods should not rely on specific contexts or mapping results to a fixed shape.

RxJS focuses on functional programming, where there is a concept called Adapting Arguments to Parameters^ref

This pattern allows for separating method signatures from contexts.

To achieve this, you can create context-specific versions of operators like map, contentMap, mergMap. The final solution may look like:

this.startUploadEvent$.pipe(
      map(withKey('event')),
      concatMap_(({event}) => this.getAuthenticationHeaders(event), 'headers'),
      map_(({ headers }) => this.generateUploadId(headers), 'id'),
      tap(({ event, id }) => this.emitUploadStartEvent(id, event)),
      concatMap_(({ id }) => this.createPdfDocument(id), 'pdfId'),
      concatMap_(({ pdfId }) => this.uploadBuilderForPdf(pdfId), 'cloudId'),
      mergeMap_(({ cloudId }) => this.closePdf(cloudId)),
      tap(({id, event, cloudId}) => this.emitUploadDoneEvent(id, event, cloudId)),
    ).subscribe(console.log);

Note the underscore _ after those operators.

Stackblitz Example

The goal of these custom operators is to pass parameters through a projection function and add the result to the original parameters object.

(Function code examples provided in the original text)

In some cases, over-engineering may not be necessary. Stick to simpler implementations if complexity is not required in your existing codebase.

Doesn't it look like over-engineering?

In most cases, follow the YAGNI (You aren't gonna need it) principle to avoid unnecessary complexity. Consider simple ways to share parameters between operators when needed instead.

(Code example given in the original text)

There are alternative methods that require changing the way you use RxJS operators:

• https://medium.com/@snorredanielsen/rxjs-accessing-a-previous-value-further-down-the-pipe-chain-b881026701c1

Here are some ways you can optimize your actions using observables:

• Assign the outcome of each action to an observable

• Link subsequent function calls based on previous results

• You can reuse those outcomes in later actions by using withLatestFrom

• To prevent earlier functions from re-executing due to subsequent withLatestFrom subscriptions, utilize shareReplay

  function startUpload(event$: Observable<string>) {
    const headers$ = event$.pipe(
      concatMap(event => getAuthenticationHeaders(event)),
      shareReplay()
      );
  
    const id$ = headers$.pipe(
      map(() => generateUploadId()),
      shareReplay()
      );
  
    const emitUploadEvent$ = id$.pipe(
      withLatestFrom(event$),   // use earlier result
      map(([id, event]) => emitUploadStartEvent(id, event)),
      shareReplay()
      );
  
     // etc
  }
  

The functions strictly take only the necessary parameters without any additional pass-through data.

Check out a demo example here: https://stackblitz.com/edit/so-rxjs-chaining-1?file=index.ts

This process can be streamlined by creating a custom RxJS operator (note that further refinements, including typing, can be implemented):

function call<T, R, TArgs extends any[], OArgs extends Observable<any>[]>(
  operator: (func: ((a: TArgs) => R)) => OperatorFunction<TArgs,R>,
  action: (...args: any[]) => R,
  ignoreInput: boolean,
  ...observableArgs: OArgs
): (args: Observable<T>) => Observable<R> {
  return (input: Observable<T>) => input.pipe(
    withLatestFrom(...observableArgs),
    operator((args: any[]) => action(...args.slice(ignoreInput ? 1: 0))),
    shareReplay(1)
  );
}

You can apply this as follows:

function startUpload(event$: Observable<string>) {
  const headers$ = event$.pipe(
    call(concatMap, getAuthenticationHeaders, true)
  );

  const id$ = headers$.pipe(
    call(map, generateUploadId, false)
  );

  const startEmitted$ = id$.pipe(
    call(map, emitUploadStartEvent, true, event$)
  );

  const pdfId$ = startEmitted$.pipe(
    call(map, createPdfDocument, false, event$, headers$, id$)
  );

  const uploaded$ = pdfId$.pipe(
    call(map, uploadBuilderForPdf, false, event$, pdfId$, headers$, id$)
  );

  const cloudId$ = uploaded$.pipe(
    call(map, closePdf, false, headers$, pdfId$)
  );

  const uploadDone$ = cloudId$.pipe(
    call(map, emitUploadDoneEvent, true, id$, event$)
  );

  // To maintain the observable chain, return cloudId$ instead of uploadDone$
  return uploadDone$.pipe(concatMap(() => cloudId$));    
}

Explore the working demo here: https://stackblitz.com/edit/so-rxjs-chaining-4?file=index.ts

Is it possible to utilize an object for storing the data set? Perhaps something along these lines:

Data Structure:

export interface Packet {
  event: string;
  headers?: string;
  id?: number;
  pdfId?: number;
  cloudId?: number;
}

Subsequently, in the code, consider implementing a setup like this:

Service Logic:

  this.startUploadEvent$.pipe(
    concatMap(packet => this.doTaskOne(packet)),
    map(packet => this.doTaskTwo(packet)),
    tap(packet => this.doTaskThree(packet)),
    // ...
  );

This method allows each function to access the necessary components of the object and pass on the remainder. However, it will be essential to adjust each function to accept and process the object accordingly.

It seems like your main concern is readability and avoiding carrying the payload from method to method.

Have you considered converting an Observable to a Promise? The key point here is that the observables need to complete for the promise to be fulfilled and resolved (similar to completion but specific to promises).

Following your suggestion mentioned above (such as using async await), I have come up with this proposal.

private async startUpload(event: StartUploadEvent) {
    const headers = await this.getAuthenticationHeaders(event).toPromise();
    const id = await this.generateUploadId().toPromise();
    
    this.emitUploadStartEvent(id, event);
    
    const pdfId = await this.createPdfDocument(event, headers, id).toPromise();
    await this.uploadBilderForPdf(event, pdfId, headers, id).toPromise();
    
    const cloudId = await this.closePdf(headers, pdfId).toPromise();
    this.emitUploadDoneEvent(id, event, cloudId)
    
    return cloudId
}

Information: You can find out what happens when you convert an observable into a promise without completing the observable here: Why converted promise from Subject (Observable) does not work as expected

Note: I am meeting your expectations accordingly.

There might be other solutions to the issue that do not go against common best practices.

It is evident that the code poses certain challenges, and the key solution lies in shifting the responsibility of combining results and passing correct arguments from methods to the pipe.

A number of simple improvements can be implemented without much difficulty. tap operator does not alter the value, allowing us to remove unnecessary properties during destructuring. map simply transforms the result, so instead of

map(({ event, headers }) => this.generateUploadId(event, headers)),

We can now write

map(({ event, headers }) => ({
  event,
  headers,
  id: this.generateUploadId(event, headers)
}))

The method this.generateUploadId no longer needs to return an object.

When it comes to high-order mapping operators, several options come to mind. Most 'xMap' operators support a result selector as the final argument, which serves our purpose effectively - combining source value with the result. Result selectors have been deprecated, so nested pipes are currently the preferred approach. However, let's explore how this could look using a result selector

Option 0. Result Selector (deprecated)

this.startUploadEvent$
  .pipe(
    concatMap(
      event => this.getAuthenticationHeaders(event),
      (event, headers) => ({ event, headers }) // <-- Result Selector
    )
  );

Option 1. Nested Pipes (aka "use closures")

This is quite similar to Option 0, but the event is preserved within the closure rather than the inner observable.

this.startUploadEvent$
  .pipe(
    concatMap(
      event => this.getAuthenticationHeaders(event)
        .pipe(map(headers => ({ event, headers })))
    )
  );

Option 2. Custom Operator (Closures here as well)

A custom operator can be created to achieve a syntax resembling Result Selectors

function withResultSelector(operator, transformer) {
  let sourceValue;
  return pipe(
    tap(value => (sourceValue = value)),
    operator,
    map(value => transformer(sourceValue, value))
  );
}

Usage:

this.startUploadEvent$
  .pipe(
    withResultSelector(
      concatMap(event => this.getAuthenticationHeaders(event)),
      (event, headers) => ({ event, headers })
    )
  );

Furthermore, it is feasible to extract repetitive elements and enhance overall functionality:

const mergeAs = propName => (a, b) => ({ ...a, [propName]: b });
const opAndMergeAs = (operator, propName) => withResultSelector(operator, mergeAs(propName));

this.startUploadEvent$
  .pipe(
    opAndMergeAs(concatMap(event => this.getAuthenticationHeaders(event)), "headers")
  );

While defining proper types for these operations might pose some challenges, it presents a separate issue

Here is the playground I utilized while crafting this response.

Your observation about the concerns and issues mentioned is accurate, but the challenge here lies in transitioning your mindset from an imperative approach to a Reactive/Functional one. Let's first examine the imperative code:

private startUpload(event: StartUploadEvent) {
    const headers = this.getAuthenticationHeaders(event)
    const id = this.generateUploadId()

    this.emitUploadStartEvent(id, event)

    const pdfId = this.createPdfDocument(event, headers, id)
    this.uploadBilderForPdf(event, pdfId, headers, id)

    const cloudId = this.closePdf(headers, pdfId)
    this.emitUploadDoneEvent(id, event, cloudId)

    return cloudId
}

In this code snippet, you can see that having the 'event' parameter allows for passing only necessary data to subsequent functions. The goal is to transition this code to a more Reactive/Functional approach.

From my perspective, the main issue is that functions lose their context. For instance, 'getAuthenticationHeaders' should solely return 'headers' without involving 'event.' This often arises when working with RxJS (Reactive Approach), maintaining purity by ensuring that 'pure' operators handle data within the same pipeline, avoiding side effects for predictability.

A potential solution lies in utilizing 'nested pipes,' which I find to be the most effective approach:

concatMap(event => this.getAuthenticationHeaders(event).pipe(
    map(headers => this.generateUploadId(event, headers).pipe())
))

This technique is prevalent in RxJS backend libraries like Marble.js. Other suggestions provided could enhance functionality while retaining code clarity.

While converting to an 'async/await' approach is possible, it sacrifices the reactivity offered by RxJS. I suggest delving into reactive programming concepts and exploring related resources such as CycleJS and Functional Programming materials, including the informative books Mostly adequate guide to FP (in javascript) and Composing Software.

I highly recommend watching the insightful talk on RxJS Recipes as it can revolutionize your approach to using RxJS effectively.

Additional Resources Worth Exploring:

• https://gist.github.com/staltz/868e7e9bc2a7b8c1f754
• https://www.youtube.com/watch?v=vS1-jzngpmw
• https://www.youtube.com/watch?v=uQ1zhJHclvs
• https://www.youtube.com/watch?v=XKfhGntZROQ

Contained within this answer are four different methods, each utilizing four consecutive streams (allowing backend calls to make use of the previous stream's response).

(While some methods may have been previously mentioned in other answers, this comparison is presented using the same input assumptions - where the streams represent HTTP calls with a single emission, thereby allowing operators like mergeMap, concatMap, or switchMap to be used.)

Each method comes with its own set of advantages and disadvantages.

Before delving into the methods, here is the common preparation code:

import { of, map, Observable, exhaustMap } from 'rxjs';

const a$ = of('a');
const b$ = of('b');
const c$ = of('c');
const d$ = of('d');

interface Result {
  a: string;
  b: string; 
  c: string;
  d: string;
} 

Method 1: Nested Pipe Using Map/Previously Known as Result Selector

This approach involves two levels of nested pipes, which may make it slightly less readable and more prone to errors.

function makeBeCalls(): Observable<Result> {
  return a$
      .pipe(
          exhaustMap(a =>
              b$.pipe(
                  map(b => ({b, a}))
              )
          ),
          exhaustMap(({b, a}) =>
              c$.pipe(
                  map(c => ({b, a, c}))
              )
          ),
          exhaustMap(({a, b, c}) =>
              d$.pipe(
                  map(d => ({a, b, c , d}))
              )
          )
      );
}

makeBeCalls().subscribe(v => console.log(v))

Method 2: Using a Shared Variable for Previous Values

In this method, an additional temporary variable is utilized to store subsequent emissions - resulting in a side effect. However, the solution appears shorter and more readable.

function makeBeCalls1(): Observable<Result> {
  const events: Result = { a: null, b: null, c: null, d: null };
  return a$
      .pipe(
          exhaustMap(a => (events.a = a) && b$),
          exhaustMap(b => (events.b = b) && c$),
          exhaustMap(c => (events.c = c) && d$),
          map(d => (events.d = d) && events) 
      );
}

Method 3: Using Nested Pipes/Closure Scope

This particular approach nests each stream, reminiscent of callback hell. It is considered the least readable and potentially more error-prone, especially when dealing with multiple levels of nesting.

function makeBeCalls2(): Observable<Result> {
  return a$
      .pipe(
          exhaustMap(a => b$
              .pipe(
                  exhaustMap(b => c$
                      .pipe(
                          exhaustMap(c => d$
                              .pipe(
                                  map(d => ({a, b, c, d}))
                              )
                          )
                      )
                  )
              )
          )
      );
}

makeBeCalls2().subscribe(v => console.log(v))

Method 4: Utilizing Promises with Async/Await

This approach, while not fully reactive, offers a straightforward and readable solution - especially beneficial for HTTP calls.

async function makeBeCalls3(): Promise<Result> {
  const a = await lastValueFrom(a$);
  const b = await lastValueFrom(b$);
  const c = await lastValueFrom(c$);
  const d = await lastValueFrom(d$);
  
  return {a, b, c, d}
}

makeBeCalls3().then(v => console.log(v))

Visit StackBlitz for a working example.