mtxspooler/src/client.rs

use std::collections::*;
use std::path::PathBuf;
use std::sync::Arc;

use futures::compat::Stream01CompatExt;
use futures::prelude::*;

use hyper::client::connect::dns::GaiResolver;
use hyper::client::connect::HttpConnector;
use hyper_tls::HttpsConnector;

use inotify::ffi::*;

use tokio::fs as tokiofs;
use tokio::prelude::*;
use tokio::sync::mpsc;

use tokio_inotify;

use url::Url;

use ruma_client::Client;
use ruma_client_api::r0::message as rumamessage;
use ruma_events::{self, room::message::*};
use ruma_identifiers::RoomId;

use crate::config::*;

#[derive(Debug)]
pub enum Error {
    BadUrl,
    MtxClient(ruma_client::Error),
}

impl From<ruma_client::Error> for Error {
    fn from(f: ruma_client::Error) -> Self {
        Self::MtxClient(f)
    }
}

type MatrixClient = Client<HttpsConnector<HttpConnector<GaiResolver>>>;
type MessageRequest = rumamessage::create_message_event::Request;

pub struct Message {
    dest_room: RoomId,
    msg: String,
    delay_secs: u32,
}

impl Message {
    pub fn new(dest_room: RoomId, msg: String) -> Self {
        Self::new_delay(dest_room, msg, 0)
    }

    pub fn new_delay(dest_room: RoomId, msg: String, delay_secs: u32) -> Self {
        Self {
            dest_room,
            msg,
            delay_secs,
        }
    }
}

pub async fn create_and_auth_client(acct: Account) -> Result<Arc<MatrixClient>, Error> {
    let hs_url = Url::parse(&acct.homeserver).map_err(|_| Error::BadUrl)?;
    let c = MatrixClient::https(hs_url, None);
    match acct.auth {
        Auth::UsernamePass(un, pw) => c.log_in(un, pw, acct.device_id, acct.display).await?,
    };
    Ok(Arc::new(c))
}

pub async fn submit_messages(cli: Arc<MatrixClient>, mut recv: mpsc::Receiver<Message>) {
    let mut rng = rand::thread_rng();

    while let Some(msg) = recv.recv().await {
        let req = make_text_request(msg.dest_room.clone(), &msg.msg, &mut rng);
        if let Err(e) = cli.request(req).await {
            panic!("[client] error sending request: {:?}", e);
        }
    }
}

fn make_text_request<R: rand::Rng>(room_id: RoomId, msg: &str, rng: &mut R) -> MessageRequest {
    let inner = TextMessageEventContent {
        body: String::from(msg),
        format: None,
        formatted_body: None,
        relates_to: None,
    };
    let mec = MessageEventContent::Text(inner);
    MessageRequest {
        room_id: room_id,
        event_type: ruma_events::EventType::RoomMessage,
        txn_id: make_txn_id(rng),
        data: mec,
    }
}

const TXN_ID_LEN: usize = 20;

fn make_txn_id<R: rand::Rng>(rng: &mut R) -> String {
    let mut buf = String::with_capacity(TXN_ID_LEN);
    for _ in 0..TXN_ID_LEN {
        buf.push((rng.gen_range(0, 26) + ('a' as u8)) as char);
    }
    buf
}