# Streams and cleanup

Source: https://libtmux.org/en/fsharp/latest/guides/streams/

> Consume events with bounded lifetimes and cancellation.

Captured snapshots are replayable local observations. Control-mode events are
live, ordered, destructive observations. They are not a replayable [`seq`](<https://learn.microsoft.com/dotnet/api/system.collections.generic.ienumerable-1>), and
a control client has one event stream with one consumer: a second reader
started while the first is reading raises [`InvalidOperationException`](<https://learn.microsoft.com/dotnet/api/system.invalidoperationexception>).

## Choose a wait or a stream

To wait for one thing a pane prints, use [`Pane.sendAndWait`](<https://libtmux.org/en/fsharp/latest/reference/libtmux-fsharp-pane-sendandwait/>),
[`Pane.waitForText`](<https://libtmux.org/en/fsharp/latest/reference/libtmux-fsharp-pane-waitfortext/>), [`Pane.waitUntil`](<https://libtmux.org/en/fsharp/latest/reference/libtmux-fsharp-pane-waituntil/>) or [`Pane.run`](<https://libtmux.org/en/fsharp/latest/reference/libtmux-fsharp-pane-run/>); each returns a single
result, and [which wait](https://libtmux.org/en/fsharp/latest/guides/getting-started/#which-wait) compares them. The waits share one control client
per session while any wait on it runs, and [`Pane.run`](<https://libtmux.org/en/fsharp/latest/reference/libtmux-fsharp-pane-run/>) learns its exit status
from a private `wait-for` channel. Read a stream when the caller reacts to
events as they arrive.

## Streams are cold

[`Control.events`](<https://libtmux.org/en/fsharp/latest/reference/libtmux-fsharp-control-events/>) and [`Control.watchPane`](<https://libtmux.org/en/fsharp/latest/reference/libtmux-fsharp-control-watchpane/>) return an [`IAsyncEnumerable`](<https://learn.microsoft.com/dotnet/api/system.collections.generic.iasyncenumerable-1>).
Nothing is read until a consumer enumerates it, and the consumer supplies the
cancellation token. [`Control.foldWhile`](<https://libtmux.org/en/fsharp/latest/reference/libtmux-fsharp-control-foldwhile/>) and [`Control.iter`](<https://libtmux.org/en/fsharp/latest/reference/libtmux-fsharp-control-iter/>) consume a stream,
await one callback at a time, dispose only their enumerator, and leave the
client open. Any [`IAsyncEnumerable`](<https://learn.microsoft.com/dotnet/api/system.collections.generic.iasyncenumerable-1>) library composes the same streams;
`TaskSeq<'T>` in FSharp.Control.TaskSeq is the same type.

[`Control.watchPane`](<https://libtmux.org/en/fsharp/latest/reference/libtmux-fsharp-control-watchpane/>) narrows a client's stream to one pane's output, and ends
once the pane is confirmed gone. [`PaneWatch`](<https://libtmux.org/en/fsharp/latest/reference/libtmux-fsharp-panewatch/>) names the six things a watch
yields, so a match that leaves one out, such as a `Dropped` loss report, draws
a compiler warning:

<!-- fsharp-snippet: WatchPaneOutput run -->
```fsharp run
open System
open System.Threading
open LibTmux
open LibTmux.FSharp

let readPaneUntilAsync
    (cancellationToken: CancellationToken)
    (marker: string)
    (pane: Pane)
    (session: IControlModeSession)
    =
    session
    |> Control.watchPane pane
    |> Control.foldWhile
        cancellationToken
        (fun output event ->
            task {
                match event with
                | PaneWatch.Output printed ->
                    let output = output + printed.Data

                    if output.Contains(marker, StringComparison.Ordinal) then
                        return StreamStep.Stop output
                    else
                        return StreamStep.Continue output
                // Output tmux held back or the buffer dropped never arrives;
                // capture the pane to read what the screen shows instead.
                | PaneWatch.Paused _
                | PaneWatch.Continued _
                | PaneWatch.Dropped _ -> return StreamStep.Continue output
                | PaneWatch.Gone _
                | PaneWatch.Exited _ -> return StreamStep.Stop output
            })
        ""
```
<!-- endfsharp-snippet -->

A client has one event stream, so watching two panes with [`Control.watchPane`](<https://libtmux.org/en/fsharp/latest/reference/libtmux-fsharp-control-watchpane/>)
takes two clients. [`Control.watchPanes`](<https://libtmux.org/en/fsharp/latest/reference/libtmux-fsharp-control-watchpanes/>) follows several through one: each
output event names its pane, each pane's end arrives as `PaneWatch.Gone`
after the output buffered before it went, and the stream ends once every pane
is gone. This complete program follows two panes, then their ends:

<!-- fsharp-snippet: WatchPanes run -->
```fsharp run
open System
open System.Threading
open LibTmux
open LibTmux.FSharp

let runAsync () =
    task {
        use deadline = new CancellationTokenSource(TimeSpan.FromSeconds 20.)
        let token = deadline.Token

        let options =
            ServerConnectionOptions(
                SocketName = "fsharp-watch-panes-" + Guid.NewGuid().ToString("N"),
                ConfigurationFile = "/dev/null"
            )

        use! owned = options |> Server.createOwned token

        let! session =
            owned.Value
            |> Server.newSession token (SessionSpec.running "work" "exec sleep 60")

        // The client buffers everything from the moment it attaches, so the
        // panes it should see can start afterwards.
        use! client = session |> Control.enterSession token
        let! first = session |> Session.panes |> Query.list token

        let start command =
            first[0] |> Pane.split token (SplitPaneRequest(Command = command))

        let! build = start "printf 'build: ok\\n'; exec sleep 60"
        let! test = start "printf 'test: ok\\n'; exec sleep 60"

        // One client, both panes: each output event names its pane.
        let! printed =
            client
            |> Control.watchPanes [ build; test ]
            |> Control.foldWhile
                token
                (fun (printed: Map<string, string>) event ->
                    task {
                        match event with
                        | PaneWatch.Output output ->
                            let pane = output.PaneId.ToString()
                            let sofar = printed |> Map.tryFind pane |> Option.defaultValue ""
                            let printed = printed |> Map.add pane (sofar + output.Data)

                            if printed.Count = 2 && printed |> Map.forall (fun _ text -> text.Contains '\n') then
                                return StreamStep.Stop printed
                            else
                                return StreamStep.Continue printed
                        | _ -> return StreamStep.Continue printed
                    })
                Map.empty

        // Each pane's end arrives as PaneWatch.Gone, and the stream ends
        // once both are gone.
        do! build |> Pane.kill token
        do! test |> Pane.kill token

        let! ended =
            client
            |> Control.watchPanes [ build; test ]
            |> Control.foldWhile
                token
                (fun ended event ->
                    task {
                        match event with
                        | PaneWatch.Gone pane -> return StreamStep.Continue(ended @ [ pane ])
                        | _ -> return StreamStep.Continue ended
                    })
                []

        for pane in [ build; test ] do
            printfn "%s" (printed[pane.Id.ToString()].Trim())

        printfn "ended in order given: %b" (ended = [ build.Id; test.Id ])
    }

runAsync().GetAwaiter().GetResult()
```
<!-- endfsharp-snippet -->

It prints:

<!-- fsharp-output: WatchPanes -->
```text
build: ok
test: ok
ended in order given: true
```
<!-- endfsharp-output -->

The watch reads the client's only stream, so it consumes and drops events
for other panes. tmux sends a control client output only from the session it
is attached to, and [`Control.withSession`](<https://libtmux.org/en/fsharp/latest/reference/libtmux-fsharp-control-withsession/>) attaches to the most recently used
one. Attach the client with [`Control.enterSession`](<https://libtmux.org/en/fsharp/latest/reference/libtmux-fsharp-control-entersession/>) to the session that holds
the pane: a watch raises [`ArgumentException`](<https://learn.microsoft.com/dotnet/api/system.argumentexception>) for a pane elsewhere when it
starts, and [`InvalidOperationException`](<https://learn.microsoft.com/dotnet/api/system.invalidoperationexception>) if the pane's window moves to another
session while it is watched.

tmux discards output it has not yet sent once a pane's program exits, so the
last lines of a program that exits at once may never arrive on any control
client. Read final output with [`Pane.run`](<https://libtmux.org/en/fsharp/latest/reference/libtmux-fsharp-pane-run/>), or capture a pane kept with
`remain-on-exit`.

## Follow live state

[`Mirror.start`](<https://libtmux.org/en/fsharp/latest/reference/libtmux-fsharp-mirror-start/>) keeps a current copy of a server's sessions, windows, panes and
clients. Each change tmux announces starts a fresh capture, and a capture that
finds nothing different publishes nothing, so [`Mirror.waitUntil`](<https://libtmux.org/en/fsharp/latest/reference/libtmux-fsharp-mirror-waituntil/>) and
[`Mirror.views`](<https://libtmux.org/en/fsharp/latest/reference/libtmux-fsharp-mirror-views/>) see each distinct state once. Activity times, cursor positions
and history sizes change with every keystroke and do not count. tmux does not announce a pane's
running command or working directory, nor a layout change in a session the
mirror is not attached to; [`Mirror.startRefreshing`](<https://libtmux.org/en/fsharp/latest/reference/libtmux-fsharp-mirror-startrefreshing/>) also captures whenever the
mirror has been quiet for an interval:

<!-- fsharp-snippet: LiveState run -->
```fsharp run
open System
open System.Threading
open LibTmux
open LibTmux.FSharp

let runAsync () =
    task {
        use deadline = new CancellationTokenSource(TimeSpan.FromSeconds 20.)
        let token = deadline.Token

        let options =
            ServerConnectionOptions(
                SocketName = "fsharp-live-" + Guid.NewGuid().ToString("N"),
                ConfigurationFile = "/dev/null"
            )

        use! owned = options |> Server.createOwned token

        let! session =
            owned.Value.CreateSessionAsync(
                NewSessionRequest(Name = "work", WindowName = "shell", Command = "/bin/sh"),
                token
            )

        // Captures again on each change tmux announces, and every 200 ms for
        // changes it does not, such as the command a pane runs.
        use! mirror =
            session |> Mirror.startRefreshing token (TimeSpan.FromMilliseconds 200.)

        let! _ =
            session.CreateWindowAsync(NewWindowRequest(Name = "logs", Command = "/bin/sh"), token)

        let! withLogs =
            mirror
            |> Mirror.waitUntil token (TimeSpan.FromSeconds 5.) (fun view ->
                view.Server.Windows |> Seq.exists (fun window -> window.Name = "logs"))

        let! panes = session |> Session.panes |> Query.list token

        do! panes[0] |> Pane.sendLine token "exec sleep 30"

        // tryWaitUntil answers None when no view matched in time.
        let! sleeping =
            mirror
            |> Mirror.tryWaitUntil token (TimeSpan.FromSeconds 5.) (fun view ->
                view.Server.Panes |> Seq.exists (fun pane -> pane.CurrentCommand = "sleep"))

        printfn "windows: %s" (String.Join(", ", [ for window in withLogs.Server.Windows -> window.Name ]))

        match sleeping with
        | Some sleeping ->
            printfn
                "sleeping panes: %d"
                (sleeping.Server.Panes
                 |> Seq.filter (fun pane -> pane.CurrentCommand = "sleep")
                 |> Seq.length)

            printfn "newer view: %b" (sleeping.Epoch > withLogs.Epoch)
        | None -> printfn "no pane ran sleep within five seconds"
    }

runAsync().GetAwaiter().GetResult()
```
<!-- endfsharp-snippet -->

It prints:

<!-- fsharp-output: LiveState -->
```text
windows: shell, logs
sleeping panes: 1
newer view: true
```
<!-- endfsharp-output -->

The mirror's control client receives notifications only, never pane output,
and does not change window sizes. If its client ends, the mirror attaches
again through the anchor session; once that session is gone, the mirror ends
and [`Mirror.views`](<https://libtmux.org/en/fsharp/latest/reference/libtmux-fsharp-mirror-views/>) raises [`TmuxObjectNotFoundException`](<https://libtmux.org/en/csharp/latest/reference/libtmux-tmuxobjectnotfoundexception/>).

## Ownership

Use [`Control.withSession`](<https://libtmux.org/en/fsharp/latest/reference/libtmux-fsharp-control-withsession/>) to own a client for one task, or pass a client from
[`Control.enter`](<https://libtmux.org/en/fsharp/latest/reference/libtmux-fsharp-control-enter/>) or [`Control.enterSession`](<https://libtmux.org/en/fsharp/latest/reference/libtmux-fsharp-control-entersession/>) to [`Control.useSession`](<https://libtmux.org/en/fsharp/latest/reference/libtmux-fsharp-control-usesession/>). The
[control-mode example](https://libtmux.org/en/fsharp/latest/guides/modes/#control-mode) shows both lifetimes.

## Loss, ends and failures

A client buffers 512 events by default
([`ServerConnectionOptions.ControlModeEventBufferCapacity`](<https://libtmux.org/en/csharp/latest/reference/libtmux-serverconnectionoptions-controlmodeeventbuffercapacity/>)). When a reader falls
behind, the buffer discards the oldest output of the pane with the most output
waiting, so a flooding pane loses its own output rather than a quieter pane's,
and notifications about sessions, windows and layout survive it.
[`TmuxEventsDroppedEvent`](<https://libtmux.org/en/csharp/latest/reference/libtmux-tmuxeventsdroppedevent/>) reports the loss; its [`OnlyOutput`](<https://libtmux.org/en/csharp/latest/reference/libtmux-tmuxeventsdroppedevent-onlyoutput/>) is true when no
notification was discarded, so state derived from notifications is still
exact. The client then pauses the flooding pane in tmux until the reader
catches up: [`TmuxPanePausedEvent`](<https://libtmux.org/en/csharp/latest/reference/libtmux-tmuxpanepausedevent/>) and [`TmuxPaneContinuedEvent`](<https://libtmux.org/en/csharp/latest/reference/libtmux-tmuxpanecontinuedevent/>) bracket output
the pane printed that the reader never receives, and a client nobody reads
keeps the pane paused. The pause is the client's own `refresh-client -A`,
sent when its buffer fills; it does not set tmux's age-based `pause-after`,
so a reader that keeps up never sees a pause. Capture the pane to read its
screen after a gap. [`TmuxExitEvent`](<https://libtmux.org/en/csharp/latest/reference/libtmux-tmuxexitevent/>) precedes normal stream completion. A stream fault arrives after buffered events. Cancellation stops
waiting and disposes the reader; it does not undo a command tmux already
received.

When a callback and the enumerator's cleanup both fail, the callback's
exception propagates unchanged and [`Control.cleanupFailure`](<https://libtmux.org/en/fsharp/latest/reference/libtmux-fsharp-control-cleanupfailure/>) returns the
cleanup's. The callback's exception keeps its type, so a handler that matches
`:? TmuxPaneException` still catches it; an `AggregateException` of both would
not be caught there.