Merging Local and Remote Edits

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

In our last session, we implemented a long-polling mechanism using the expand operator. That pattern was all about creating a sequential chain of asynchronous operations, where each step depends on the completion of the previous one.

Today, we'll tackle a different kind of concurrency: handling multiple, independent, parallel streams of events. Our goal is to model a collaborative editing session by merging a local user input stream with a simulated remote update stream. This is a classic real-world problem that perfectly demonstrates how RxJS can elegantly manage complex, multi-source state.

This lesson will show you how to:

• Use a Subject to represent a stream of local user actions.
• Simulate a stream of incoming remote data.
• Combine these independent streams into one using the merge operator.
• Manage the application's state (the document content) over time using the scan operator.

1. The Architectural Model

Before diving into code, let's visualize the architecture of our collaborative editor.

1. Local Input Stream: The user types into a text area. Each keystroke that changes the content is an event. We'll capture these events in a dedicated stream.
2. Remote Update Stream: Simultaneously, changes from other users arrive over the network. We'll simulate this with a stream that emits "remote change" events periodically.
3. Merge: We need a single, unified stream of all changes, regardless of their origin. The merge operator is perfect for this. It will subscribe to both the local and remote streams and pass through emissions from either one as soon as they occur.
4. State Accumulation: We need to maintain the "current state" of the document. For every change that comes through our merged stream, we must apply it to the document. The scan operator acts like a stateful reducer for observables, making it the ideal tool for this job.
5. UI Update: The new state produced by scan is then used to update the UI, ensuring the user sees their own changes and the remote changes reflected in the text area.

Here is the data flow:

[Local User Input] ---> (local$ Subject) --map--> |
                                                    | --> merge --> scan --> (documentState$) ---> [UI Display]
[Remote Server Push] --> (remote$ Observable) -map-> |

2. Setting Up the Streams

Let's start by creating our two source streams: one for local edits and one for simulated remote edits.

Local Updates via a Subject

A Subject is an excellent way to bridge the imperative world of UI event handlers with the declarative world of RxJS. We'll create a Subject that we can push new document content into whenever the user types.

import { Subject } from 'rxjs';
import { map } from 'rxjs/operators';

// This Subject will act as the source for local user input.
// In a React app, the component's onChange handler would call:
// localUpdate$.next(event.target.value);
const localUpdate$ = new Subject<string>();

// To distinguish local from remote changes, we'll map the raw value
// into a structured object.
const localChange$ = localUpdate$.pipe(
  map(content => ({
    source: 'local',
    content: content
  }))
);

Simulated Remote Updates

To simulate updates from other users, we'll use interval to create a stream that periodically emits a "remote change" object. In a real application, this would be a WebSocket stream. To make it more realistic, our remote change won't be the full document, but a small piece of text to append.

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

// Simulate a remote change arriving every 3 seconds.
const remoteUpdate$ = interval(3000).pipe(
  map(i => ({
    source: 'remote',
    content: ` [remote edit #${i + 1}] `
  }))
);

Now we have two distinct streams: localChange$ and remoteUpdate$. Both emit objects that describe a change and its origin.

3. Combining Streams with merge

Our next task is to combine these two streams into a single timeline of events. We need an operator that emits a value whenever any of its sources emit. This is the exact job of the merge operator.

To understand why merge is the right choice, let's consult a resource that compares the primary combination operators.

As the resource explains, merge is ideal when you need to react to events from multiple sources in the order they occur.

• zip would be wrong because it would require one local edit and one remote edit to happen before emitting anything.
• combineLatest would also be incorrect. It would emit whenever either stream produced a value, but it would always give us the latest value from both. This doesn't represent a sequential log of edits.

Let's apply merge:

import { merge } from 'rxjs';

const allChanges$ = merge(localChange$, remoteUpdate$);

Simple as that. allChanges$ will now emit a value whenever localChange$ or remoteUpdate$ emits.

4. Accumulating State with scan

We now have a single stream of change events. The final step is to use this stream to build and update our document's state over time. The scan operator is designed for this. It works like Array.prototype.reduce(), but for observables: it takes an initial value and an accumulator function. For each item from the source observable, it runs the function, and the return value becomes the new accumulated state.

