连接生产集群

生产集群要求 TLS 和认证，服务则要决定等待集群多久、发送多大的消息。

TLS 和令牌

src/main.rs
/// Where the cluster is and who the service is, as the deployment provides them.
struct Cluster {
    url: String,
    token: String,
}

impl Cluster {
    fn from_env() -> Self {
        let read = |name: &str| env::var(name).unwrap_or_default();
        Self {
            url: read("PULSAR_URL"),
            token: read("PULSAR_TOKEN"),
        }
    }
}

fn broker(cluster: Cluster) -> PulsarBroker {
    // Ten tries, three seconds apart; then the error.
    let retries = OperationRetries::attempts(nonzero!(10u32)).delay(Duration::from_secs(3));
    PulsarBroker::new(cluster.url)
        .token(cluster.token)
        .operation_retries(retries)
        // The cluster runs with `maxMessageSize=10485760`.
        .max_message_size(nonzero!(10_485_760usize))
}

部署环境通过环境变量提供服务 URL 和令牌。 Pulsar 客户端只连接一个服务 URL。由多个 Broker 组成的集群，通过它们前面的一个入口点访问：DNS 名称、负载均衡器，或者 Pulsar 代理。

pulsar+ssl:// URL 连接集群的 TLS 监听端口，按惯例是 6651。客户端用系统信任的根证书校验服务器的证书，并校验它的主机名。 token 用 JWT 认证，客户端在连接和重新连接时发送它。 StreamNative Cloud 和启用了令牌认证的自建集群都使用这一组设置。

重试与消息大小

Broker 重启之后，客户端会自己重新连接，并为集群尚未接受的操作重新请求。默认情况下，它不限次数地请求，间隔五秒，因此不论重启持续多久都能等过去。 OperationRetries 给这个过程设定上限：尝试十次，间隔三秒，之后操作以集群的应答失败。这个上限覆盖查找、生产者以及每个消费者的重新连接。它还把 Exclusive 订阅的第二个消费者从启动时的等待变成启动时的错误。

max_message_size 是一次发布能携带的最大载荷，默认 5 MiB，与服务器相同。修改过服务器 maxMessageSize 的部署，在这里设置相同的值。更大的发布在服务内部就会失败，不会到达集群。

服务

宏
手写

src/main.rs
#[subscriber(PulsarSubscription::new("orders", "orders-svc"), reply)]
async fn confirm(order: &Order) -> Confirmation {
    tracing::info!(order.id, "order confirmed");
    Confirmation { id: order.id }
}

fn service(broker: PulsarBroker) -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(confirm);
    })
}

#[ruststream::app]
fn app() -> impl App {
    service(broker(Cluster::from_env()))
}

src/main.rs
struct Confirm;

impl Handle<Order, Confirmation> for Confirm {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Confirmation, HandlerOutcome>> {
        tracing::info!(order.id, "order confirmed");
        ready(Ok(Confirmation { id: order.id }))
    }
}

fn service(broker: PulsarBroker) -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(
            subscriber(PulsarSubscription::new("orders", "orders-svc"), Confirm)
                .reply()
                .build(),
        );
    })
}

fn app() -> RustStream {
    service(broker(Cluster::from_env()))
}

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

Broker 的每项设置都是一个不发起连接的构建器步骤，所以 #[ruststream::app] 像在本地一样构建应用。连接在服务启动时才打开。生成的 AsyncAPI 文档报告 URL 中的主机和端口，从不包含令牌。重放主题的服务还需要集群上命名空间的保留策略。

测试配置

测试在 TestApp::start 下启动生产环境的 service，配上生产环境的 broker 和测试令牌。进程内 Pulsar 像客户端那样解析校验服务 URL，但不会去连接集群。使用其他协议的 URL 会让启动失败，错误信息会列出客户端支持连接的协议：

the service URL's scheme "https" is not one the client dials: pulsar or pulsar+ssl

TLS、令牌和重试上限只在真实集群上生效。

TLS 和令牌​

重试与消息大小​

服务​

测试配置​

TLS 和令牌

重试与消息大小

服务

测试配置