Implementing Long-Polling with `expand`

Hello! Welcome to the next lesson in our "Real-World Architectural Patterns" module.

In our previous lesson, we built a robust, time-based caching layer using shareReplay and a timer to periodically refresh data. That pattern is excellent for keeping data fresh on a fixed schedule. However, some scenarios require a different approach.

Today, we'll tackle a classic problem: tracking the status of a long-running process, like a data export or a video encoding job. We don't want to poll on a fixed interval, which could either be too slow or hammer the server unnecessarily. Instead, we want to make a request and, as soon as it completes, make the next one. This is known as long-polling.

Our learning outcome is to implement a long-polling mechanism that re-initiates a request only after the previous one completes, using the expand operator. This pattern is highly efficient and showcases the power of RxJS for managing complex, sequential asynchronous workflows.

1. Polling vs. Long-Polling

Let's clarify the distinction between the two main polling strategies:

• Interval Polling (what we used for our cache): A request is sent at a fixed interval (e.g., every 10 seconds). This is managed by an external clock, like timer or interval.

  • Pro: Simple to implement.
  • Con: Inefficient. If a request takes 8 seconds, the next one fires just 2 seconds later. If it takes 12 seconds, requests can overlap (unless managed with switchMap) or drift.

• Long-Polling: A request is sent. The client waits for the response. Upon receiving the response, it immediately (or after a short, fixed delay) sends the next request. The "clock" is the completion of the previous request itself.

  • Pro: Highly efficient. It adapts to network latency and server response time. No overlapping requests.
  • Con: Can be slightly more complex to implement imperatively.

This "request, then upon completion, request again" flow is fundamentally a recursive pattern. In RxJS, the operator designed specifically for this kind of asynchronous recursion is expand.

2. Introducing the expand Operator

The expand operator is one of the most powerful in RxJS, but its use case is very specific: it creates an output Observable by recursively calling a function that returns an Observable.

Here’s the mental model:

1. An initial (source) Observable emits a value.
2. expand receives this value and passes it to a project function you provide.
3. Your project function returns a new Observable (e.g., another HTTP request).
4. expand subscribes to this new Observable.
5. Values emitted by this new Observable are:
   a. Pushed to the downstream observer (the final subscribe block).
   b. Fed back into the expand operator, starting the process over again.

This creates a feedback loop that continues as long as the project function keeps returning new Observables that emit values.

To stop the recursion for a given branch, the project function must return an Observable that completes without emitting a value, such as EMPTY.

3. Implementing Long-Polling with expand

Let's see how to apply this. We'll build a poller that checks the status of a simulated background job. The job will start as 'PENDING' and eventually become 'COMPLETE'.

First, let's look at a real-world problem description and a concise solution.

Inspired by that solution, let's build our own long-polling mechanism from scratch.

Step-by-Step Implementation

Imagine we have an API function checkJobStatus() that returns an Observable.

import { of, timer, EMPTY } from 'rxjs';
import { expand, delay, switchMap, tap } from 'rxjs/operators';

// 1. Simulate the API call
// This function simulates checking a job status.
// After 3 attempts, the job is considered 'COMPLETE'.
let attempts = 0;
function checkJobStatus() {
    attempts++;
    console.log(`[API] Checking job status (Attempt ${attempts})...`);
    
    const currentStatus = (attempts >= 4) ? 'COMPLETE' : 'PENDING';

    // Simulate an API call with a 1-second network delay
    return of({ status: currentStatus, data: `Some data from attempt ${attempts}` }).pipe(
        delay(1000)
    );
}

const POLL_INTERVAL = 2000; // 2 seconds between the completion of one poll and the start of the next.

// 2. Start the long-polling chain with an initial request
console.log("Starting long-poll...");
checkJobStatus().pipe(
    // 3. Use `tap` to inspect the response from the current request
    tap(response => console.log(`[Client] Response received:`, response)),

    // 4. Use `expand` to recursively call the next poll
    expand(response => {
        // 5. Check the condition to stop polling
        if (response.status === 'COMPLETE') {
            console.log('[Client] Job is complete. Stopping poll.');
            // Returning EMPTY stops the recursion for this branch.
            return EMPTY; 
        }
        
        // 6. If not complete, schedule the next poll
        console.log(`[Client] Job is pending. Waiting ${POLL_INTERVAL}ms before next poll...`);
        // We return a new observable that waits for the interval, then makes the next API call.
        // This observable's emission will be fed back into `expand`.
        return timer(POLL_INTERVAL).pipe(
            switchMap(() => checkJobStatus())
        );
    })
).subscribe({
    next: value => console.log('-> Processing value:', value),
    error: err => console.error('Something went wrong:', err),
    complete: () => console.log('-> Polling stream complete.')
});

Deconstructing the Code:

• Step 1 & 2: We define our async operation (checkJobStatus) and trigger it once to start the entire chain.
• Step 3 (tap): This is for logging. It shows us the response as soon as it arrives, before the expand logic is evaluated.
• Step 4 (expand): This is the heart of the pattern. It takes the response from the previous checkJobStatus() call.
• Step 5 (Termination Condition): Inside expand, we check if response.status === 'COMPLETE'. If it is, we return EMPTY. EMPTY is a special Observable that completes immediately without emitting any values. This tells expand to terminate this recursive path. Since there's only one path, the entire stream completes.
• Step 6 (Recursion Step): If the job is not complete, we return a new Observable. This Observable first waits for POLL_INTERVAL milliseconds using timer(), then uses switchMap to make the next call to checkJobStatus(). The value emitted by checkJobStatus() will be sent to the final subscriber and fed back into expand to continue the loop.

This pattern perfectly guarantees that the next poll is only initiated after the previous one has fully completed and a specified delay has passed.

4. expand vs. retry vs. repeat

It's useful to contrast expand with other operators that can cause re-subscription:

• retry(n): Resubscribes to the same source Observable n times, but only upon error. It's for resilience against failures, not for sequential polling.
• repeat(n): Resubscribes to the same source Observable n times, but only upon completion. It's for re-running a finite stream, not for chaining calls based on their emitted values.
• expand(...): Resubscribes to a new Observable derived from the previous emission. It's for building dynamic, recursive chains where each step can be different from the last. This makes it perfect for long-polling where you might pass a cursor or timestamp from the last response into the next request.

Conclusion

You have now mastered a powerful and efficient pattern for handling long-running asynchronous tasks. While WebSockets are often the go-to for real-time updates, long-polling with expand is a robust, simpler-to-implement alternative that relies only on standard HTTP requests.

Key Takeaways:

• Long-polling initiates a new request only after the previous one completes, making it more efficient and robust than fixed-interval polling.
• The expand operator is the canonical RxJS tool for implementing asynchronous recursive patterns like long-polling.
• The function passed to expand receives the value from the previous step and must return a new Observable representing the next step.
• To terminate the polling loop, return the EMPTY observable from within expand when a stop condition is met.

Next Lesson Preview:

We've just built a powerful pattern for fetching a sequence of data from a server. In our next lesson, we'll shift our focus to combining data streams from different sources in real-time. We will model a collaborative editing session by merging a local user input stream with a simulated remote update stream. This will introduce us to operators like merge and patterns for managing state from multiple asynchronous inputs.