Function get_or_create_topic_buffer

nodedb/src/event/topic/publish.rs:94–118

Get or create a StreamBuffer for a topic.

(
    state: &SharedState,
    tenant_id: u64,
    topic_name: &str,
    retention: &RetentionConfig,
)

/// Get or create a StreamBuffer for a topic.
fn get_or_create_topic_buffer(
    state: &SharedState,
    tenant_id: u64,
    topic_name: &str,
    retention: &RetentionConfig,
) -> Arc<StreamBuffer> {
    // Topics use the CdcRouter's buffer pool with a "topic:" prefix
    // to avoid name collisions with change streams.
    let buffer_key = format!("topic:{topic_name}");

    if let Some(buf) = state.cdc_router.get_buffer(tenant_id, &buffer_key) {
        return buf;
    }

    // Create a new buffer. Use the router's internal mechanism.
    // Since CdcRouter.get_or_create_buffer is private, we route through
    // a dummy event to force buffer creation, then return it.
    // Instead, let's add a public create method to CdcRouter.
    // For now, use the public get_buffer after forcing creation.
    //
    // Actually, we can just create the buffer directly and register it.
    state
        .cdc_router
        .ensure_buffer(tenant_id, &buffer_key, retention)
}

/// Determine the home node for a topic.
///