O01eg/sendxmppd-rs - Change 7WM64CI7MA25GOY5BEQQQFUXNKLXSAN65VX7XA3UGJKGS6OVLUAQC

Implement all previous functionality

Created by O01eg on September 29, 2020

7WM64CI7MA25GOY5BEQQQFUXNKLXSAN65VX7XA3UGJKGS6OVLUAQC

Dependencies

In channels

main

Change contents

Insertion in src/xmpp/xmpp_connection.rs at line 862 [4.21]

[3.29419]

[2.19220]

    fn process_jid(&mut self, xmpp_to: &xmpp_parsers::Jid) {
        if let Some(ref mut mailbox) = self.state.data.outgoing_mailbox.get_mut(xmpp_to) {
            if !mailbox.is_empty() {
                if let Some(ref mut rdata) = self
                    .state
                    .data
                    .roster
                    .get_mut(&std::convert::From::from(xmpp_to.clone()))
                {
                    info!("Jid {} in roster", xmpp_to);
                    let sub_to = match rdata.0 {
                        xmpp_parsers::roster::Subscription::To => true,
                        xmpp_parsers::roster::Subscription::Both => true,
                        _ => false,
                    };
                    if sub_to {
                        info!("Subscribed to {}", xmpp_to);
                        self.state.data.send_queue.extend(
                            mailbox.drain(..).map(|message| {
                                stanzas::make_chat_message(xmpp_to.clone(), message)
                            }),
                        );
                    } else if rdata.1 == xmpp_parsers::roster::Ask::None {
                        info!("Not subscribed to {}", xmpp_to);
                        self.state.data.counter += 1;
                        let id_presence_subscribe =
                            format!("id_presence_subscribe{}", self.state.data.counter);
                        self.state
                            .data
                            .send_queue
                            .push_back(stanzas::make_ask_subscribe(
                                id_presence_subscribe,
                                xmpp_to.clone(),
                            ));
                    } else {
                        warn!(
                            "Not subscribed to {}. Currently in {:?} state",
                            xmpp_to, rdata.1
                        );
                        self.state.data.counter += 1;
                        let id_presence_subscribe =
                            format!("id_presence_subscribe{}", self.state.data.counter);
                        self.state
                            .data
                            .send_queue
                            .push_back(stanzas::make_ask_subscribe(
                                id_presence_subscribe,
                                xmpp_to.clone(),
                            ));
                    }
                    let sub_from = match rdata.0 {
                        xmpp_parsers::roster::Subscription::From => true,
                        xmpp_parsers::roster::Subscription::Both => true,
                        _ => false,
                    };
                    if !sub_from {
                        info!("Not subscription from {}", xmpp_to);
                        self.state.data.counter += 1;
                        let id_presence_subscribed =
                            format!("id_presence_subscribed{}", self.state.data.counter);
                        self.state
                            .data
                            .send_queue
                            .push_back(stanzas::make_allow_subscribe(
                                id_presence_subscribed,
                                xmpp_to.clone(),
                            ));
                    }
                } else {
                    info!("Jid {} not in roster", xmpp_to);
                    self.state.data.counter += 1;
                    let id_add_roster = format!("id_add_roster{}", self.state.data.counter);
                    let add_roster = stanzas::make_add_roster(
                        &id_add_roster,
                        std::convert::From::from(xmpp_to.clone()),
                    );
                    info!("Adding jid {} to roster id {}", xmpp_to, id_add_roster);
                    self.state.data.pending_ids.insert(
                        id_add_roster,
                        IqWait::new(
                            60,
                            AddRosterIqHandler {
                                jid: std::convert::From::from(xmpp_to.clone()),
                            },
                        ),
                    );
                    self.state.data.send_queue.push_back(add_roster);
                }
            }
        }
    }

Insertion in src/xmpp/xmpp_connection.rs at line 956 [4.21]

[3.29527]

