Взаимодействие с другими клиентами

Это руководство показывает, что сервис на RustStream читает и пишет в Redis, чтобы производители и потребители на других языках работали с его потоками, списками и каналами.

Пример - сервис заказов. Магазин, написанный на другом языке, добавляет заказы в поток, а почтовый сервис на другом языке извлекает письма из списка.

Чтение записи потока

Макросы

src/main.rs

/// An order as the shop's producer appends it: `XADD orders * _payload '{"id":1,...}'`.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "orders")]
struct Order {
    id: u64,
    email: String,
}

/// An email for the mailer, a worker written in another language.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "emails")]
struct Email {
    to: String,
    subject: String,
}

#[subscriber(RedisStream::new("orders").group("mail"), reply)]
async fn take_order(order: &Order) -> Email {
    tracing::info!(order.id, "order taken");
    Email {
        to: order.email.clone(),
        subject: format!("Order {} received", order.id),
    }
}

Вручную

src/main.rs

/// An order as the shop's producer appends it: `XADD orders * _payload '{"id":1,...}'`.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
#[outgoing(name = "orders")]
struct Order {
    id: u64,
    email: String,
}

/// An email for the mailer, a worker written in another language.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
#[outgoing(name = "emails")]
struct Email {
    to: String,
    subject: String,
}

struct TakeOrder;

impl Handle<Order, Email> for TakeOrder {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Email, HandlerOutcome>> {
        tracing::info!(order.id, "order taken");
        ready(Ok(Email {
            to: order.email.clone(),
            subject: format!("Order {} received", order.id),
        }))
    }
}

Запись потока Redis - это набор полей. Сервис на RustStream хранит тело сообщения в поле _payload, а каждый заголовок в поле с именем h: и именем заголовка после него. Производитель на любом языке записывает заказ, который прочитает сервис, так:

redis-cli XADD orders '*' _payload '{"id":1,"email":"ann@example.com"}' h:trace-id 4bf92f35

Тело - это то, что декодирует кодек. С кодеком по умолчанию это JSON, поэтому производитель записывает заказ в JSON. Поле заголовка необязательно. Запись с одним _payload приходит без заголовков. Поля с другими именами пропускаются, поэтому производитель может добавлять свои. У записи без _payload пустое тело, и кодек не может его декодировать. Такую запись завершает политика ошибок декодирования (FailurePolicy), и по умолчанию она отбрасывает запись.

Потребитель на другом языке читает записи, которые пишет сервис, так же: _payload содержит тело, а поля h: содержат заголовки.

Запись значения в список

Макросы

src/main.rs

#[ruststream::app]
fn app() -> impl App<State = ()> {
    service(RedisBroker::standalone("redis://localhost:6379"))
}

fn service(broker: RedisBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("shop", "0.1.0")).with_broker(broker, |b| {
        // The envelope in JSON: a value the mailer reads with any JSON parser.
        b.include(take_order)
            .out_reply(list::Publish::default().codec(JsonCodec));
    })
}

Вручную

src/main.rs

fn app() -> impl App<State = ()> {
    service(RedisBroker::standalone("redis://localhost:6379"))
}

fn service(broker: RedisBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("shop", "0.1.0")).with_broker(broker, |b| {
        let orders = RedisStream::new("orders").group("mail");
        // The envelope in JSON: a value the mailer reads with any JSON parser.
        b.include(subscriber(orders, TakeOrder).reply().build())
            .out_reply(list::Publish::default().codec(JsonCodec));
    })
}

fn main() -> ExitCode {
    run_main(app)
}

Запись списка и сообщение Pub/Sub - это одно значение, поэтому заголовки передаются в кадре вокруг тела. По умолчанию кадр двоичный:

Байты	Значение
4	число заголовков, u32 в порядке big-endian
4, затем имя	для каждого заголовка: длина имени и имя
4, затем значение	длина значения и значение
остаток	тело

Кодек в политике публикации вместо этого пишет конверт в формате этого кодека. С JsonCodec почтовый сервис извлекает объект JSON:

redis-cli RPOP emails
# {"headers":{},"payload":"{\"to\":\"ann@example.com\",\"subject\":\"Order 1 received\"}"}

payload - это тело, здесь письмо в виде текста JSON, поэтому почтовый сервис разбирает конверт, а затем его payload. Тело или заголовок в корректном UTF-8 записывается строкой, а любые другие байты - массивом чисел. В таком конверте значения читаются и в инструменте вроде RedisInsight.

Чтение значения списка или сообщения канала

Подписка читает кадр, который пишет её издатель: RedisList::codec и RedisPubSub::codec задают кодек конверта, а без него подписка читает двоичный кадр. Подписка с кодеком читает и двоичные кадры, которые остаются после перехода сервиса на кодек.

Любое другое значение, например простой JSON, который добавляет или публикует другой клиент, приходит как тело без заголовков:

redis-cli LPUSH jobs '{"id":7}'
redis-cli PUBLISH prices '{"sku":"tea","cents":450}'

Поэтому производитель на другом языке пишет простые значения, а если ему нужны заголовки, пишет конверт JSON.

Тестирование

Тест в файлах примера работает внутри процесса: заказ приходит в том виде, в каком его пишет магазин, а значение в emails побайтно совпадает с конвертом JSON. Тест с сервером читает и то и другое вторым клиентом fred, как это сделала бы другая программа: самая новая запись orders хранит заказ в _payload, а самое новое значение emails - это конверт.