Подключение к промышленному кластеру

Промышленный кластер требует 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))
}

Адрес подключения и токен развёртывание передаёт в переменных окружения. Клиент Pulsar подключается к одному адресу. К кластеру из нескольких брокеров ведёт одна точка входа перед ними: DNS-имя, балансировщик нагрузки или прокси Pulsar.

Адрес pulsar+ssl:// ведёт к TLS-порту кластера, по соглашению это порт 6651. Клиент проверяет сертификат сервера по доверенным корневым сертификатам системы, а также его имя хоста. token выполняет аутентификацию по JWT, который клиент отправляет при подключении и при переподключении. Эту пару принимают и StreamNative Cloud, и собственный кластер с аутентификацией по токену.

Повторы и размер сообщения

Клиент сам переподключается после перезапуска брокера и повторяет любую операцию, которую кластер ещё не принял. По умолчанию он повторяет без ограничения, раз в пять секунд, и так пережидает перезапуск любой длительности. OperationRetries это ограничивает: десять попыток с интервалом три секунды, после чего операция возвращает ошибку с ответом кластера. Ограничение действует на поиск топиков, продюсеры и переподключение каждого консьюмера. Оно же превращает второго консьюмера подписки Exclusive в ошибку при запуске вместо ожидания.

max_message_size - наибольшая полезная нагрузка публикации, по умолчанию пять мебибайт, как на сервере. Развёртывание, в котором изменён 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)
}

Каждая настройка брокера - шаг билдера, который никуда не подключается, поэтому #[ruststream::app] строит приложение так же, как локально. Соединение открывается при запуске сервиса. Генерируемый документ AsyncAPI указывает хост и порт из адреса, но никогда не токен. Сервису, который перечитывает свои топики, нужна ещё политика хранения пространства имён на кластере.

Проверка конфигурации

Тесты запускают рабочий service с рабочим broker, которому передан тестовый токен, через TestApp::start. Pulsar внутри процесса проверяет адрес подключения так, как его разбирает клиент, но к кластеру не подключается. С адресом другой схемы запуск возвращает ошибку, и она называет схемы, по которым подключается клиент:

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

TLS, токен и ограничение повторов действуют только на настоящем кластере.