Issho.Streams (Issho v0.4.0)

Copy Markdown View Source

Streams of a room: output that a process writes a piece at a time (a model's answer, an agent's report, an import's log), that people see as it comes, and find again when they come back: from another page, another tab, after their connection dropped, on another node.

{:ok, stream} =
  Issho.Streams.start(MyApp.Issho, "notebook:7",
    kind: "summary",
    for: ada,
    place: %{"doc" => "notebook:7/intro", "at" => %{"node" => block_id}},
    by: claude
  )

:ok = Issho.Streams.append(stream, "## Summary\n\n")
:ok = Issho.Streams.append(stream, "The plan holds…")
:ok = Issho.Streams.finish(stream)

Issho keeps a stream's chunks and sends them on; it does not read them. What a stream is, each component or app decides: a suggestion in a text, a card waiting on a canvas, a chat's answer, a log.

Who sees it

:for is required: a person (an actor, or a key), who sees it, and so do their agents (on_behalf_of); :room, everyone the policy lets read the room; or :policy, whomever the policy allows the action :read_stream in the room. Its writer (:by) sees it too. Every reader may read the room (:read).

Seen as it comes, found again

subscribe/4 gives the stream as it is now, its content so far, and from then on sends the subscriber what comes, in order, with nothing missed or sent twice:

  • {Issho, :stream, id, {:chunk, data}}
  • {Issho, :stream, id, :finished}
  • {Issho, :stream, id, {:failed, reason}}
  • {Issho, :stream, id, {:closed, outcome}}
  • {Issho, :stream, id, :gone}: kept no longer.

watch/3 subscribes to every stream of a room the caller may see, and to the news of new ones: {Issho, :stream_started, room, stream}, on which the watcher calls subscribe/4 (it is refused a stream it may not see). A LiveView does it with Issho.Live.streams/1, which does both.

Its life

A stream lives while it is written. Finished (finish/1) or failed (fail/2), it is kept for :keep milliseconds (the instance's :stream_keep, five minutes), for those who come back, then goes. close/2 ends it at once, with an outcome the app names ("accepted", "rejected", "stopped", its own): a suggestion taken or turned down.

Its writer is the process that starts it (or :writer): when it exits while the stream is written, the stream fails ("writer_exited"), or finishes with on_exit: :finish. No stream is left being written.

Each start, finish, failure and close is an event of the room (Issho.Event): "stream.started", "stream.finished", "stream.failed" and "stream.closed", with data %{stream: id, kind: kind, for: for} and the reason or outcome; so an agent learns whether its draft was taken. Their content is not in them.

Limits

A stream holds at most the instance's :stream_max_bytes (1 MB): one that would hold more fails ("too_large"). A room has at most :room_max_streams (32) at once. Streams are in memory, as presence is: a stream does not outlive its node.

Each stream is a process (:gen_statem), named in the whole cluster (:global), so it is found from any node; the streams of a room are a process group (:pg).

Summary

Types

ref()

Functions

append(stream, data)

Adds a chunk (a binary) to the stream. {:error, :not_writing} once it is done; {:error, :too_large} when the stream would hold more than its limit, and it fails.

close(stream, outcome)

Closes the stream now, with an outcome the app names (:accepted, :rejected, :stopped, its own): it goes, and its subscribers hear how. From a person's page, Issho.Live.close_stream/3 checks they may see it first.

fail(stream, reason)

Fails the stream, with a reason (a string or an atom).

finish(stream)

Finishes the stream: it is complete, and kept for those who come back.

get(instance, id)

The stream as it is now: {:ok, stream} or {:error, :not_found}.

list(instance, room, who)

The streams of a room that who (an actor or an identity) may see, as they are now, oldest first.

start(instance, room, opts)

Starts a stream in a room, written by the caller (or :writer). Options: :for (required), :kind, :place, :meta, :by (an actor or an identity: who writes it, as the room sees them), :writer, :keep (milliseconds), :on_exit (:fail or :finish).

subscribe(instance, stream_or_id, who, pid \\ self())

Subscribes pid (the caller) to a stream that who may see: gives the stream as it is now, then sends what comes (see above). {:error, :forbidden} for one they may not see.

unsubscribe(stream, pid \\ self())

Stops sending a stream to pid (the caller).

unwatch(instance, room)

Stops the news of a room's new streams (watch/3).

visible?(stream, identity)

Whether who (an identity) may see the stream: they may read its room, and it is for them (or for whom they act), for the room, for whom the policy allows :read_stream, or they write it.

watch(instance, room, who)

Subscribes the caller to every stream of the room that who may see, and to the news of new ones ({Issho, :stream_started, room, stream}). Gives the streams, as they are now.

Types

ref()

@type ref() :: Issho.Stream.t() | {atom(), String.t()}

Functions

append(stream, data)

@spec append(ref(), iodata()) :: :ok | {:error, term()}

Adds a chunk (a binary) to the stream. {:error, :not_writing} once it is done; {:error, :too_large} when the stream would hold more than its limit, and it fails.

close(stream, outcome)

@spec close(ref(), String.t() | atom()) :: :ok | {:error, term()}

Closes the stream now, with an outcome the app names (:accepted, :rejected, :stopped, its own): it goes, and its subscribers hear how. From a person's page, Issho.Live.close_stream/3 checks they may see it first.

fail(stream, reason)

@spec fail(ref(), String.t() | atom()) :: :ok | {:error, term()}

Fails the stream, with a reason (a string or an atom).

finish(stream)

@spec finish(ref()) :: :ok | {:error, term()}

Finishes the stream: it is complete, and kept for those who come back.

get(instance, id)

@spec get(atom(), String.t()) :: {:ok, Issho.Stream.t()} | {:error, :not_found}

The stream as it is now: {:ok, stream} or {:error, :not_found}.

list(instance, room, who)

@spec list(atom(), String.t(), Issho.Actor.t() | Issho.Identity.t()) :: [
  Issho.Stream.t()
]

The streams of a room that who (an actor or an identity) may see, as they are now, oldest first.

start(instance, room, opts)

@spec start(atom(), String.t(), keyword()) ::
  {:ok, Issho.Stream.t()} | {:error, term()}

Starts a stream in a room, written by the caller (or :writer). Options: :for (required), :kind, :place, :meta, :by (an actor or an identity: who writes it, as the room sees them), :writer, :keep (milliseconds), :on_exit (:fail or :finish).

subscribe(instance, stream_or_id, who, pid \\ self())

@spec subscribe(
  atom(),
  String.t() | Issho.Stream.t(),
  Issho.Actor.t() | Issho.Identity.t(),
  pid()
) ::
  {:ok, Issho.Stream.t()} | {:error, :forbidden | :not_found}

Subscribes pid (the caller) to a stream that who may see: gives the stream as it is now, then sends what comes (see above). {:error, :forbidden} for one they may not see.

unsubscribe(stream, pid \\ self())

@spec unsubscribe(ref(), pid()) :: :ok

Stops sending a stream to pid (the caller).

unwatch(instance, room)

@spec unwatch(atom(), String.t()) :: :ok

Stops the news of a room's new streams (watch/3).

visible?(stream, identity)

@spec visible?(Issho.Stream.t(), Issho.Identity.t()) :: boolean()

Whether who (an identity) may see the stream: they may read its room, and it is for them (or for whom they act), for the room, for whom the policy allows :read_stream, or they write it.

watch(instance, room, who)

@spec watch(atom(), String.t(), Issho.Actor.t() | Issho.Identity.t()) ::
  {:ok, [Issho.Stream.t()]}

Subscribes the caller to every stream of the room that who may see, and to the news of new ones ({Issho, :stream_started, room, stream}). Gives the streams, as they are now.