async fn group_and_fetch_log_futures(
    conn: Connection,
    vers: KfApiVersions,
    grp_id: String,
    mbr_id: String,
    gen_id: i32,
    max_bytes: i32,
    from_beginning: bool,
    tp_param: TopicPartitionParam,
    response_params: ResponseLogParams,
) -> Result<(), CliError> {
    let mut send_channels = vec![];
    let mut fetch_log_futures = vec![];

    // group heartbeat channel
    let (sender, receiver) = mpsc::channel::<bool>(5);
    send_channels.push(sender);
    for leader in &tp_param.leaders {
        let (sender, receiver) = mpsc::channel::<bool>(5);
        let topic_name = tp_param.topic_name.clone();
        send_channels.push(sender);
        fetch_log_futures.push(kf_fetch_log_loop(
            topic_name,
            max_bytes,
            from_beginning,
            leader.clone(),
            response_params.clone(),
            receiver,
        ));
    }

    // attach all send channels to Ctrl-C event handler
    //  - all futures will exit exit on Ctrl-C event
    if let Err(err) = ctrlc::set_handler(move || {
        debug!("<Ctrl-C> received");
        send_ctrlc_signal(send_channels.clone());
    }) {
        return Err(CliError::IoError(IoError::new(
            ErrorKind::InvalidData,
            format!("CTRL-C handler: {}", err),
        )));
    }

    // spin-off all futures
    let _r = join(
        group_heartbeat_loop(conn, vers, grp_id, mbr_id, gen_id, receiver),
        join_all(fetch_log_futures),
    )
    .await;

    Ok(())
}
/// Generate futures for group keep-alive and fetch-logs
Generate futures for group keep-alive and fetch-logs
Generate futures for group keep - alive and fetch - logs
async fn group_and_fetch_log_futures(
    conn: Connection,
    vers: KfApiVersions,
    grp_id: String,
    mbr_id: String,
    gen_id: i32,
    max_bytes: i32,
    from_beginning: bool,
    tp_param: TopicPartitionParam,
    response_params: ResponseLogParams,
) -> Result<(), CliError> {
    let mut send_channels = vec![];
    let mut fetch_log_futures = vec![];
    // group heartbeat channel
    let (sender, receiver) = mpsc::channel::<bool>(5);
    send_channels.push(sender);
    for leader in &tp_param.leaders {
        let (sender, receiver) = mpsc::channel::<bool>(5);
        let topic_name = tp_param.topic_name.clone();
        send_channels.push(sender);
        fetch_log_futures.push(kf_fetch_log_loop(
            topic_name,
            max_bytes,
            from_beginning,
            leader.clone(),
            response_params.clone(),
            receiver,
        ));
    }
    // attach all send channels to Ctrl-C event handler
    //  - all futures will exit exit on Ctrl-C event
    if let Err(err) = ctrlc::set_handler(move || {
        debug!("<Ctrl-C> received");
        send_ctrlc_signal(send_channels.clone());
    }) {
        return Err(CliError::IoError(IoError::new(
            ErrorKind::InvalidData,
            format!("CTRL-C handler: {}", err),
        )));
    }
    // spin-off all futures
    let _r = join(
        group_heartbeat_loop(conn, vers, grp_id, mbr_id, gen_id, receiver),
        join_all(fetch_log_futures),
    )
    .await;
    Ok(())
}
async fn group_and_fetch_log_futures ( conn : Connection , vers : KfApiVersions , grp_id : String , mbr_id : String , gen_id : i32 , max_bytes : i32 , from_beginning : bool , tp_param : TopicPartitionParam , response_params : ResponseLogParams , ) -> Result < ( ) , CliError > { let mut send_channels = vec ! [ ] ; let mut fetch_log_futures = vec ! [ ] ; // group heartbeat channel let ( sender , receiver ) = mpsc :: channel :: < bool > ( 5 ) ; send_channels . push ( sender ) ; for leader in & tp_param . leaders { let ( sender , receiver ) = mpsc :: channel :: < bool > ( 5 ) ; let topic_name = tp_param . topic_name . clone ( ) ; send_channels . push ( sender ) ; fetch_log_futures . push ( kf_fetch_log_loop ( topic_name , max_bytes , from_beginning , leader . clone ( ) , response_params . clone ( ) , receiver , ) ) ; } // attach all send channels to Ctrl-C event handler //  - all futures will exit exit on Ctrl-C event if let Err ( err ) = ctrlc :: set_handler ( move | | { debug ! ( " " ) ; send_ctrlc_signal ( send_channels . clone ( ) ) ; } ) { return Err ( CliError :: IoError ( IoError :: new ( ErrorKind :: InvalidData , format ! ( " " err ) , ) ) ) ; } // spin-off all futures let _r = join ( group_heartbeat_loop ( conn , vers , grp_id , mbr_id , gen_id , receiver ) , join_all ( fetch_log_futures ) , ) . await ; Ok ( ( ) ) }
Generate futures for group keep-alive and fetch-logs
Generate futures for group keep - alive and fetch - logs
conn
Connection
vers
KfApiVersions
grp_id
String
mbr_id
String
gen_id
i32
max_bytes
i32
from_beginning
bool
tp_param
TopicPartitionParam
response_params
ResponseLogParams
conn
Connection
vers
KfApiVersions
grp_id
String
mbr_id
String
gen_id
i32
max_bytes
i32
from_beginning
bool
tp_param
TopicPartitionParam
response_params
ResponseLogParams