--- title: 'Reactive Streams' source: 'https://academia.sh/en/courses/data-networking-and-concurrency/reactive-streams' course: 'Data, Networking and Concurrency' language: en updated: '2026-10-21T05:03:08+00:00' license: 'CC BY-SA 4.0' objectives: - 'Tells publisher, subscriber, and operator apart in a reactive stream' - 'Measures that subscribing twice under the same name to an unshared stream repeats the request' - 'Measures that on a shared stream the subscription does not repeat the request and the operation stays robust' - 'Shows that the pair of values determines the class of the operator that drops consecutive repeats' --- # Reactive Streams A measurement showing that a repeated subscription to a publisher repeats the request, that a shared stream stays robust, and that the operator dropping consecutive repeats depends on the pair of values. In the previous lesson, appending to a queue came out sensitive to both repetition and order, and assigning a value came out sensitive only to order. In this lesson we look at a third form of concurrent code: reactive streams, in which values flow through time and many places listen to them. The core file is the same in every lesson of the topic. ```swift // Replay.swift — classifies an operation's sensitivity to repetition and order struct Trial { let label: String let once: State let twice: State let forward: State let backward: State var sensitivity: String { switch (once != twice, forward != backward) { case (false, false): return "robust" case (true, false): return "repeat-sensitive" case (false, true): return "order-sensitive" case (true, true): return "doubly-sensitive" } } func report() -> String { print("\(label) | once=\(once) twice=\(twice) ab=\(forward) ba=\(backward) -> \(sensitivity)") return sensitivity } } func replay(_ label: String, from start: State, _ a: (State) -> State, then b: (State) -> State) -> Trial { Trial(label: label, once: a(start), twice: a(a(start)), forward: b(a(start)), backward: a(b(start))) } func countSensitivities(_ results: [String]) { for name in ["robust", "repeat-sensitive", "order-sensitive", "doubly-sensitive"] { print("\(name): \(results.filter { $0 == name }.count)") } } ``` In this lesson the operations write only the information of a stream to the state and are not asynchronous; so they are measured with `replay`. The model builds what the operation does to the state; it is not the network or the framework that runs the stream itself. ## Publisher, Subscriber, and Operator In a **reactive stream**, a **publisher** produces values, a **subscriber** receives those values, and **operators** in between transform or filter the values. A `map` operator transforms every value, and a `removeDuplicates` operator drops consecutive repeats. A publisher can be cold: it starts its work when subscribed to, for example it sends the network request. If the publisher is shared, the work starts only at the first subscriber, and later subscribers get the same result. This distinction has a measurable consequence. Each time an unshared stream is subscribed to, the request goes out again. On a shared stream, the second subscriber does not repeat the request. ```swift // Feed.swift — subscribing to a stream and an operator that drops consecutive repeats struct Delivery: Equatable, CustomStringConvertible { var requests: Int var listeners: [String] var description: String { "requests=\(requests) listeners=\(listeners)" } } func subscribing(_ name: String) -> (Delivery) -> Delivery { { state in var next = state next.requests += 1 if !next.listeners.contains(name) { next.listeners = (next.listeners + [name]).sorted() } return next } } func subscribingShared(_ name: String) -> (Delivery) -> Delivery { { state in var next = state if next.listeners.isEmpty { next.requests += 1 } if !next.listeners.contains(name) { next.listeners = (next.listeners + [name]).sorted() } return next } } struct Output: Equatable, CustomStringConvertible { var emitted: [Int] var description: String { "\(emitted)" } } func emitting(_ value: Int) -> (Output) -> Output { { state in state.emitted.last == value ? state : Output(emitted: state.emitted + [value]) } } ``` `subscribing` builds the unshared subscription: each call raises the request count by one and adds the name to the listeners. `subscribingShared` builds the shared one: the request count rises only while there is no listener. The listeners are kept sorted, because the order of subscribers should not enter the measurement; this is a design choice and is mentioned in CC19. `emitting` builds the operator that drops consecutive repeats: if the new value equals the last value, it leaves the output untouched. ## Why Streams Exist If a value is computed once and returned, an asynchronous function is enough. Streams are for cases where the value arrives many times over time: every letter in a search box, location updates, events from a server. In these cases, instead of "get the value", you need to say "do this as values arrive", and a stream is the way of building this second sentence. There is a contract between publisher and subscriber: the publisher gives the values in order, the subscriber handles them in the order it receives them, and the operators in between pass each value on, transform it, or drop it. This contract leaves two things to the subscriber. The first is the lifetime of the subscription: values arrive as long as the subscription is held, and the stream ends when it is released. The second is the cost of the subscription: every subscription, if the publisher is cold, restarts the work behind it. The questions of the earlier lessons continue here unchanged; sensitivity to repetition and order has moved to the way values arrive. ## Subscription and Sharing First the same name subscribes twice. Then two different names subscribe, to both an unshared and a shared stream. ```swift // main.swift — unshared and shared subscription let fresh = Delivery(requests: 0, listeners: []) let cold = replay("cold-same-name", from: fresh, subscribing("alice"), then: subscribing("alice")).report() let sharedSame = replay("shared-same-name", from: fresh, subscribingShared("alice"), then: subscribingShared("alice")).report() let sharedPair = replay("shared-two-names", from: fresh, subscribingShared("alice"), then: subscribingShared("bob")).report() countSensitivities([cold, sharedSame, sharedPair]) ``` ``` cold-same-name | once=requests=1 listeners=["alice"] twice=requests=2 listeners=["alice"] ab=requests=2 listeners=["alice"] ba=requests=2 listeners=["alice"] -> repeat-sensitive shared-same-name | once=requests=1 listeners=["alice"] twice=requests=1 listeners=["alice"] ab=requests=1 listeners=["alice"] ba=requests=1 listeners=["alice"] -> robust shared-two-names | once=requests=1 listeners=["alice"] twice=requests=1 listeners=["alice"] ab=requests=1 listeners=["alice", "bob"] ba=requests=1 listeners=["alice", "bob"] -> robust robust: 2 repeat-sensitive: 1 order-sensitive: 0 doubly-sensitive: 0 ``` On the unshared stream the second subscription sends the request again: `once` is 1, `twice` is 2. The listener list stays the same, but the request count rises. The operation is **repeat-sensitive**. On the shared stream the request count rises only at the first subscriber; subscribing again under the same name or adding another name does not change the request. Because the listeners are kept sorted, the two orders give the same list, and both shared operations are **robust**. The count is two robust, one repeat-sensitive. **CC19.** The listener list was kept sorted; a stream in which the order of subscribers changes the result was not measured. Whether a shared stream restarts after the first subscriber leaves was not measured either. ## Dropping Consecutive Repeats The `removeDuplicates` operator does not pass a value on if it is the same as the last value. We measure the operator with two values: 3 and 5. ```swift // main.swift — the operator that drops consecutive repeats let fresh = Output(emitted: []) let distinct = replay("drop-repeat", from: fresh, emitting(3), then: emitting(5)).report() countSensitivities([distinct]) ``` ``` drop-repeat | once=[3] twice=[3] ab=[3, 5] ba=[5, 3] -> order-sensitive robust: 0 repeat-sensitive: 0 order-sensitive: 1 doubly-sensitive: 0 ``` The second 3 is dropped, so `once` and `twice` are the same. But the order in which the values arrive is the order of the output: `ab` is `[3, 5]`, `ba` is `[5, 3]`. The operation is **order-sensitive**. The count is one order-sensitive. **CC20.** The measurement models only the operator that drops consecutive repeats. Time-dependent operators such as debounce were not measured, because their result depends on timing. ## Accumulating and Latest-Value Operators Two operators are used a great deal in streams. One is the operator that accumulates the incoming values and publishes the total at each step; counters, cart totals, and unread message counts are built this way. The other is the operator that keeps only the last value; the value of a field shown on screen, or the current state of a setting. Both are measured with the same two values (3 and 5) and the same four runs. The state here is a plain integer. ```swift // main.swift — an accumulating and a latest-value operator let running = replay("running-sum", from: 0, { $0 + 3 }, then: { $0 + 5 }).report() let latest = replay("latest-value", from: 0, { _ in 3 }, then: { _ in 5 }).report() countSensitivities([running, latest]) ``` ``` running-sum | once=3 twice=6 ab=8 ba=8 -> repeat-sensitive latest-value | once=3 twice=3 ab=5 ba=3 -> order-sensitive robust: 0 repeat-sensitive: 1 order-sensitive: 1 doubly-sensitive: 0 ``` In the running-sum line, `once` is 3 and `twice` is 6: if the same value arrives twice, the total rises twice as much. Order does not matter; 3 then 5 and 5 then 3 both give eight. The operation is **repeat-sensitive**. In the latest-value line the second arrival of the same value changes nothing, but because the last to arrive wins, `ab` is 5 and `ba` is 3. The operation is **order-sensitive**. The same stream can do two different jobs, and which operator it is built with determines the sensitivity. In a real application this difference looks like this. If a publisher delivers the same event twice while reconnecting in the network layer, an accumulating indicator (the unread count) rises twice and stays wrong; a field keeping the latest value writes the same value twice and nothing is noticed on screen. The danger of a repeated event is measured by the type of the operator. If you are using a repeat-sensitive operator such as a counter, you need to carry an identifier that tells events apart and drop the repeat before the operator. **CC21.** The running-sum and latest-value measurement uses an integer state. Overflow of the total and what the first value will be on an empty stream were not measured. ## A Real Stream A real stream comes from a framework, and the measured model does not replace it. The fragment below only shows what the model stands for. ```swift import Combine let subject = PassthroughSubject() let cancellable = subject .removeDuplicates() .sink { value in print(value) } subject.send(3) subject.send(3) subject.send(5) ``` Here `PassthroughSubject` is a publisher, `removeDuplicates` is the operator, and `sink` is the subscription. Of the two consecutive 3s, only one reaches the subscriber. If the subscription is not kept, the stream ends immediately; this is why `cancellable` is held in a variable. ## In a Real Application On a search screen the stream is typically built like this: the text in the box is a publisher; the operator that drops consecutive identical texts prevents a request when the user types the same text again; a debounce operator waits until the user stops typing; finally a subscriber sends the request and writes the result to the screen. Each link corresponds to a measurement in this lesson. Dropping repeats is order-sensitive, because the order in which the texts arrive matters. Subscription is repeat-sensitive, because every subscription can restart a request. Writing to the screen reopens the version question of the previous lesson. The most common setup mistake is rebuilding the subscription every time the screen appears. ```swift // Wrong: a new subscription each time the screen appears, the old one is not released @MainActor final class ListScreen { private var subscriptions: [AnyCancellable] = [] func screenAppeared(_ feed: AnyPublisher<[String], Never>) { feed .sink { items in self.show(items) } .store(in: &subscriptions) } func show(_ items: [String]) {} } ``` Each time the user returns to the screen, `screenAppeared` adds a new subscription and does not release the old ones. On an unshared publisher every subscription sends the request again; after returning three times, the same value is written to the screen three times and three requests go to the network. This is the on-screen counterpart of the measured `cold-same-name` line: subscribing a second time under the same name raised the request count. The fix has two forms. The subscription is built once, where the screen is created; or if it is to be built at each appearance, the old one is released first. Using a shared publisher also makes the request single, but in the case of two screens listening to the same stream with separate lifetimes, what will happen to the stream when the first subscriber leaves was not measured in this lesson. This fragment was not measured: the lifetime of a real subscription, the time of release, and whether the publisher is cold or shared depend on the behaviour of the framework. ## Common Mistakes - Not holding the subscription in a variable. When the subscription is freed, the stream is cancelled. - Subscribing several times to an unshared stream and thinking the request goes out once. - Thinking that dropping consecutive repeats is independent of order. The operator breaks consecutiveness, not order. - Running the subscriber that updates the screen outside the main actor. ## The Bounding Measurement The same operator falls into a different class depending on the pair of values. If the pair is 3 and 3, the two values are the same; if the pair is 3 and 5, they are different. ```swift // main.swift — the same operator, two pairs of values let fresh = Output(emitted: []) let equalPair = replay("pair-3-3", from: fresh, emitting(3), then: emitting(3)).report() let unequalPair = replay("pair-3-5", from: fresh, emitting(3), then: emitting(5)).report() countSensitivities([equalPair, unequalPair]) ``` ``` pair-3-3 | once=[3] twice=[3] ab=[3] ba=[3] -> robust pair-3-5 | once=[3] twice=[3] ab=[3, 5] ba=[5, 3] -> order-sensitive robust: 1 repeat-sensitive: 0 order-sensitive: 1 doubly-sensitive: 0 ``` When the same values arrive consecutively, the operator drops the second and all four states come out equal: the operation is **robust**. With different values the order is written into the output and the operation becomes **order-sensitive**. The rule does not see the cause, it sees the inequality; it classifies not the operator but the pair of values that comes with the operator. Before saying an operator is robust, it must be written down with which values it was tried. The count is one robust, one order-sensitive. ## Stream or Asynchronous Function A stream is not the answer to every problem. For a one-time request, an asynchronous function offers fewer concepts and a more readable error path. A stream earns its place where the value arrives again and again and many places listen to the same values. Distributing the changes of a setting to several screens is an example of this; fetching a profile once is not. The most common mistake in mixed use is wrapping a one-time result in a stream and then waiting for the first value. This brings a subscription lifetime problem and doubles the error path. There is a mistake in the opposite direction too: if values keep arriving and the code abandons them after the first value with a `for` loop, the real gain of the stream is lost. The common conclusion of this lesson's measurements is this: a stream carries time into the code, but the questions of repetition and order do not change. For every operator the same two questions are asked: if the same value arrives twice, does the result change; if two values swap places, does the result change. The answer differs from operator to operator, and asking these two questions before choosing an operator prevents most of the bugs that would appear later. One last note: testing streams requires a scheduler that controls time. Testing time-dependent operators such as debounce with a real clock is slow and brittle. In the test environment a mechanism that advances the scheduler by hand is used; this way a five-second wait becomes an instant step in the test. Advancing time by hand also makes race-like bugs reproducible: because the moment at which the response arrives is written inside the test, the bug appears in the same place on every run. Being able to reproduce a bug is half of fixing it; the other half is showing with the same scenario that the fix really works. The model of this lesson is timeless and deliberately leaves out the time-dependent aspects of the operators. ## Summary - In a reactive stream the publisher produces values, the subscriber receives them, and operators transform or filter. - Every subscription to an unshared stream repeats the request; the operation is repeat-sensitive. - On a shared stream the request goes out only at the first subscriber and the operation is robust. - The operator that drops consecutive repeats is order-sensitive on different values. - The same operator comes out robust on an equal pair of values; the class is determined by the pair of values, not the operator. - This lesson's sensitivity: **repeat-sensitive** (unshared subscription), **order-sensitive** (different values), and **robust** (shared stream). ## Next Step The results of concurrent code may not stay on screen; they are lost when the application closes. In the next topic, the first lesson, Simple Preference Storage, will measure a model in which two screens write the same preference and look at how persistent state reacts to repetition and order.