[3.31069]

            XmppCommand::Chat { xmpp_to, message } => {
                self.state
                    .data
                    .outgoing_mailbox
                    .entry(xmpp_to.clone())
                    .or_default()
                    .push(message);
                self.process_jid(&xmpp_to);
            }
            XmppCommand::Chatroom { muc_id, message } => {
                if let Some(muc) = self.state.data.mucs.get(&muc_id) {
                    self.state
                        .data
                        .send_queue
                        .push_back(stanzas::make_muc_message(muc.clone(), message));
                } else {
                    error!("Not found MUC {}", muc_id);
                }
            }
            XmppCommand::ChatroomPresence {
                muc_id,
                show,
                message,
            } => {
                if let Some(muc) = self.state.data.mucs.get(&muc_id) {
                    self.state.data.counter += 1;
                    let id_presence = format!("id_presence{}", self.state.data.counter);
                    self.state
                        .data
                        .send_queue
                        .push_back(stanzas::make_muc_presence(
                            &id_presence,
                            self.account.jid.clone(),
                            muc.clone(),
                            Some(show),
                            Some(message),
                        ));
                } else {
                    error!("Not found MUC {}", muc_id);
                }
            }

Replacement in src/xmpp/xmpp_connection.rs at line 1078 [4.21]

∅:D[2.21530] → [2.21530:21608]

                    Either::Right((Ok(_), stop_future)) => Some(stop_future),

[2.21530]

[2.21608]

                    Either::Right((Ok(_), stop_future)) => {
                        self.state
                            .data
                            .mucs
                            .insert(chatroom.0.clone(), chatroom.1.clone());
                        Some(stop_future)
                    }

Replacement in src/xmpp/element_processor.rs at line 0 [5.11301]

∅:D[5.11301] → [2.26659:27756]

