/* This file is part of DarkFi (https://dark.fi)
*
* Copyright (C) 2020-2022 Dyne.org foundation
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as
* published by the Free Software Foundation, either version 3 of the
* License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see .
*/
use async_std::sync::{Arc, Mutex};
use async_trait::async_trait;
use darkfi_serial::{SerialDecodable, SerialEncodable};
use log::debug;
use smol::Executor;
use darkfi::{net, Result};
use crate::{buffers::SeenIds, Privmsg};
#[derive(SerialDecodable, SerialEncodable, Clone, Debug)]
struct InvObject(String);
pub struct ProtocolPrivmsg {
jobsman: net::ProtocolJobsManagerPtr,
notify: smol::channel::Sender,
msg_sub: net::MessageSubscription,
p2p: net::P2pPtr,
channel: net::ChannelPtr,
seen: Arc>,
}
impl ProtocolPrivmsg {
pub async fn init(
channel: net::ChannelPtr,
notify: smol::channel::Sender,
p2p: net::P2pPtr,
seen: Arc>,
) -> net::ProtocolBasePtr {
let message_subsytem = channel.get_message_subsystem();
message_subsytem.add_dispatch::().await;
let msg_sub =
channel.clone().subscribe_msg::().await.expect("Missing Privmsg dispatcher!");
Arc::new(Self {
notify,
msg_sub,
jobsman: net::ProtocolJobsManager::new("ProtocolPrivmsg", channel.clone()),
p2p,
channel,
seen,
})
}
async fn handle_receive_msg(self: Arc) -> Result<()> {
debug!(target: "ircd", "ProtocolPrivmsg::handle_receive_msg() [START]");
let exclude_list = vec![self.channel.address()];
loop {
let msg = self.msg_sub.receive().await?;
let msg = (*msg).to_owned();
{
let ids = &mut self.seen.lock().await;
if !ids.push(msg.id) {
continue
}
}
self.notify.send(msg.clone()).await?;
self.p2p.broadcast_with_exclude(msg, &exclude_list).await?;
}
}
}
#[async_trait]
impl net::ProtocolBase for ProtocolPrivmsg {
/// Starts ping-pong keep-alive messages exchange. Runs ping-pong in the
/// protocol task manager, then queues the reply. Sends out a ping and
/// waits for pong reply. Waits for ping and replies with a pong.
async fn start(self: Arc, executor: Arc>) -> Result<()> {
debug!(target: "ircd", "ProtocolPrivmsg::start() [START]");
self.jobsman.clone().start(executor.clone());
self.jobsman.clone().spawn(self.clone().handle_receive_msg(), executor.clone()).await;
debug!(target: "ircd", "ProtocolPrivmsg::start() [END]");
Ok(())
}
fn name(&self) -> &'static str {
"ProtocolPrivmsg"
}
}
impl net::Message for Privmsg {
fn name() -> &'static str {
"privmsg"
}
}