One system for your events, state and work.

use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;

use anyhow::{Context, Result};
use felix_client::{AckMode, ClientConfig, ClusterClient, quic_client_config};

#[tokio::main]
async fn main() -> Result<()> {
    let mut config = ClientConfig::from_env_or_yaml(quic_client_config(None, false)?, None)?;
    config.auth_tenant_id = Some("dev".into());
    config.auth_token = Some(std::env::var("FELIX_TOKEN").context("set FELIX_TOKEN")?);

    let seeds: Vec<SocketAddr> = vec!["127.0.0.1:5000".parse()?];
    let client = Arc::new(ClusterClient::connect(&seeds, "localhost", config).await?);

    // Subscribe before publishing: a subscription starts at the stream's tail.
    let mut subscription = client.subscribe("dev", "default", "orders").await?;
    client
        .publish("dev", "default", "orders", b"hello".to_vec(), AckMode::PerMessage)
        .await?;

    let event = tokio::time::timeout(Duration::from_secs(5), subscription.next_event())
        .await
        .context("no event within 5 seconds")??
        .context("the broker ended the subscription")?;
    println!("received {:?}", event.payload);
    Ok(())
}

Docs for the Rust client

import os

import felix

with felix.Client(
    "127.0.0.1:5000", tenant_id="dev", token=os.environ["FELIX_TOKEN"]
) as client:
    # Subscribe first, so the publish below is not missed.
    with client.subscribe("dev", "default", "orders") as events:
        client.publish("dev", "default", "orders", b"hello")
        event = events.next_event(timeout=5)
        if event is None:
            raise SystemExit("no event within 5 seconds")
        print("received", event.payload)

Docs for the Python client

import { Client } from "felix-client";

const token = process.env.FELIX_TOKEN;
if (!token) throw new Error("set FELIX_TOKEN");

const client = await Client.connect("127.0.0.1:5000", "dev", { token });
try {
  // Subscribe first, so the publish below is not missed.
  const events = await client.subscribe("dev", "default", "orders");
  try {
    await client.publish("dev", "default", "orders", Buffer.from("hello"));
    const event = await events.nextEvent();
    if (event === null) throw new Error("the subscription ended");
    console.log("received", event.payload.toString());
  } finally {
    await events.close();
  }
} finally {
  client.close();
}

Docs for the TypeScript client

using Felix;

var token = Environment.GetEnvironmentVariable("FELIX_TOKEN")
    ?? throw new InvalidOperationException("set FELIX_TOKEN");

await using var client = await FelixClient.ConnectAsync(new FelixClientOptions
{
    Addrs = ["127.0.0.1:5000"],
    TenantId = "dev",
    Token = token,
});

// Subscribe first, so the publish below is not missed.
await using var sub = await client.SubscribeAsync(
    "dev", "default", "orders", shard: 0, StartPosition.Latest);

await client.PublishAsync("dev", "default", "orders", "hello"u8.ToArray());

var ev = await sub.ReadAsync(TimeSpan.FromSeconds(5))
    ?? throw new TimeoutException("no event within 5 seconds");
Console.WriteLine($"{ev.Offset}: {ev.Payload.Length} bytes");

Docs for the C# client

#include <pthread.h>
#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include "felix.h"

struct waiter {
    pthread_mutex_t mu;
    pthread_cond_t cv;
    bool done;
    FelixStatus status;
    void *handle;      /* the new client, subscription or event */
    char message[256]; /* copied: the callback's message dies with the callback */
};

static void finish(struct waiter *w, FelixStatus status, const char *message, void *handle) {
    pthread_mutex_lock(&w->mu);
    w->done = true;
    w->status = status;
    w->handle = handle;
    snprintf(w->message, sizeof w->message, "%s", message ? message : "");
    pthread_cond_signal(&w->cv);
    pthread_mutex_unlock(&w->mu);
}

/* Waits for the callback of a call that returned `started`. A refused call
   never runs its callback, so its error is taken from the thread instead. */
