pub mod registry {
    tonic::include_proto!("bky.registry.api");
}

use futures::stream::Stream;
use std::time::Duration;
use tokio_stream::StreamExt;
use tonic::{codegen::http::response, server::ClientStreamingService, transport::Channel};

use registry::{metadata_ops_client::MetadataOpsClient, Metadata, PullCreateReq};

fn generate_request() -> impl Stream<Item = PullCreateReq> {
    tokio_stream::iter(1..3).map(|i| PullCreateReq {
        tcode: format!("tcode:{:02}", i),
        sendtype: i as i32,
    })
}

fn my_request() -> impl Stream<Item = PushCreateReq> {
    let rdata = fs::read("./test.txt");
    // String::from_utf8_lossy(&fs::read("address.txt").unwrap());
    let mut Send_metadata = vec![];
    match rdata {
        Ok(data) => {
            let c_data = std::str::from_utf8(&data).unwrap();
            // println!("data = {}", c_data.clone());
            let d = Metadata {
                kv: c_data.to_string(),
            };
            Send_metadata.push(d);
            return tokio_stream::iter(1..2).map(move |i| PushCreateReq {
                tcode: "my_tcode".to_string(),
                sendtype: 1,
                metadata: Send_metadata.clone(),
            });
        }
        Err(e) => {
            panic!("dfdf");
            // println!("d={:?}", e);
        }
    }
}

fn pull_request() -> impl Stream<Item = PullCreateReq> {
    tokio_stream::iter(1..2).map(|i| PullCreateReq {
        tcode: format!("tcode:{:02}", i),
        sendtype: i as i32,
    })
}

async fn create_by_pull(client: &mut MetadataOpsClient<Channel>) {
    // let in_stream = generate_request().throttle(dur);
    let in_stream = pull_request();
    let respone = client.create_by_pull(in_stream).await.unwrap();
    let mut resp_stream = respone.into_inner();
    while let Some(received) = resp_stream.next().await {
        let received = received.unwrap();
        println!("received :{:?}", received.md[0].kv);
    }

    // let in_stream_2 = generate_request();
    // let respone = client.create_by_pull(in_stream_2).await.unwrap();
    // let mut resp_stream = respone.into_inner();
    // while let Some(received) = resp_stream.next().await {
    //     let received = received.unwrap();
    //     println!("received :{}", received.tcode);
    // }
}

use std::fs;

use crate::registry::PushCreateReq;

async fn create_by_push(client: &mut MetadataOpsClient<Channel>) {
    // let rdata = fs::read("./test.txt");
    // // String::from_utf8_lossy(&fs::read("address.txt").unwrap());
    // match rdata {
    //     Ok(data) => {
    //         println!("data = {}", std::str::from_utf8(&data).unwrap());

    let in_stream = my_request();
    let respone = client.create_by_push(in_stream).await.unwrap();
    let mut resp_stream = respone.into_inner();
    while let Some(received) = resp_stream.next().await {
        let received = received.unwrap();
        println!("received: {}", received.tcode);
    }
    // let req_data = PushCreateReq {
    //     sendtype: 1,
    //     tcode: "hello".to_string(),
    //     metadata: Vec::new(),
    // };
    // let b = tokio_stream::iter(1..3).map(|i| PullCreateReq {
    //     tcode: format!("tcode:{:02}", i),
    //     sendtype: i as i32,
    // });
    // // client.record_route(Request::new(stream::iter(messages.clone())));
    // let respone = client
    //     .create_by_push(b as impl Stream<Item = PullCreateReq>)
    //     .await
    //     .unwrap();
    //     }
    //     Err(e) => println!("d={:?}", e),
    // }
    // let respone = client.create_by_push()
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut client = MetadataOpsClient::connect("http://192.168.110.183:50050")
        // let mut client = MetadataOpsClient::connect("http://192.168.110.183:50050")
        .await
        .unwrap();
    println!("start bidirectional stream echo ");
    create_by_pull(&mut client).await;
    // create_by_push(&mut client).await;
    Ok(())
}
