Function poll_stream

nodedb/src/control/server/http/routes/stream_poll.rs:58–155

`GET /v1/streams/{stream}/poll`

(
    identity: ResolvedIdentity,
    Path(stream_name): Path<String>,
    Query(params): Query<PollParams>,
    State(state): State<AppState>,
)

/// `GET /v1/streams/{stream}/poll`
pub async fn poll_stream(
    identity: ResolvedIdentity,
    Path(stream_name): Path<String>,
    Query(params): Query<PollParams>,
    State(state): State<AppState>,
) -> impl IntoResponse {
    // Reject any attempt to override the caller's tenant via query string.
    if params.tenant_id.is_some() {
        return (
            StatusCode::FORBIDDEN,
            Json(serde_json::json!({
                "error": "tenant_id must not be supplied as a query parameter; \
                          tenant is determined from the bearer token"
            })),
        )
            .into_response();
    }

    let group = match params.group {
        Some(g) => g.to_lowercase(),
        None => {
            return (
                StatusCode::BAD_REQUEST,
                Json(serde_json::json!({"error": "missing 'group' query parameter"})),
            )
                .into_response();
        }
    };

    let tenant_id = identity.tenant_id().as_u64();
    let limit = params.limit.unwrap_or(100).min(10_000);
    let stream_name = stream_name.to_lowercase();

    let consume_params = ConsumeParams {
        tenant_id,
        stream_name: &stream_name,
        group_name: &group,
        partition: params.partition,
        limit,
    };

    let result = match consume_stream(&state.shared, &consume_params) {
        Ok(r) => r,
        Err(ConsumeError::RemotePartition { leader_node, .. }) => {
            // Forward to remote node.
            match crate::event::cdc::consume::consume_remote(
                &state.shared,
                &consume_params,
                leader_node,
            )
            .await
            {
                Ok(r) => r,
                Err(e) => {
                    return (
                        StatusCode::BAD_GATEWAY,
                        Json(serde_json::json!({"error": e.to_string()})),
                    )