与其他客户端互通

本指南说明 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`
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 信封。 live 测试像另一个程序那样，用第二个 fred 客户端从服务器读取这两者：orders 最新的条目在 _payload 下保存订单，emails 最新的值就是信封。

读取流条目​

写入列表值​

读取列表值或频道消息​

测试​

读取流条目

写入列表值

读取列表值或频道消息

测试