pi_logger 0.8.0

use derivative::Derivative;
use io::ErrorKind;
use log::Record;

use log4rs::{
    append::Append,
    config::{Deserialize, Deserializers},
    encode::{
        self, pattern::PatternEncoder, writer::simple::SimpleWriter, Encode, EncoderConfig, Style,
    },
};

use async_httpc::{
    AsyncHttpRequest, AsyncHttpRequestBody, AsyncHttpRequestMethod, AsyncHttpc, AsyncHttpcBuilder,
};

use pi_async_rt::rt::{
    multi_thread::{MultiTaskRuntime, MultiTaskRuntimeBuilder, StealableTaskPool},
    single_thread::{SingleTaskRunner, SingleTaskRuntime},
    spawn_worker_thread, AsyncRuntime
};

use lazy_static::lazy_static;
use std::sync::RwLock;
use std::{
    fmt::{self, Debug},
    fs::{self, File, OpenOptions},
    io::{self, BufWriter, Error, Write},
    path::{Path, PathBuf},
    sync::Arc,
    thread,
    time::Duration,
};

/// An appender which logs to a http.
#[derive(Derivative)]
#[derivative(Debug)]
pub struct HttpAppender {
    write: HttpWrite,
    encoder: Box<dyn Encode>,
}

impl Append for HttpAppender {
    fn append(&self, record: &Record) -> anyhow::Result<()> {
        let mut write = self.write.clone();
        self.encoder.encode(&mut write, record)?;
        write.flush()?;
        Ok(())
    }

    fn flush(&self) {}
}

impl HttpAppender {
    /// Creates a new `HttpAppender` builder.
    pub fn builder() -> HttpAppenderBuilder {
        HttpAppenderBuilder {
            encoder: None,
            append: true,
        }
    }
}

/// A builder for `HttpAppender`s.
pub struct HttpAppenderBuilder {
    encoder: Option<Box<dyn Encode>>,
    append: bool,
}

impl HttpAppenderBuilder {
    /// Sets the output encoder for the `HttpAppender`.
    pub fn encoder(mut self, encoder: Box<dyn Encode>) -> HttpAppenderBuilder {
        self.encoder = Some(encoder);
        self
    }

    /// Determines if the appender will append to or truncate the output file.
    ///
    /// Defaults to `true`.
    pub fn append(mut self, append: bool) -> HttpAppenderBuilder {
        self.append = append;
        self
    }

    /// Consumes the `HttpAppenderBuilder`, producing a `HttpAppender`.
    /// The path argument can contain environment variables of the form $ENV{name_here},
    /// where 'name_here' will be the name of the environment variable that
    /// will be resolved. Note that if the variable fails to resolve,
    /// $ENV{name_here} will NOT be replaced in the path.
    pub fn build(
        self,
        url: String,
        rt: MultiTaskRuntime<()>,
        httpc: AsyncHttpc,
        authorization: String,
    ) -> io::Result<HttpAppender> {
        let http_w = HttpWrite {
            url,
            rt,
            buf: Vec::new(),
            httpc,
            authorization,
        };

        Ok(HttpAppender {
            write: http_w,
            encoder: self
                .encoder
                .unwrap_or_else(|| Box::new(PatternEncoder::default())),
        })
    }
}

#[derive(Clone)]
struct HttpWrite {
    buf: Vec<u8>,
    rt: MultiTaskRuntime<()>,
    url: String,
    httpc: AsyncHttpc,
    authorization: String,
}

impl Debug for HttpWrite {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        Ok(())
    }
}

impl io::Write for HttpWrite {
    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
        self.buf.write_all(buf);
        Ok(buf.len())
    }

    fn flush(&mut self) -> io::Result<()> {
        let httpc = self.httpc.clone();
        let buf = self.buf.clone();
        let url = self.url.clone();
        let rt = self.rt.clone();
        let auth = self.authorization.clone();
        self.rt.spawn(async move {
            http_request(httpc, url, buf, rt, auth).await;
        });
        self.buf.clear();
        Ok(())
    }
}

impl encode::Write for HttpWrite {
    fn set_style(&mut self, style: &Style) -> io::Result<()> {
        // self.0.set_style(style)
        Ok(())
    }
}

/// 发送日志到logstash
pub async fn http_request(
    httpc: AsyncHttpc,
    url: String,
    body: Vec<u8>,
    rt: MultiTaskRuntime<()>,
    authorization: String,
) -> io::Result<Vec<u8>> {

    let body = AsyncHttpRequestBody::with_binary(body);
    let httpc_copy = httpc.clone();

    let mut resp = httpc_copy
        .build_request(&url, AsyncHttpRequestMethod::Post)
        .add_header("Content-Type", "application/json")
        .add_header("Authorization", &authorization)
        .set_body(body)
        .send()
        .await?;
    let mut bodyVec: Vec<u8> = Vec::new();
    loop {
        match resp.get_body().await? {
            Some(body) => {
                bodyVec.write_all(&*body);
            }
            None => {
                return Ok(bodyVec);
            }
        }
    }
}