pub async fn join_group(
        &mut self,
        topic_name: &str,
        group_id: &str,
        member_id: &str,
    ) -> Result<KfJoinGroupResponse, KfSocketError > {

        let mut request = KfJoinGroupRequest::default();
       
        // metadata
        let mut metadata = Metadata::default();
        metadata.topics = vec![topic_name.to_owned()];

        // protocol metadata
        let mut protocol_metadata = ProtocolMetadata::default();
        protocol_metadata.content = Some(metadata);

        // join group protocol
        let join_group_protocol = JoinGroupRequestProtocol {
            name: "range".to_owned(),
            metadata: protocol_metadata,
        };

        // request with protocol
        request.group_id = group_id.to_owned();
        request.session_timeout_ms = 10000;
        request.rebalance_timeout_ms = 300000;
        request.member_id = member_id.to_owned();
        request.protocol_type = "consumer".to_owned();
        request.protocols = vec![join_group_protocol];

        self.0.send_receive(request).await
    }
// Ask Group Coordinator to join the group
Ask Group Coordinator to join the group
Ask Group Coordinator to join the group
pub async fn join_group(
        &mut self,
        topic_name: &str,
        group_id: &str,
        member_id: &str,
    ) -> Result<KfJoinGroupResponse, KfSocketError > {
        let mut request = KfJoinGroupRequest::default();
        // metadata
        let mut metadata = Metadata::default();
        metadata.topics = vec![topic_name.to_owned()];
        // protocol metadata
        let mut protocol_metadata = ProtocolMetadata::default();
        protocol_metadata.content = Some(metadata);
        // join group protocol
        let join_group_protocol = JoinGroupRequestProtocol {
            name: "range".to_owned(),
            metadata: protocol_metadata,
        };
        // request with protocol
        request.group_id = group_id.to_owned();
        request.session_timeout_ms = 10000;
        request.rebalance_timeout_ms = 300000;
        request.member_id = member_id.to_owned();
        request.protocol_type = "consumer".to_owned();
        request.protocols = vec![join_group_protocol];
        self.0.send_receive(request).await
    }
pub async fn join_group ( & mut self , topic_name : & str , group_id : & str , member_id : & str , ) -> Result < KfJoinGroupResponse , KfSocketError > { let mut request = KfJoinGroupRequest :: default ( ) ; // metadata let mut metadata = Metadata :: default ( ) ; metadata . topics = vec ! [ topic_name to_owned ( ) ] ; // protocol metadata let mut protocol_metadata = ProtocolMetadata :: default ( ) ; protocol_metadata . content = Some ( metadata ) ; // join group protocol let join_group_protocol = JoinGroupRequestProtocol { name : " " . to_owned ( ) , metadata : protocol_metadata , } ; // request with protocol request . group_id = group_id . to_owned ( ) ; request . session_timeout_ms = 10000 ; request . rebalance_timeout_ms = 300000 ; request . member_id = member_id . to_owned ( ) ; request . protocol_type = " " . to_owned ( ) ; request . protocols = vec ! [ join_group_protocol ] ; self . 0 . send_receive ( request ) . await }
Ask Group Coordinator to join the group
Ask Group Coordinator to join the group
&mut self
topic_name
&str
group_id
&str
member_id
&str
&mut self
topic_name
&str
group_id
&str
member_id
&str