/// Validate topic, takes advantage of the validation routines inside topic action workflow
Validate topic, takes advantage of the validation routines inside topic action workflow
Validate topic takes advantage of the validation routines inside topic action workflow
fn validate_topic_request(
    name: &str,
    topic_spec: &TopicSpec,
    metadata: &LocalStores,
) -> Result<(), FlvResponseMessage> {
    debug!("validating topic: {}", name);
    // check if topic already exists
    if metadata.topics().contains_key(name) {
        return Err(FlvResponseMessage::new(
            name.to_string(),
            FlvErrorCode::TopicAlreadyExists,
            Some(format!("topic '{}' already defined", name)),
        ));
    }
    // create temporary topic status to return validation result
    let topic_kv = TopicKV::with_spec(name.to_owned(), topic_spec.clone());
    match topic_spec {
        TopicSpec::Computed(param) => {
            let next_state = topic_kv.validate_computed_topic_parameters(param);
            trace!("validating, computed topic: {:#?}", next_state);
            if next_state.resolution.is_invalid() {
                Err(FlvResponseMessage::new(
                    name.to_string(),
                    FlvErrorCode::TopicError,
                    Some(next_state.reason),
                ))
            } else {
                let next_state = topic_kv.generate_replica_map(metadata.spus(), param);
                trace!("validating, generate replica map topic: {:#?}", next_state);
                if next_state.resolution.no_resource() {
                    Err(FlvResponseMessage::new(
                        name.to_string(),
                        FlvErrorCode::TopicError,
                        Some(next_state.reason),
                    ))
                } else {
                    Ok(())
                }
            }
        }
        TopicSpec::Assigned(ref partition_map) => {
            let next_state = topic_kv.validate_assigned_topic_parameters(partition_map);
            trace!("validating, computed topic: {:#?}", next_state);
            if next_state.resolution.is_invalid() {
                Err(FlvResponseMessage::new(
                    name.to_string(),
                    FlvErrorCode::TopicError,
                    Some(next_state.reason),
                ))
            } else {
                let next_state =
                    topic_kv.update_replica_map_for_assigned_topic(partition_map, metadata.spus());
                trace!("validating, assign replica map topic: {:#?}", next_state);
                if next_state.resolution.is_invalid() {
                    Err(FlvResponseMessage::new(
                        name.to_string(),
                        FlvErrorCode::TopicError,
                        Some(next_state.reason),
                    ))
                } else {
                    Ok(())
                }
            }
        }
    }
}
fn validate_topic_request ( name : & str , topic_spec : & TopicSpec , metadata : & LocalStores , ) -> Result < ( ) , FlvResponseMessage > { debug ! ( " " name ) ; // check if topic already exists if metadata . topics ( ) . contains_key ( name ) { return Err ( FlvResponseMessage :: new ( name . to_string ( ) , FlvErrorCode :: TopicAlreadyExists , Some ( format ! ( " " name ) ) , ) ) ; } // create temporary topic status to return validation result let topic_kv = TopicKV :: with_spec ( name . to_owned ( ) , topic_spec . clone ( ) ) ; match topic_spec { TopicSpec :: Computed ( param ) => { let next_state = topic_kv . validate_computed_topic_parameters ( param ) ; trace ! ( " " next_state ) ; if next_state . resolution . is_invalid ( ) { Err ( FlvResponseMessage :: new ( name . to_string ( ) , FlvErrorCode :: TopicError , Some ( next_state . reason ) , ) ) } else { let next_state = topic_kv . generate_replica_map ( metadata . spus ( ) , param ) ; trace ! ( " " next_state ) ; if next_state . resolution . no_resource ( ) { Err ( FlvResponseMessage :: new ( name . to_string ( ) , FlvErrorCode :: TopicError , Some ( next_state . reason ) , ) ) } else { Ok ( ( ) ) } } } TopicSpec :: Assigned ( ref partition_map ) => { let next_state = topic_kv . validate_assigned_topic_parameters ( partition_map ) ; trace ! ( " " next_state ) ; if next_state . resolution . is_invalid ( ) { Err ( FlvResponseMessage :: new ( name . to_string ( ) , FlvErrorCode :: TopicError , Some ( next_state . reason ) , ) ) } else { let next_state = topic_kv . update_replica_map_for_assigned_topic ( partition_map , metadata . spus ( ) ) ; trace ! ( " " next_state ) ; if next_state . resolution . is_invalid ( ) { Err ( FlvResponseMessage :: new ( name . to_string ( ) , FlvErrorCode :: TopicError , Some ( next_state . reason ) , ) ) } else { Ok ( ( ) ) } } } } }
Validate topic, takes advantage of the validation routines inside topic action workflow
Validate topic takes advantage of the validation routines inside topic action workflow
name
&str
topic_spec
&TopicSpec
metadata
&LocalStores