Quickstart

With axum and PostgreSQL. Any WebSocket server works: feed incoming messages to client.consume and send what comes out of the receiver.

[dependencies]
axum = { version = "0.8", features = ["ws"] }
owservable-core = { git = "https://github.com/owservable/owservable-rs" }
owservable-postgres = { git = "https://github.com/owservable/owservable-rs" }
serde_json = "1"
tokio = { version = "1", features = ["full"] }

use std::sync::Arc;

use axum::Router;
use axum::extract::State;
use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade};
use axum::response::Response;
use axum::routing::get;
use owservable_core::{Client, ConnectionManager, Error, Registry};
use owservable_postgres::Postgres;

struct Session; // authentication hooks; all optional
impl ConnectionManager for Session {}

#[tokio::main]
async fn main() -> Result<(), Error> {
	let db = Postgres::connect("host=localhost user=app dbname=app").await?;
	let mut registry = Registry::default();
	registry.register("tasks", db.watch("tasks").await?);

	let app = Router::new().route("/ws", get(upgrade)).with_state(Arc::new(registry));
	let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await?;
	axum::serve(listener, app).await?;
	Ok(())
}

async fn upgrade(ws: WebSocketUpgrade, State(registry): State<Arc<Registry>>) -> Response {
	ws.on_upgrade(|socket| serve(socket, registry))
}

async fn serve(mut socket: WebSocket, registry: Arc<Registry>) {
	let (mut client, mut outgoing) = Client::new(registry, Arc::new(Session));
	loop {
		tokio::select! {
			Some(message) = outgoing.recv() => {
				if socket.send(Message::Text(message.to_string().into())).await.is_err() {
					break;
				}
			}
			incoming = socket.recv() => match incoming {
				Some(Ok(Message::Text(text))) => {
					if let Ok(message) = serde_json::from_str(&text) {
						let _ = client.consume(message).await;
					}
				}
				Some(Ok(_)) => {}
				_ => break,
			},
		}
	}
	client.disconnected().await;
}

For MongoDB use MongoDb::connect(uri, "app").await? and db.watch("tasks", &[]); for SQLite, Sqlite::open("app.db", Duration::from_millis(250))? and db.watch("tasks")?.

TypeScript v3, with ws, MongoDB and a mongoose model:

import {WebSocketServer} from 'ws';
import {OwservableClient, BackendRegistry} from '@owservable/core';
import {MongoDBConnector, MongoBackend} from '@owservable/mongodb';
import Task from './models/task.js'; // a mongoose model

await MongoDBConnector.init('mongodb://localhost:27017/app?replicaSet=rs0');
BackendRegistry.register('tasks', new MongoBackend(Task));

const session = {user: () => null, connected() {}, ping() {}, location() {}, disconnected() {}, checkSession: async () => null};

new WebSocketServer({port: 3000, path: '/ws'}).on('connection', (socket) => {
	const client = new OwservableClient(session);
	client.subscribe((message) => socket.send(JSON.stringify(message)));
	socket.on('message', (data) => client.consume(JSON.parse(String(data))));
	socket.on('close', () => client.disconnected());
});