Airstream

The flatMap operator is considered a "loopy" operator because it allows for two behaviors that break the standard synchronous transaction model:

1. Multiple emissions: It can emit more than one event for a single incoming event.
2. Graph cycles: It can create a loop where an observable depends on itself (directly or indirectly).

Because Airstream relies on a static topological rank to prevent glitches, and flatMap makes the graph structure dynamic (the dependency can change based on the emitted value), it cannot be part of the synchronous DAG. Consequently, every event from a flatMap is emitted in a new transaction.

To avoid complexity and potential issues, use combineWith if your logic does not strictly require the dynamic re-shaping of the stream provided by flatMap.

val intStream: EventStream[Int] = ???
val flatIntStream: EventStream[Int] = intStream.flatMap { ev =>
  EventStream.fromSeq(List(ev + 1, ev + 2))
}

Airstream uses a transaction system to prevent FRP glitches. When you call methods like Var#set or EventBus.emit, a new transaction is created. If a transaction is already running (e.g., inside a stream.foreach callback), the new transaction is scheduled to run after the current one finishes.

Important Behavior:

• Var#set: Because the transaction is scheduled for later, you cannot read the updated value in the same transaction using .now(). Multiple .set calls in one transaction may result in only the final state being visible, or multiple updates being collapsed if they occur in the same transaction.
• Var#update: If you need to read the freshest state immediately after a write within the same block of code, use Var#update. It evaluates its modification function only when the transaction actually runs, ensuring it sees the most recent state.
• Transaction Hierarchy: Airstream executes transactions in a depth-first search order. If transaction B is scheduled while transaction A is running, B (and all its descendants) will complete before A's sibling transactions start.
• Synchronous Execution: All transactions happen synchronously. There are no async boundaries within a transaction. Any event following an async delay is automatically part of a new transaction.

// Using .set (will NOT see updates immediately)
stream.foreach { ev =>
  logVar.set(logVar.now() :+ ev)
  println(logVar.now()) // Does NOT contain `ev`
}

// Using .update (WILL see updates correctly)
stream.foreach { ev =>
  logVar.update(_ :+ ev)
  logVar.update(_ :+ ev)
  // After transactions execute, logVar will have two `ev`-s
}

A Transaction encapsulates a part of the propagation that is synchronous and contains no loops. Within a single transaction, Airstream guarantees:

1. No glitches: Inconsistent states are never exposed.
2. No duplicate emissions: No observable will emit more than once.

Transaction Boundaries

New transactions are automatically created in the following scenarios:

• Async streams: Methods like stream.delay(ms) emit in a new transaction because transactions are executed sequentially.
• External events: Events coming from outside Airstream (e.g., browser events) start a new transaction with a topoRank of 1.
• EventBus: Because EventBus can allow manual event firing and might create loops, it emits events in a new transaction.
• Flattening: The eventStream.flatten method produces an event stream that emits in a new transaction.
• Observers: Emitting events inside an Observer (side effects) will necessarily happen in a new transaction.

An EventStream is a reactive variable representing a stream of discrete events.

Key characteristics:

• No current value: Unlike a Signal, an EventStream has no concept of a "current event". It only emits discrete occurrences.
• Lazy: It is a lazy observable. It will not receive or process events unless it has at least one Observer listening to it (either an external observer added via addObserver or foreach, or an internal observer representing a dependent observable).
• Subscription Management: Calling observable.addObserver(observer)(owner) or observable.foreach(onNext)(owner) returns a Subscription. You can manually stop a subscription by calling subscription.kill(), though it is recommended to use the owner pattern for automatic lifecycle management.

// Example of adding an observer with an owner
val subscription = eventStream.foreach(event => println(event))(owner)

// Manual cleanup
subscription.kill()

Airstream uses Source and Sink abstractions to provide flexibility in method signatures, allowing them to accept various types that can be converted into observables or observers.

Source[A]

Anything that exposes a toObservable method. This allows a method to accept "anything that you can get a stream from."

• EventSource: Includes EventStream and EventBus. Calling .toObservable returns an EventStream.
• SignalSource: Includes Signal and Var. Calling .toObservable returns a Signal.

Sink[A]

Anything that exposes a toObserver method. This allows a method to accept anything that can act as a consumer of values.

• Observer is a sink.
• EventBus and Var are sinks.
• js.Function1[A, Unit] has an implicit conversion to Sink[A].

Important Note on Lambdas: There is no implicit conversion from A => Unit to Sink[A] because Scala requires type ascription for the lambda's type parameter. If you need to use a raw function where a Sink is expected, wrap it in Observer() and ascribe the type.

// Example of using Source in a flexible method (as seen in Laminar)
val textBus = new EventBus[String]
div(value <-- textBus.events) // Explicit
div(value <-- textBus)       // Implicitly uses Source

