zbus-lib 0.1.6

use std::collections::HashMap;
use std::sync::Arc;
use std::sync::mpsc::{channel, Sender};
use std::thread;
use std::time::Duration;

use log::debug;
use serde_json::Value;
use threadpool::ThreadPool;

use crate::err::{ZbusErr, ZbusResult};
use crate::message::{Message, Request, Response, ResponseBuilder};
use crate::rpc::Protocol;
use crate::wsocket::Instruct;

pub struct IOHandler {
    method: Box<dyn Handler + Send + Sync + 'static>,
}

impl IOHandler {
    pub fn execute(&self, req: Request) -> ZbusResult<Value> {
        self.method.execute(req.body())
    }
}

pub struct IOHandlers {
    methods: HashMap<String, Arc<IOHandler>>,
    pool: ThreadPool,
}

impl IOHandlers {
    pub fn new(n_workers: usize) -> Self {
        Self {
            methods: HashMap::new(),
            pool: ThreadPool::new(n_workers),
        }
    }
    pub fn add_method<H>(&mut self, name: String, handle: H) where H: Handler, H: Send + Sync + 'static {
        self.methods.insert(name, Arc::new(IOHandler {
            method: Box::new(handle)
        }));
    }
    pub fn handler_request(&self, req: Request, tx: Sender<Instruct>) {
        let mut builder = Response::builder();
        let mut headers = HashMap::new();
        headers.insert(Protocol::CMD.into(), Value::from(Protocol::ROUTE));
        let id = req.id().unwrap();/// 这个存在判断是在调用rpc前验证过的
        let req_headers = req.headers();
        let source = req_headers.get(Protocol::SOURCE);
        source.map(|s| headers.insert(Protocol::TARGET.into(), s.clone()));
        headers.insert(Protocol::ID.into(), Value::from(id));
        let url = req.url();
        if url == "" {
            tx.send(Instruct::Delivery(reply(builder, headers, 400, "url required"), None));
            return;
        }
        let method = self.methods.get(req.url());
        match method {
            None => {
                debug!("404 {} ", req.url());
                tx.send(Instruct::Delivery(reply(builder, headers, 404, format!("URL={} Not Found", url)), None));
            }
            Some(handler) => {
                let handler = handler.clone();
                self.pool.execute(move || {
                    let result: ZbusResult<Value> = handler.method.execute(req.body());
                    let response = match result {
                        Ok(body) => {
                            headers.insert("Content-Type".into(), Value::from("application/json; charset=utf8"));
                            builder.status(200).headers(headers).body(body);
                            Message::Response(builder.build())
                        }
                        Err(e) => {
                            reply(builder, headers, 500, format!("request params is error {}", e))
                        }
                    };
                    tx.send(Instruct::Delivery(response, None));
                });
            }
        }
    }
}

fn reply<S: Into<String>>(mut builder: ResponseBuilder, mut headers: HashMap<String, Value>, status: u32, message: S) -> Message {
    builder.status(status);
    headers.insert("Content-Type".into(), Value::from("text/plain; charset=utf8"));
    builder.body(Value::String(message.into()));
    Message::Response(builder.build())
}

pub trait Handler: Send + Sync {
    fn execute(&self, params: Value) -> ZbusResult<Value>;
}

impl<F> Handler for F where F: Fn(Value) -> ZbusResult<Value>, F: Sync + Send {
    fn execute(&self, params: Value) -> ZbusResult<Value> { self(params) }
}

#[test]
fn handler_test() {
    let (tx, rx) = channel::<(Request, Sender<Instruct>)>();
    let mut io_handler = IOHandlers::new(10);
    io_handler.add_method("test".into(), |req: Value| {
        Ok(Value::Null)
    });
    io_handler.add_method("test1".into(), |req: Value| {
        Ok(Value::Null)
    });
    thread::spawn(move || {
        loop {
            if let Ok((req, sender)) = rx.recv() {
                io_handler.handler_request(req, sender);
            }
        }
    });
    let (sender, receiv) = channel::<Instruct>();
    for i in 0..10 {
        let req = Request::builder().method("test").build();
        tx.send((req, sender.clone()));
    }
    thread::sleep(Duration::from_secs(10));
}