Implementing forkJoin Functionality in Angular

This challenge involves creating a custom implementation of forkJoin within an Angular application. forkJoin is a powerful RxJS operator that allows you to combine multiple Observables and emit a single value when all of them have completed. This is incredibly useful for scenarios where you need to fetch data from multiple independent sources simultaneously and then process the combined results.

Problem Description

Your task is to create a service in Angular that mimics the behavior of RxJS's forkJoin operator. This service should accept an array of Observables and return a new Observable. This new Observable should emit an array of the last emitted values from each of the input Observables, but only after all of the input Observables have successfully completed.

Key Requirements:

1. Service Creation: Create an Angular service (e.g., CustomForkJoinService).
2. customForkJoin Method: Implement a method within the service named customForkJoin.
3. Input: The customForkJoin method should accept an array of Observables (Observable<any>[]).
4. Output: The method should return an Observable<any[]>.
5. Behavior:
   • The returned Observable should subscribe to all input Observables concurrently.
   • It should collect the last emitted value from each input Observable.
   • It should emit an array containing these collected values only when all input Observables have completed.
   • If any of the input Observables errors, the returned Observable should also error immediately with the error from the first Observable that errored.
   • If the input array is empty, the returned Observable should complete immediately without emitting any value.
6. No RxJS forkJoin Usage: You are explicitly forbidden from using RxJS's built-in forkJoin operator in your implementation. You should leverage other RxJS operators like Observable.create, combineLatest, zip, take, map, catchError, etc., as needed.

Expected Behavior:

When subscribing to the Observable returned by customForkJoin:

• If all input Observables emit values and complete: the subscriber receives an array where each element corresponds to the last value of an input Observable in the original order.
• If any input Observable errors: the subscriber receives the error.
• If the input array is empty: the subscriber completes without any emissions.

Edge Cases:

• Empty Input Array: The customForkJoin method should handle an empty array of Observables gracefully.
• Observables Completing Immediately: Consider scenarios where input Observables complete very quickly.
• Observables Emitting Multiple Values: Ensure only the last value is captured for each Observable upon completion.

Examples

Example 1: Successful Completion

// Assuming this service is injected into a component
const observable1$ = of('Result 1').pipe(delay(100));
const observable2$ = of('Result 2').pipe(delay(50));
const observable3$ = of('Result 3').pipe(delay(150));

this.customForkJoinService.customForkJoin([observable1$, observable2$, observable3$])
  .subscribe(
    (results) => {
      console.log(results); // Expected: ['Result 1', 'Result 2', 'Result 3']
    },
    (error) => {
      console.error("Error:", error);
    }
  );

Input: An array of three Observables, each emitting a string and completing. Output: An Observable that emits ['Result 1', 'Result 2', 'Result 3'] once all input Observables have completed. Explanation: observable2$ completes first, then observable1$, then observable3$. The customForkJoin waits for observable3$ to complete before emitting the array of all last emitted values.

Example 2: Handling Errors

const observableWithError$ = throwError(() => new Error('Something went wrong!')).pipe(delay(80));
const observableSuccess$ = of('Success!').pipe(delay(120));

this.customForkJoinService.customForkJoin([observableSuccess$, observableWithError$])
  .subscribe(
    (results) => {
      console.log(results);
    },
    (error) => {
      console.error("Error:", error.message); // Expected: "Error: Something went wrong!"
    }
  );

Input: An array containing one Observable that errors and another that succeeds. Output: An Observable that immediately errors with the error from observableWithError$. Explanation: observableWithError$ emits an error after 80ms. The customForkJoin Observable immediately propagates this error to its subscriber, and observableSuccess$ is unsubscribed.

Example 3: Empty Input Array

this.customForkJoinService.customForkJoin([])
  .subscribe(
    (results) => {
      console.log("Completed with results:", results); // Expected: "Completed with results: undefined" or no value if subscription handles empty emission differently
    },
    (error) => {
      console.error("Error:", error);
    },
    () => {
      console.log("Completed"); // Expected: "Completed"
    }
  );

Input: An empty array of Observables. Output: An Observable that completes immediately without emitting any value. Explanation: When no Observables are provided, the customForkJoin logic should recognize this and trigger completion immediately.

Constraints

• Language: TypeScript
• Framework: Angular
• RxJS Usage: You may use any RxJS operators except forkJoin.
• Observables: Input Observables can be of any type (Observable<any>).
• Completion Order: The order of emitted values in the output array must match the order of Observables in the input array.
• Error Handling: Errors should be propagated immediately.

Notes

• Consider how you will manage subscriptions to the input Observables.
• Think about how to track the completion of each individual Observable.
• You'll need a mechanism to store the last emitted value from each Observable.
• The use of Observable.create or creating a custom Observable class might be a good starting point.
• Pay close attention to unsubscribing from input Observables when the main customForkJoin Observable completes or errors to prevent memory leaks.