ddk 1.0.11

application tooling for DLCs 🌊
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

use crate::logger::{log_info, log_warn, WriteLog};
use crate::{error::TransportError, DlcDevKitDlcManager, Oracle, Storage, Transport};
use async_trait::async_trait;
use bitcoin::secp256k1::PublicKey;
use lightning_net_tokio::connect_outbound;
use std::sync::Arc;
use tokio::sync::watch;

pub(crate) mod peer_manager;
pub use peer_manager::LightningTransport;

#[async_trait]
impl Transport for LightningTransport {
    fn name(&self) -> String {
        "lightning".into()
    }

    fn public_key(&self) -> PublicKey {
        self.node_id
    }

    /// Sends a message to a peer.
    async fn send_message(&self, counterparty: PublicKey, message: ddk_messages::Message) {
        log_info!(
            self.logger,
            "Sending message to counter_party={}",
            counterparty.to_string()
        );
        if self.peer_manager.peer_by_node_id(&counterparty).is_some() {
            self.message_handler.send_message(counterparty, message);
            self.peer_manager.process_events();
        } else {
            log_warn!(
                self.logger,
                "Not connected to counterparty. Message not sent. counter_party={}",
                counterparty.to_string()
            );
        }
    }

    /// Gets and clears the message queue with messages to be processed.
    /// Takes the manager to process the DLC messages that are received.
    async fn start<S: Storage, O: Oracle>(
        &self,
        mut stop_signal: watch::Receiver<bool>,
        manager: Arc<DlcDevKitDlcManager<S, O>>,
    ) -> Result<(), TransportError> {
        let listen_handle = self.listen(stop_signal.clone());

        let process_handle = self.process_messages(stop_signal.clone(), manager.clone());

        // Wait for either task to complete or stop signal
        tokio::select! {
            _ = stop_signal.changed() => Ok(()),
            res = listen_handle => res.map_err(|e| TransportError::Listen(e.to_string()))?,
            res = process_handle => res.map_err(|e| TransportError::MessageProcessing(e.to_string()))?,
        }
    }

    async fn connect_outbound(&self, pubkey: PublicKey, host: &str) {
        connect_outbound(self.peer_manager.clone(), pubkey, host.parse().unwrap()).await;
    }
}