import { scan } from 'rxjs/operators';

const initialDocument = "Hello, world!";

const documentState$ = allChanges$.pipe(
  scan((currentDoc, change) => {
    console.log('Applying change:', change);
    
    // The logic depends on the source of the change
    if (change.source === 'local') {
      // For local changes, the content is the new state of the document.
      return change.content;
    }
    
    if (change.source === 'remote') {
      // For remote changes, we append the content.
      // A real app would use a more sophisticated patching algorithm (e.g., Operational Transform or CRDTs).
      return currentDoc + change.content;
    }

    return currentDoc;
  }, initialDocument)
);

The documentState$ observable now represents the state of our document over time. It starts with initialDocument and updates itself every time a local or remote change occurs.

5. Putting It All Together

Let's see the complete example in action. We'll subscribe to the final documentState$ and manually trigger some local updates to simulate a user typing.

import { Subject, interval, merge } from 'rxjs';
import { map, scan, startWith } from 'rxjs/operators';

// --- Stream Definitions ---

// 1. Local updates
const localUpdate$ = new Subject<string>();
const localChange$ = localUpdate$.pipe(
  map(content => ({ source: 'local', content }))
);

// 2. Remote updates
const remoteUpdate$ = interval(4000).pipe(
  map(i => ({ source: 'remote', content: ` [remote edit #${i + 1}] ` }))
);

// 3. Merged stream
const allChanges$ = merge(localChange$, remoteUpdate$);

// --- State Management ---

const initialDocument = "Start typing...";

const documentState$ = allChanges$.pipe(
  scan((currentDoc, change) => {
    if (change.source === 'local') {
      return change.content;
    }
    if (change.source === 'remote') {
      // To correctly handle remote edits, we need to know the cursor position.
      // For this simplified model, we'll just append.
      return currentDoc + change.content;
    }
    return currentDoc;
  }, initialDocument),
  // Ensure the initial state is emitted immediately upon subscription
  startWith(initialDocument)
);

// --- Subscription (The "UI") ---

console.log("--- Collaborative Editor Simulation ---");
console.log("The 'UI' will now display document state changes.");
console.log("Simulating local user typing via localUpdate$.next().\n");

documentState$.subscribe(documentContent => {
  console.log(`[UI Display] \n${documentContent}\n`);
});

// --- Simulate User Actions ---

// User types "Hello"
setTimeout(() => {
  console.log(">>> User types: 'Hello'");
  localUpdate$.next('Hello');
}, 1000);

// User types "Hello RxJS"
setTimeout(() => {
  console.log(">>> User types: 'Hello RxJS'");
  localUpdate$.next('Hello RxJS');
}, 2500);

// A remote edit will arrive at ~4000ms

// User types "Hello RxJS!"
setTimeout(() => {
  console.log(">>> User types: 'Hello RxJS!'");
  localUpdate$.next('Hello RxJS!');
}, 5000);

// Another remote edit will arrive at ~8000ms

If you run this code, you will see the UI display updates from both local typing and the periodic remote edits, processed in the order they occur. This demonstrates how merge and scan work together to create a reactive state machine from multiple event sources.

---

Conclusion

You've just built the core logic for a real-time collaborative application. This pattern is incredibly versatile and extends far beyond text editors. You can use it for any feature that needs to combine user actions with asynchronous data from a server, like a real-time dashboard, a chat application, or a multiplayer game.

Key Takeaways:

• merge is the operator of choice for combining multiple independent event streams into a single stream, processing events as they arrive.
• scan is the canonical operator for managing state over time by applying a reducer function to each event in a stream.
• By tagging stream sources with map before merging, you can implement different logic for each source within your scan accumulator.
• This merge + scan pattern is a fundamental building block for creating complex, multi-source reactive state management systems.

Next Lesson Preview:

We've now constructed a fairly complex observable chain involving multiple sources and stateful logic. What happens when it doesn't behave as expected? In our next and final lesson of this module, we will dive into debugging complex Observable chains using advanced techniques like custom logging operators and RxJS DevTools.