type Func<S, T, E> = dyn Fn(&mut S, E) -> T + Sync;
pub struct Processor<S: 'static, T: 'static, E: Clone + 'static> {
    processors: Vec<Box<Func<S, Option<T>, E>>>,
    default: &'static Func<S, T, E>,
}
impl<S: 'static, T: 'static, E: Clone + 'static> Processor<S, T, E> {
    pub fn new<F>(f: &'static F) -> Processor<S, T, E>
    where
        F: Fn(&mut S, E) -> T + Sync + 'static,
    {
        Processor {
            processors: vec![],
            default: f,
        }
    }
    pub fn register<F, A>(&mut self, f: &'static F)
    where
        F: Fn(&mut S, A) -> T + Sync + 'static,
        A: std::convert::TryFrom<E>,
    {
        self.processors.push(Box::new(move |s, e: E| {
            use std::convert::TryInto;
            (e.try_into().ok() as Option<A>).map(|a| f(s, a))
        }));
    }
    pub fn process(&self, s: &mut S, e: E) -> T {
        for processor in self.processors.iter() {
            match processor(s, e.clone()) {
                Some(t) => return t,
                None => continue,
            }
        }
        (*self.default)(s, e)
    }
}

[5.11301]

type Func<S, T, E> = dyn Fn(&mut S, E) -> T + Sync;
pub struct Processor<S: 'static, T: 'static, E: Clone + 'static> {
    processors: Vec<Box<Func<S, Option<T>, E>>>,
    default: &'static Func<S, T, E>,
}
impl<S: 'static, T: 'static, E: Clone + 'static> Processor<S, T, E> {
    pub fn new<F>(f: &'static F) -> Processor<S, T, E>
    where
        F: Fn(&mut S, E) -> T + Sync + 'static,
    {
        Processor {
            processors: vec![],
            default: f,
        }
    }
    pub fn register<F, A>(&mut self, f: &'static F)
    where
        F: Fn(&mut S, A) -> T + Sync + 'static,
        A: std::convert::TryFrom<E>,
    {
        self.processors.push(Box::new(move |s, e: E| {
            use std::convert::TryInto;
            (e.try_into().ok() as Option<A>).map(|a| f(s, a))
        }));
    }
    pub fn process(&self, s: &mut S, e: E) -> T {
        for processor in self.processors.iter() {
            match processor(s, e.clone()) {
                Some(t) => return t,
                None => continue,
            }
        }
        (*self.default)(s, e)
    }
}

Replacement in src/main.rs at line 103 [6.17]

∅:D[2.31243] → [2.31243:31338]

                                        .body(Body::from(format!("Cann't get presence text")))

[2.31243]

[2.31338]

                                        .body(Body::from(format!("Cann't send command")))
                                        .map_err(std::convert::Into::into)
                                }
                            }
                        }
                        Err(e) => Response::builder()
                            .status(hyper::StatusCode::BAD_REQUEST)
                            .body(Body::from(format!("Cann't get presence text")))
                            .map_err(std::convert::Into::into),
                    }
                })
            }
            (None, Ok(xmpp_to), _) => {
                info!("Got chat request. Reading body...");
                let mut cmd_send = self.cmd_send.clone();
                Box::pin(async move {
                    match body_to_string(req).await {
                        Ok(message) => {
                            match cmd_send.send(XmppCommand::Chat { xmpp_to, message }).await {
                                Ok(_) => Response::builder()
                                    .body(Body::from(format!("Accepted")))
                                    .map_err(std::convert::Into::into),
                                Err(e) => {
                                    error!("Cann't send chat command: {}", e);
                                    Response::builder()
                                        .status(hyper::StatusCode::BAD_REQUEST)
                                        .body(Body::from(format!("Cann't send command")))
                                        .map_err(std::convert::Into::into)
                                }
                            }
                        }
                        Err(e) => Response::builder()
                            .status(hyper::StatusCode::BAD_REQUEST)
                            .body(Body::from(format!("Cann't get chat text")))
                            .map_err(std::convert::Into::into),
                    }
                })
            }
            (Some(Ok(muc_id)), _, Ok(show)) => {
                info!("Got MUC presence request. Reading body...");
                let mut cmd_send = self.cmd_send.clone();
                Box::pin(async move {
                    match body_to_string(req).await {
                        Ok(message) => {
                            match cmd_send
                                .send(XmppCommand::ChatroomPresence {
                                    muc_id,
                                    show,
                                    message,
                                })
                                .await
                            {
                                Ok(_) => Response::builder()
                                    .body(Body::from(format!("Accepted")))
                                    .map_err(std::convert::Into::into),
                                Err(e) => {
                                    error!("Cann't send MUC presence command: {}", e);
                                    Response::builder()
                                        .status(hyper::StatusCode::BAD_REQUEST)
                                        .body(Body::from(format!("Cann't send command")))

Replacement in src/main.rs at line 174 [6.17]

∅:D[7.50472] → [2.31793:32912]

            (None, Ok(xmpp_to), _) => Box::pin(async {
                Response::builder()
                    .status(hyper::StatusCode::BAD_REQUEST)
                    .body(Body::from(format!("SOON")))
                    .map_err(std::convert::Into::into)
            }),
            (Some(Ok(muc_id)), _, Ok(show)) => Box::pin(async {
                Response::builder()
                    .status(hyper::StatusCode::BAD_REQUEST)
                    .body(Body::from(format!("SOON")))
                    .map_err(std::convert::Into::into)
            }),
            (Some(Ok(muc_id)), _, _) => Box::pin(async {
                Response::builder()
                    .status(hyper::StatusCode::BAD_REQUEST)
                    .body(Body::from(format!("SOON")))
                    .map_err(std::convert::Into::into)
            }),
            (Some(Err(err)), _, _) => Box::pin(async {
                Response::builder()
                    .status(hyper::StatusCode::BAD_REQUEST)
                    .body(Body::from(format!("SOON")))
                    .map_err(std::convert::Into::into)
            }),

[7.50472]

[7.50966]

            (Some(Ok(muc_id)), _, _) => {
                info!("Got MUC chat request. Reading body...");
                let mut cmd_send = self.cmd_send.clone();
                Box::pin(async move {
                    match body_to_string(req).await {
                        Ok(message) => {
                            match cmd_send
                                .send(XmppCommand::Chatroom { muc_id, message })
                                .await
                            {
                                Ok(_) => Response::builder()
                                    .body(Body::from(format!("Accepted")))
                                    .map_err(std::convert::Into::into),
                                Err(e) => {
                                    error!("Cann't send MUC chat command: {}", e);
                                    Response::builder()
                                        .status(hyper::StatusCode::BAD_REQUEST)
                                        .body(Body::from(format!("Cann't send command")))
                                        .map_err(std::convert::Into::into)
                                }
                            }
                        }
                        Err(e) => Response::builder()
                            .status(hyper::StatusCode::BAD_REQUEST)
                            .body(Body::from(format!("Cann't get chat text")))
                            .map_err(std::convert::Into::into),
                    }
                })
            }
            (Some(Err(err)), _, _) => {
                warn!("Unknown muc destination: {}", err);
                Box::pin(async move {
                    Response::builder()
                        .status(hyper::StatusCode::BAD_REQUEST)
                        .body(Body::from(format!("Unknown muc destination: {}", err,)))
                        .map_err(std::convert::Into::into)
                })
            }