From direct calls to observers
Think of subscribing to a YouTube channel. The channel doesn’t keep a list of people to phone when a new video is out. You press Subscribe, and when a video is published, every subscriber gets a notification. You can unsubscribe at any time, and the channel doesn’t care how many subscribers it has or who they are.
In code:
- the subject (the cart) keeps a list of observers, also called listeners or subscribers,
- observers subscribe to be told about changes, and unsubscribe when they stop caring,
- when something changes, the subject notifies every observer on the list.
Observer decouples a subject from an open set of dependents: the subject publishes changes through a narrow interface, and observers register and unregister at runtime. You rarely hand-roll it today; it’s the foundation under NotificationCenter, KVO, Combine and the Observation framework on Apple platforms, listeners, LiveData, Flow/StateFlow and Compose state on Android, and java.beans, java.util.concurrent.Flow, Reactor and Spring application events on the server; in TypeScript, the DOM’s addEventListener, Node’s EventEmitter, RxJS and the signals in modern UI frameworks.
The design questions are the same everywhere: state or events (does a late subscriber get the current value?), push or pull, which thread observers run on and under which locks, and who owns the subscription’s lifetime. The Detailed version goes through each, with the production bugs they cause.
Before: The cart knows every screen
final class Cart {
private(set) var items: [Product] = []
weak var badge: CartBadgeView?
weak var cartScreen: CartViewController?
weak var checkoutButton: CheckoutButton?
func add(_ product: Product) {
items.append(product)
badge?.update(count: items.count)
cartScreen?.reload()
checkoutButton?.isEnabled = true
Analytics.log("cart_add", product.id)
}
}class Cart(
private val badge: CartBadge,
private val cartScreen: CartScreen,
private val checkoutButton: CheckoutButton,
) {
private val items = mutableListOf<Product>()
fun add(product: Product) {
items += product
badge.update(items.size)
cartScreen.reload()
checkoutButton.isEnabled = true
Analytics.log("cart_add", product.id)
}
}class Cart {
private final List<Product> items = new ArrayList<>();
private final CartBadge badge;
private final CartScreen cartScreen;
private final CheckoutButton checkoutButton;
Cart(CartBadge badge, CartScreen cartScreen, CheckoutButton checkoutButton) {
this.badge = badge; this.cartScreen = cartScreen; this.checkoutButton = checkoutButton;
}
void add(Product product) {
items.add(product);
badge.update(items.size());
cartScreen.reload();
checkoutButton.setEnabled(true);
Analytics.log("cart_add", product.id());
}
}class Cart {
private readonly items: Product[] = [];
constructor(
private readonly badge: CartBadge,
private readonly cartScreen: CartScreen,
private readonly checkoutButton: CheckoutButton,
) {}
add(product: Product) {
this.items.push(product);
this.badge.update(this.items.length);
this.cartScreen.reload();
this.checkoutButton.enabled = true;
analytics.log("cart_add", product.id);
}
}After: The cart announces changes; listeners decide
final class Cart {
private(set) var items: [Product] = []
private var observers: [UUID: ([Product]) -> Void] = [:]
/// Returns a token; the subscription ends when the token is cancelled.
func observe(_ onChange: @escaping ([Product]) -> Void) -> Subscription {
let id = UUID()
observers[id] = onChange
onChange(items) // send the current state right away
return Subscription { [weak self] in self?.observers[id] = nil }
}
func add(_ product: Product) {
items.append(product)
observers.values.forEach { $0(items) }
}
}
final class Subscription {
private let onCancel: () -> Void
init(_ onCancel: @escaping () -> Void) { self.onCancel = onCancel }
func cancel() { onCancel() }
deinit { onCancel() } // released token = unsubscribed
}
// Any screen can listen, and the cart doesn't know it exists:
subscription = cart.observe { [weak self] items in
self?.badgeLabel.text = "\(items.count)"
}class Cart {
private val items = mutableListOf<Product>()
private val listeners = CopyOnWriteArrayList<(List<Product>) -> Unit>()
fun addListener(listener: (List<Product>) -> Unit): () -> Unit {
listeners += listener
listener(items.toList()) // send the current state right away
return { listeners -= listener } // call it to unsubscribe
}
fun add(product: Product) {
items += product
val snapshot = items.toList()
listeners.forEach { it(snapshot) }
}
}
// Any screen can listen, and the cart doesn't know it exists:
val unsubscribe = cart.addListener { items -> badge.text = items.size.toString() }interface CartListener {
void cartChanged(List<Product> items);
}
class Cart {
private final List<Product> items = new ArrayList<>();
private final List<CartListener> listeners = new CopyOnWriteArrayList<>();
Runnable addListener(CartListener listener) {
listeners.add(listener);
listener.cartChanged(List.copyOf(items)); // send the current state right away
return () -> listeners.remove(listener); // run it to unsubscribe
}
void add(Product product) {
items.add(product);
List<Product> snapshot = List.copyOf(items);
listeners.forEach(l -> l.cartChanged(snapshot));
}
}
// Any screen can listen, and the cart doesn't know it exists:
Runnable unsubscribe = cart.addListener(items -> badge.setText(String.valueOf(items.size())));type CartListener = (items: readonly Product[]) => void;
class Cart {
private items: Product[] = [];
private readonly listeners = new Set<CartListener>();
/** Returns a function; call it to unsubscribe. */
addListener(listener: CartListener): () => void {
this.listeners.add(listener);
listener([...this.items]); // send the current state right away
return () => this.listeners.delete(listener);
}
add(product: Product) {
this.items = [...this.items, product];
const snapshot = [...this.items];
for (const listener of this.listeners) listener(snapshot);
}
}
// Any component can listen, and the cart doesn't know it exists:
const unsubscribe = cart.addListener((items) => (badge.textContent = String(items.length)));What changed:
- The cart knows nothing about screens. It has a list of listeners and calls them. A new feature (say, a “free shipping at 50 EUR” banner) subscribes on its own, and the cart doesn’t change.
- Listeners get a snapshot, a copy of the items, so they can’t change the cart’s list by accident.
- Every subscription can end. Subscribing returns something you call (or release) to unsubscribe. That turns out to be the most important part, as the next bug shows.
Observer in the platforms you use
You’ve already used this pattern, maybe without the name:
| Platform | Observer in practice |
|---|---|
| iOS | NotificationCenter, @Observable and SwiftUI updates, Combine’s @Published and sink, KVO, delegates with one observer |
| Android | Click listeners, LiveData.observe, StateFlow.collect, Compose recomposition when state changes |
| Java server | Spring @EventListener and ApplicationEventPublisher, PropertyChangeListener, message queues (observers in other processes) |
| TypeScript | addEventListener in the browser, Node’s EventEmitter, RxJS subscribe, React re-rendering when state changes, signals |
Choosing the observable type
| Need | Swift | Kotlin | Java / Spring | TypeScript |
|---|---|---|---|---|
| UI state with a current value | @Observable (pull-based tracking), @Published | StateFlow (conflated, dedups with equals) | PropertyChangeSupport, Reactor Sinks.many().replay().latest() | RxJS BehaviorSubject, signals, a store read with useSyncExternalStore |
| One-off events, each delivered once | AsyncStream, PassthroughSubject | Channel + receiveAsFlow(), SharedFlow(replay = 0) | ApplicationEventPublisher, Sinks.many().multicast() | EventEmitter, EventTarget, RxJS Subject |
| Cross-module broadcast | NotificationCenter (stringly typed) | a shared SharedFlow in a DI-scoped bus | Spring events, a message broker between services | BroadcastChannel (between tabs), a module-level emitter |
| Backpressure | AsyncStream buffering policy | buffer, conflate, collectLatest | java.util.concurrent.Flow / Reactive Streams request(n) | RxJS throttleTime, sampleTime, bufferTime; async iterators |
The Observation framework is worth calling out: @Observable doesn’t push values at all. SwiftUI (or withObservationTracking) records which properties a closure read, and gets a single “will change” callback when one of them changes, then re-reads. Pull-based tracking means no subscriptions to manage and no over-notification, but also no stream of values: if you need every intermediate value, use an AsyncStream.
Threads, locks and reentrancy
Synchronous observers run inside the subject’s call: on its thread, under its locks, before it returns. Consequences:
- Notify after releasing locks (puzzle below), on a documented thread or executor. UI observers must hop to the main thread; on Apple platforms, isolate UI models to
@MainActor. - An observer that calls back into the subject sees a half-updated state if notification happens mid-mutation. Mutate fully, then notify.
- One slow or throwing observer shouldn’t break the others: isolate failures per observer (catch and log) or dispatch asynchronously.
- In JavaScript there are no locks to hold, but the reentrancy point stands: an observer that calls back into the subject runs in the middle of the update. And an observer that throws stops the loop for the ones after it, unless you catch per observer.
- Ordering between observers is unspecified unless you specify it. Don’t build features that rely on “the analytics listener runs first”.
Notifying under a lock
Tap the line with the bug.
final class Inventory: @unchecked Sendable {
private let lock = NSLock()
private var stock: [String: Int] = [:]
private var observers: [(String, Int) -> Void] = []
func remove(_ sku: String, quantity: Int) {
lock.withLock {
let left = (stock[sku] ?? 0) - quantity
stock[sku] = left
observers.forEach { $0(sku, left) }
}
}
}class Inventory {
private val lock = ReentrantLock()
private val stock = HashMap<String, Int>()
private val observers = CopyOnWriteArrayList<(String, Int) -> Unit>()
fun remove(sku: String, quantity: Int) = lock.withLock {
val left = (stock[sku] ?: 0) - quantity
stock[sku] = left
observers.forEach { it(sku, left) }
}
}class Inventory {
private final Map<String, Integer> stock = new HashMap<>();
private final List<BiConsumer<String, Integer>> observers = new CopyOnWriteArrayList<>();
synchronized void remove(String sku, int quantity) {
int left = stock.getOrDefault(sku, 0) - quantity;
stock.put(sku, left);
observers.forEach(o -> o.accept(sku, left));
}
}Hint: One observer is the pricing service. When stock is low, it raises the price, which takes the pricing lock and reads the inventory from another thread.
Observers run while the inventory lock is held. That’s foreign code under your lock:
- Deadlock: thread A holds the inventory lock and calls the pricing observer, which waits for the pricing lock. Thread B holds the pricing lock and calls
inventory.remove, which waits for the inventory lock. Both wait forever, and it only happens under load. - Throughput: every observer’s work (logging, a network call, a database write) now extends the critical section, so all writers queue behind the slowest observer.
- Reentrancy: an observer calling
removeon the same thread re-enters the lock (ReentrantLockandsynchronizedallow it;NSLockdeadlocks immediately) and sees a state mid-update.
Change the state under the lock, then notify outside it:
func remove(_ sku: String, quantity: Int) {
let (left, toNotify) = lock.withLock {
let left = (stock[sku] ?? 0) - quantity
stock[sku] = left
return (left, observers) // copy the list while locked
}
toNotify.forEach { $0(sku, left) } // foreign code runs unlocked
}fun remove(sku: String, quantity: Int) {
val left = lock.withLock {
((stock[sku] ?: 0) - quantity).also { stock[sku] = it }
}
observers.forEach { it(sku, left) } // foreign code runs unlocked
}void remove(String sku, int quantity) {
int left;
synchronized (this) {
left = stock.getOrDefault(sku, 0) - quantity;
stock.put(sku, left);
}
observers.forEach(o -> o.accept(sku, left)); // foreign code runs unlocked
}Notifications may now arrive out of order between threads (two removals, notified in the opposite order). If observers need ordering, include a version number, or deliver through a single-threaded executor or an actor.
- Notify with a snapshot of the data and of the listener list, after the mutation is complete and outside any lock, on a documented thread.
- Synchronous observers run inside the subject’s call: its thread, its locks, before it returns. Isolate slow or throwing observers.
| Swift | Kotlin | Java | TypeScript | |
|---|---|---|---|---|
| Hand-rolled subscription | token (AnyCancellable, your own) | returned unsubscribe function | returned handle / remove…Listener | returned unsubscribe function |
| Platform observers | @Observable, @Published, NotificationCenter | StateFlow, SharedFlow, listeners | Spring events, PropertyChangeSupport, Flow.Publisher | EventTarget, Node EventEmitter, RxJS, signals |
Swift
@Observable is pull-based: SwiftUI records which properties a view read and gets one “will change” callback, with no subscriptions to manage and no stream of values (use AsyncStream for that).
TypeScript
There are no locks, but reentrancy remains: an observer that calls back into the subject during notification sees a half-updated state. Mutate fully, then notify.
Check this part
After the refactor, what does the cart know about the screens that show it?
Why do listeners receive a copy (snapshot) of the items?
Observers are called while the subject holds its lock. What can go wrong?Observers are called inside a
synchronizedmethod. What can go wrong?JavaScript has no threads here. What can still go wrong when a subject notifies observers in the middle of an update?Where do synchronous observers run?