Let's start tying these concepts together. We'll define a struct called ProtocolDchat that contains a MessageSubscription to DchatMsg and a pointer to the ProtocolJobsManager. We'll also include the DchatMsgsBuffer in the struct as it will come in handy later on.

use async_std::sync::Arc;
use async_trait::async_trait;
use darkfi::{net, Result};
use log::debug;
use smol::Executor;

use crate::dchatmsg::{DchatMsg, DchatMsgsBuffer};

pub struct ProtocolDchat {
    jobsman: net::ProtocolJobsManagerPtr,
    msg_sub: net::MessageSubscription<DchatMsg>,
    msgs: DchatMsgsBuffer,

Next we'll implement the trait ProtocolBase. ProtocolBase requires two functions, start() and name(). In start() we will start up the ProtocolJobsManager. name() will return a str of the protocol name.

impl net::ProtocolBase for ProtocolDchat {
    async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {

    fn name(&self) -> &'static str {

Once that's done, we'll need to create a ProtocolDchat constructor that we will pass to the ProtocolRegistry to register our protocol. We'll invoke the MessageSubsystem and add DchatMsg as to the list of dispatchers. Next, we'll create a MessageSubscription to DchatMsg using the method subscribe_msg().

We'll also initialize the ProtocolJobsManager and finally return a pointer to the protocol.

impl ProtocolDchat {
    pub async fn init(channel: net::ChannelPtr, msgs: DchatMsgsBuffer) -> net::ProtocolBasePtr {
        debug!(target: "dchat", "ProtocolDchat::init() [START]");
        let message_subsytem = channel.get_message_subsystem();

        let msg_sub =
            channel.subscribe_msg::<DchatMsg>().await.expect("Missing DchatMsg dispatcher!");

        Arc::new(Self {
            jobsman: net::ProtocolJobsManager::new("ProtocolDchat", channel.clone()),

We're nearly there. But right now the protocol doesn't actually do anything. Let's write a method called handle_receive_msg() which receives a message on our MessageSubscription and adds it to DchatMsgsBuffer.

Put this inside the ProtocolDchat implementation:

    async fn handle_receive_msg(self: Arc<Self>) -> Result<()> {
        debug!(target: "dchat", "ProtocolDchat::handle_receive_msg() [START]");
        while let Ok(msg) = self.msg_sub.receive().await {
            let msg = (*msg).to_owned();


As a final step, let's add that task to the ProtocolJobManager that is invoked in start():

    async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
        debug!(target: "dchat", "ProtocolDchat::ProtocolBase::start() [START]");
        self.jobsman.clone().spawn(self.clone().handle_receive_msg(), executor.clone()).await;
        debug!(target: "dchat", "ProtocolDchat::ProtocolBase::start() [STOP]");