gizdb:relay

realtime as a position in a log the database keeps. Every committed change appends its topic and key (not its value: the store is the state, so a subscriber reads the rows a change names); a subscriber asks for a topic's changes after the position it holds and waits (a long poll) until there are some or its wait ends. Positions belong to the database, not to a process, so every worker process of the application (GIZLANG_WORKERS) and every later run answers the same position the same way. A position older than what the log keeps, or from another database, gets a snapshot answer: the caller renders the topic again. So a reconnect, a missed message, a restart and a request served by another worker are one path.

import * as Relay from "gizdb:relay";

Types

Notice

type Notice = { topic: string; key: string; gone: bool; }

A change to the row key of topic: written, or gone.

• topic: string
• key: string
• gone: bool

Delta

type Delta = | { kind: "changes"; changes: Vec<Notice>; position: string; } | { kind: "snapshot"; position: string; }

What a subscriber gets: the topic's changes and the position after them, or a snapshot to render at that position.

• changes
• snapshot

Relay

type Relay

_store is held by reference: a feed travels in an app context (a record of fixed size).

Opaque: create and use it through this module's functions.

Functions

relay

function relay(sys: System, store: Database, keep: i64): Result<Relay, StoreError>

The feed of store, keeping the last keep changes (its table is created when missing).

Effects: fail, read, write, console, io.

Methods

Relay.position

function Relay.position(self: Relay): Result<string, StoreError>

The current position, for the first render of a topic.

Effects: fail, read, write, console, io.

Relay.transaction

function Relay.transaction<F>(self: Relay, work: F): Result<Rows, StoreError>

Runs work (as Database.transaction does) and, once it committed, wakes this process's subscribers. Work writes its rows and publishes each change.

Relay.publish

function Relay.publish(self: Relay, tx: Database, change: Notice): Result<bool, StoreError>

Inside Relay.transaction: append a change. Publishers take the feed's lock until they commit, so positions commit in order and a subscriber that read up to one never later finds an earlier one appear (PostgreSQL hands out identities to transactions that may commit in another order).

Effects: fail, read, write, console, io.

Relay.changed

function Relay.changed(self: Relay, change: Notice, write: (tx: Database) => Result<Rows, StoreError>): Result<Rows, StoreError>

Runs write and publishes change in one Relay.transaction.

Effects: fail, read, write, console, io.

Relay.notify

function Relay.notify(self: Relay, change: Notice): Result<Rows, StoreError>

Publishes change alone, in a Relay.transaction of its own.

Effects: fail, read, write, console, io.

Relay.since

function Relay.since(self: Relay, topic: string, position: string, waitMs: i64): Result<Delta, StoreError>

The changes to topic after position, waiting up to waitMs for one.

Effects: fail, read, write, console, io.

Delta.position

function Delta.position(self: Delta): string

The position a delta brings its subscriber to.