static FelixStatus wait_for(struct waiter *w, FelixStatus started) {
    if (started != FELIX_STATUS_OK) {
        snprintf(w->message, sizeof w->message, "%s", felix_last_error_message());
        return started;
    }
    pthread_mutex_lock(&w->mu);
    while (!w->done) pthread_cond_wait(&w->cv, &w->mu);
    w->done = false;
    FelixStatus status = w->status;
    pthread_mutex_unlock(&w->mu);
    return status;
}

static void on_connect(void *user_data, FelixStatus status, const char *message,
                       FelixClient *client) {
    finish(user_data, status, message, client);
}
static void on_subscribe(void *user_data, FelixStatus status, const char *message,
                         FelixSubscription *subscription) {
    finish(user_data, status, message, subscription);
}
static void on_publish(void *user_data, FelixStatus status, const char *message,
                       bool has_offset, uint64_t offset) {
    (void)has_offset;
    (void)offset;
    finish(user_data, status, message, NULL);
}
static void on_event(void *user_data, FelixStatus status, const char *message,
                     FelixEvent *event) {
    finish(user_data, status, message, event);
}

int main(void) {
    const char *token = getenv("FELIX_TOKEN");
    if (!token || !*token) {
        fprintf(stderr, "set FELIX_TOKEN\n");
        return 1;
    }

    struct waiter w = {.done = false};
    pthread_mutex_init(&w.mu, NULL);
    pthread_cond_init(&w.cv, NULL);
    FelixClient *client = NULL;
    FelixSubscription *sub = NULL;
    FelixEvent *event = NULL;
    const uint8_t *data = NULL;
    size_t len = 0;
    int exit_code = 1;
    FelixStatus status;

    status = felix_client_connect("127.0.0.1:5000", "dev", token, "localhost",
                                  NULL /* system CAs */, NULL, on_connect, &w);
    if (wait_for(&w, status) != FELIX_STATUS_OK) {
        fprintf(stderr, "connect: %s\n", w.message);
        goto cleanup;
    }
    client = w.handle;

    /* Subscribe first, so the publish below is not missed. */
    status = felix_client_subscribe(client, "dev", "default", "orders", 0,
                                    FELIX_START_LATEST, 0, NULL, on_subscribe, &w);
    if (wait_for(&w, status) != FELIX_STATUS_OK) {
        fprintf(stderr, "subscribe: %s\n", w.message);
        goto cleanup;
    }
    sub = w.handle;

    static const uint8_t payload[] = "hello";
    status = felix_client_publish(client, "dev", "default", "orders", payload,
                                  sizeof payload - 1, NULL, 0, FELIX_ACK_PER_MESSAGE,
                                  FELIX_DELIVERY_AT_MOST_ONCE, NULL, on_publish, &w);
    if (wait_for(&w, status) != FELIX_STATUS_OK) {
        fprintf(stderr, "publish: %s\n", w.message);
        goto cleanup;
    }

    status = felix_subscription_next(sub, 5000 /* ms */, NULL, on_event, &w);
    status = wait_for(&w, status);
    if (status != FELIX_STATUS_OK) {
        fprintf(stderr, "read: %s\n", status == FELIX_STATUS_TIMEOUT ? "timed out" : w.message);
        goto cleanup;
    }
    event = w.handle;
    if (felix_event_payload(event, &data, &len) != FELIX_STATUS_OK) {
        fprintf(stderr, "payload: %s\n", felix_last_error_message());
        goto cleanup;
    }
    printf("%.*s\n", (int)len, (const char *)data);
    exit_code = 0;

cleanup:
    /* Every call has completed by here, so no callback still holds &w. */
    felix_event_free(event);
    felix_subscription_free(sub);
    felix_client_free(client);
    pthread_cond_destroy(&w.cv);
    pthread_mutex_destroy(&w.mu);
    return exit_code;
}

Docs for the C client

# one terminal
felixctl sub orders

# another
felixctl pub orders 'hello'
felixctl cache put users alice 'online'
felixctl cache get users alice
felixctl topology orders

Docs for felixctl