// Handling lambdas that need to be Sinks
div(value <-- (x: String) => println(x)) 
// Or explicitly wrapping:
div(value <-- Observer((x: String) => println(x)))

When performing asynchronous operations, you can track the lifecycle of a request using the Status[In, Out] type. A status can be Pending(input) (the operation is in progress) or Resolved(input, output, ix) (the operation completed successfully).

Key operators:

• flatMapWithStatus: Similar to flatMapSwitch, it wraps the input into a Pending state and the resulting inner stream's emissions into Resolved states.
• foldStatus: Folds a Status observable into a single value by providing handlers for pending and resolved cases.
• delayWithStatus, throttleWithStatus, debounceWithStatus: Async timing operators that also emit Status wrappers.

Other status operators include mapInput, mapOutput, mapPending, and mapResolved.

val requestS: EventStream[Request] = ???
type Response = String

val responseS: EventStream[Status[Request, Response]] =
  requestS.flatMapWithStatus { request =>
    FetchStream.get(request.url, request.options)
  }

// Using foldStatus to drive UI
val textS: EventStream[String] =
  responseS.foldStatus(
    resolved = _.toString,
    pending = _ => "Loading..."
  )

// Example Laminar usage
div(
  child <-- responseS.splitStatus(
    (resolved, _) => div("Response: " + resolved.output.toString),
    (pending, _) => div(img(src("spinner.gif")), "Loading ...")
  )
)

The split operator transforms an observable of a collection (e.g., Signal[List[M]]) into an observable of a collection of pairs (e.g., Signal[List[(Key, Signal[M])]]) based on a key extractor function.

This is highly useful for rendering dynamic lists (like in Laminar) because it allows you to maintain individual signals for each item in a list, enabling efficient updates without re-rendering the entire list.

Parameters:

• key: A function Input => Key used to identify unique items.
• project: A function used to define how the output for each key is constructed. It receives (key, initialValue, currentSignal).
• distinctCompose (optional): A function to customize how items are checked for changes (defaults to _.distinct).

Variants:

• splitByIndex: Use this if your items lack a unique ID and you can rely on their position in the list (works best for append-only lists).
• splitOne: A specialized version for observables of a single item (e.g., Signal[Foo]) rather than a collection.

// Example: Splitting a Signal[List[Foo]] into a Signal[Map[String, Signal[Foo]]]
case class Foo(id: String, version: Int)
val inputSignal: Signal[List[Foo]] = ???

val outputSignal: Signal[List[(String, Signal[Foo])]] = inputSignal.split(
  key = _.id
)(
  project = (key, initialFoo, thisFooSignal) => (key, thisFooSignal)
)

val resultSignal: Signal[Map[String, Signal[Foo]]] = outputSignal.map(list => Map(list: _*))

Starting with Airstream 15, observables follow a "pause/resume" paradigm rather than a "tear down/restart" model. When an observable is stopped, it is effectively paused and attempts to retain its internal state so that it can seamlessly resume when restarted.

Stream State Retention

Even though EventStreams do not represent state like Signals, they can manage internal state that persists through a stop/start cycle:

• stream.take(numEvents): Remembers how many events have already been processed. If numEvents = 2 and the stream is stopped after 2 events, restarting it will not emit more events unless resetOnStop = true is passed.
• stream1.combineWith(stream2): Remembers the last events seen from both streams. Upon restart, it behaves as if it were never stopped (waiting for new emissions) rather than waiting for both streams to emit for the first time.

Signal Behavior

• Signals: Automatically sync their value with their parent signal when restarted.
• Signals depending on Streams: Because streams have no "current value", a signal depending on a stream cannot sync its value upon restart. It will simply start listening again, and its value may remain stale until the parent stream emits a new event. To prevent this, ensure the signal's owner is not unmounted (e.g., use CSS display: none instead of unmounting in Laminar).

A Signal is a reactive variable representing a time-varying value or accumulated state.

Key characteristics:

• Always has a current value: Unlike EventStream, a Signal maintains a state. You can create one from an EventStream using .startWith(initialValue).
• Lazy: Like EventStream, a Signal only processes updates and evaluates its initial value when it has at least one observer.
• Observation Behavior: When you add an observer to a Signal, the observer immediately receives the current value, followed by any future updates.
• Accessing Changes Only: If you want to observe only future updates and not the current value, observe the stream signal.changes instead.

Lazy Initial Value

Initial values in a Signal are evaluated lazily. The expression used to provide the initial value is only evaluated when the Signal acquires its first observer, and it is not re-evaluated after that.

val fooStream: EventStream[Foo] = ???
// signal's current value is initially 'myFoo'
val fooSignal: Signal[Foo] = fooStream.startWith(myFoo)

// barSignal's initial value is computed lazily only when it gets an observer
val barSignal: Signal[Bar] = fooSignal.map(fooToBar)

// To observe only updates (not the current value):
fooSignal.changes.foreach(value => println(s"New value: $value"))(owner)