Reactive from scratch — building an Observer

“Reactive” is a loaded word, but at its core it’s a variation on the Observer pattern: instead of the caller waiting for a result, the result arrives at the caller. This tutorial builds an Observer in plain Core Java, finds its limits the hard way, and ends with a clear motivation for CompletableFuture — the JDK’s far more capable version of the same idea.

A 30-line Observable

A keyed map from consumer-IDs to consumers is all you need to start:

public class Observable<KEY, VALUE> {
    private final Map<KEY, Consumer<VALUE>> listeners = new ConcurrentHashMap<>();

    public void register(KEY key, Consumer<VALUE> listener) {
        listeners.put(key, listener);
    }
    public void unregister(KEY key) {
        listeners.remove(key);
    }
    public void sendEvent(VALUE event) {
        listeners.values().forEach(c -> c.accept(event));
    }
}

var bus = new Observable<String, String>();
bus.register("a", System.out::println);
bus.register("b", System.out::println);

bus.sendEvent("Hello World");      // printed twice
bus.unregister("a");
bus.sendEvent("Hello again");      // printed once

It works. It’s even concurrent-safe at the registration layer thanks to ConcurrentHashMap. But it has a problem nobody talks about until production: memory leaks.

The memory leak problem

Promote the bus to a static registry, sprinkle register calls around the codebase, and watch what happens when callers forget to unregister:

public class Registry {
    private static final Observable<String, String> bus = new Observable<>();
    public static void register(String key, Consumer<String> c) { bus.register(key, c); }
    public static void unregister(String key)                   { bus.unregister(key); }
    public static void sendEvent(String input)                  { bus.sendEvent(input); }
}

Now imagine a Vaadin component registering itself with Registry. The consumer captures the component instance, the component captures its UI, the UI captures its session — and Registry is static, so the GC can never reclaim any of it. The session ends, the component is “gone,” and somewhere in the static map there is still a reference that keeps the entire session graph alive.

finalize() is not the answer. The component needs an explicit lifecycle.

Self-unsubscribe via `Registration`

Make the register method return a handle the caller can use to detach later:

public interface Registration {
    void remove();
}

public class Observable<KEY, VALUE> {
    private final Map<KEY, Consumer<VALUE>> listeners = new ConcurrentHashMap<>();

    public Registration register(KEY key, Consumer<VALUE> listener) {
        listeners.put(key, listener);
        return () -> listeners.remove(key);
    }
    public void sendEvent(VALUE event) {
        listeners.values().forEach(c -> c.accept(event));
    }
}

Now the caller holds the Registration for as long as it wants events and calls .remove() from its own lifecycle hook — detach, @PreDestroy, close, whatever fits:

Registration r = Registry.register("a", System.out::println);
Registry.sendEvent("Hello");
r.remove();                        // detach, the component is now collectable
Registry.sendEvent("Hello again"); // r no longer receives

This is the smallest design change that fixes the leak: ownership of the subscription returns to the subscriber.

Chaining observers

So far we’ve delivered one event to many consumers. Real pipelines need consumers that transform and forward events to a next stage. With our current Observer, you wire that by registering each stage’s input as the previous stage’s output:

var stageA = new Observable<String, String>();
var stageB = new Observable<String, String>();
var stageC = new Observable<String, String[]>();

stageA.register("up",    s -> stageB.sendEvent(s.toUpperCase()));
stageB.register("split", s -> stageC.sendEvent(s.split(" ")));
stageC.register("sink",  parts -> System.out.println(Arrays.toString(parts)));

stageA.sendEvent("Hello reactive world");

It works. But notice: you defined the pipeline backwards (sink first, source last) and the relationship between stages is implicit — to understand what happens when you push into stageA, you have to read every other register call to follow the chain.

Where this design breaks

A handful of questions reveal the limits:

How do you detach an entire subtree of consumers without manually iterating?
How do you get the final result back to the caller? The chain ends in a Consumer, not a value — every termination needs an external collection.
How do you parallelize stages? The basic Observable is synchronous in sendEvent; consumers run on whatever thread calls sendEvent.
How do you replace a stage at runtime?
How do you handle failures in a stage without poisoning every downstream consumer?

You can build all of this. People have. The result is usually a small framework that closely resembles… java.util.concurrent.CompletableFuture.

Why `CompletableFuture` is the answer

CompletableFuture<T> is the JDK’s solution to exactly these problems:

It carries a value (or a failure) — you don’t need an external ArrayList<String> to capture the result.
It chains via thenApply / thenCompose / thenCombine — every stage receives the previous stage’s output as a parameter, not as a side channel.
It runs on an Executor — you choose where each stage executes.
Failures travel through the chain as CompletionException — one chain, one error path.
A chain returned by a function is detachable by going out of scope — no static registry to leak from.

The Observable you just built is useful for understanding what reactive means. For anything beyond a teaching exercise, reach for CompletableFuture instead.

Recap

Reactive ≈ Observer with a value, not a side effect.
A naïve static Observer pattern leaks memory; the fix is returning a Registration the caller owns.
Chaining observers manually is verbose, defined backwards, hard to parallelize and hard to fail safely.
CompletableFuture solves all of these with a couple of method calls. It’s the version you actually ship.

CompletableFuture for the impatient — the JDK primitive, in 10 minutes.
Async pipelines with CompletableFutureQueue — pipelines as values, applied later, on the executor of your choice.