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

use futures::Stream;
use registry::{
    Metadata, PullCreateReq, PullCreateResp, PushCreateReq, PushCreateResp, WatchUpdateReq,
    WatchUpdateResp,
};
use std::{
    error::Error, io::ErrorKind, net::ToSocketAddrs, pin::Pin, result, thread::sleep,
    time::Duration,
};
use tokio::sync::mpsc;
use tokio_stream::{wrappers::ReceiverStream, StreamExt};
use tonic::{transport::Server, Request, Response, Status, Streaming};

type PushCreateResult<T> = Result<Response<T>, Status>;
#[derive(Debug)]
pub struct RegistryServer {}

type RegistryResult<T> = Result<Response<T>, Status>;
type ResponseStream = Pin<Box<dyn Stream<Item = Result<PullCreateResp, Status>> + Send>>;

fn match_for_io_error(err_status: &Status) -> Option<&std::io::Error> {
    let mut err: &(dyn Error + 'static) = err_status;

    loop {
        if let Some(io_err) = err.downcast_ref::<std::io::Error>() {
            return Some(io_err);
        }

        // h2::Error do not expose std::io::Error with `source()`
        // https://github.com/hyperium/h2/pull/462
        if let Some(h2_err) = err.downcast_ref::<h2::Error>() {
            if let Some(io_err) = h2_err.get_io() {
                return Some(io_err);
            }
        }

        err = match err.source() {
            Some(err) => err,
            None => return None,
        };
    }
}
use std::fs;
use std::{thread, time};
#[tonic::async_trait]
impl registry::metadata_ops_server::MetadataOps for RegistryServer {
    type CreateByPullStream = ResponseStream;
    async fn create_by_pull(
        &self,
        req: Request<Streaming<PullCreateReq>>,
    ) -> RegistryResult<Self::CreateByPullStream> {
        println!("create_by_pull ");
        println!("req={:?}", req);
        let mut in_stream = req.into_inner();
        println!("in_stream={:?}", in_stream);
        let (tx, rx) = mpsc::channel(128);
        tokio::spawn(async move {
            println!("tokio spawn..");
            while let Some(result) = in_stream.next().await {
                println!("result = {:?}", result);
                match result {
                    Ok(v) => {
                        let rdata = fs::read("./test_rcv.txt");
                        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);
                            }
                            Err(e) => {
                                panic!("dfdf");
                                // println!("d={:?}", e);
                            }
                        }

                        tx.send(Ok(PullCreateResp {
                            tcode: v.tcode,
                            ret_code: 0,
                            md: Send_metadata,
                        }))
                        .await
                        .expect("working rx");
                    }
                    Err(err) => {
                        if let Some(io_err) = match_for_io_error(&err) {
                            if io_err.kind() == ErrorKind::BrokenPipe {
                                eprintln!("\tclient disconnected: broken pipe");
                                break;
                            }
                        }
                        match tx.send(Err(err)).await {
                            Ok(_) => (),
                            Err(_err) => break,
                        }
                    }
                }
                println!("\tstream ended");
            }
        });
        let out_stream = ReceiverStream::new(rx);
        Ok(Response::new(Box::pin(out_stream)))
    }

    //======================================
    type CreateByPushStream = Pin<Box<dyn Stream<Item = Result<PushCreateResp, Status>> + Send>>;
    async fn create_by_push(
        &self,
        req: Request<Streaming<PushCreateReq>>,
    ) -> RegistryResult<Self::CreateByPushStream> {
        println!("create_by_push ");
        println!("req={:?}", req);
        let mut in_stream = req.into_inner();
        println!("in_stream={:?}", in_stream);
        let (tx, rx) = mpsc::channel(128);
        tokio::spawn(async move {
            println!("tokio spawn..");
            while let Some(result) = in_stream.next().await {
                // println!("result = {:?}", result);
                match result {
                    Ok(v) => {
                        let mut r_metadata = &v.clone().metadata[0].kv;
                        // println!("tx={:?}", r_metadata);
                        let wdata_ret = fs::write("./test_rcv.txt", r_metadata);
                        println!("wdata_ret = {:?}", wdata_ret);
                        tx.send(Ok(PushCreateResp {
                            tcode: v.tcode,
                            ret_code: 0,
                        }))
                        .await
                        .expect("working rx");
                    }
                    Err(err) => {
                        if let Some(io_err) = match_for_io_error(&err) {
                            if io_err.kind() == ErrorKind::BrokenPipe {
                                eprintln!("\tclient disconnected: broken pipe");
                                break;
                            }
                        }
                        match tx.send(Err(err)).await {
                            Ok(_) => (),
                            Err(_err) => break,
                        }
                    }
                }
                println!("\tstream ended");
            }
        });
        let out_stream = ReceiverStream::new(rx);
        Ok(Response::new(Box::pin(out_stream)))
    }
    //======================================
    type WatchUpdateStream = Pin<Box<dyn Stream<Item = Result<WatchUpdateResp, Status>> + Send>>;
    async fn watch_update(
        &self,
        req: Request<Streaming<WatchUpdateReq>>,
    ) -> RegistryResult<Self::WatchUpdateStream> {
        println!("watch_update ");
        println!("req={:?}", req);
        let mut in_stream = req.into_inner();
        println!("in_stream={:?}", in_stream);
        let (tx, rx) = mpsc::channel(128);
        tokio::spawn(async move {
            println!("tokio spawn..");
            while let Some(result) = in_stream.next().await {
                // println!("result = {:?}", result);
                match result {
                    Ok(v) => {
                        println!("watch_update sleep");
                        // let ten_millis = time::Duration::from_millis(3000);
                        // thread::sleep(ten_millis);
                        // let sleep_time = ;
                        let sleep_op = tokio::time::sleep(tokio::time::Duration::from_secs(4));
                        println!("watch_update() ...");
                        loop {
                            sleep_op.await;
                            if true {
                                tx.send(Ok(WatchUpdateResp {
                                    client_id: v.client_id,
                                    notify_msg: "hello".to_string(),
                                }))
                                .await
                                .expect("working rx");
                            }
                            break;
                        }
                    }
                    Err(err) => {
                        if let Some(io_err) = match_for_io_error(&err) {
                            if io_err.kind() == ErrorKind::BrokenPipe {
                                eprintln!("\tclient disconnected: broken pipe");
                                break;
                            }
                        }
                        match tx.send(Err(err)).await {
                            Ok(_) => (),
                            Err(_err) => break,
                        }
                    }
                }
                println!("\tstream ended");
            }
        });
        let out_stream = ReceiverStream::new(rx);
        Ok(Response::new(Box::pin(out_stream)))
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    println!("main...");
    let server = RegistryServer {};
    Server::builder()
        .add_service(registry::metadata_ops_server::MetadataOpsServer::new(
            server,
        ))
        .serve(
            "192.168.110.183:50050"
                .to_socket_addrs()
                .unwrap()
                .next()
                .unwrap(),
        )
        .await
        .unwrap();
    Ok(())
}
