tentacle 0.7.5

Minimal implementation for a multiplexed p2p network framework.
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
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93

use tentacle::{
    ProtocolId, async_trait,
    builder::{MetaBuilder, ServiceBuilder},
    context::{ProtocolContext, ServiceContext},
    secio::SecioKeyPair,
    service::{ProtocolHandle, ProtocolMeta, Service, ServiceError, ServiceEvent},
    traits::{ServiceHandle, ServiceProtocol},
};

use futures::{
    StreamExt,
    channel::mpsc::{Receiver, channel},
    prelude::*,
};
use log::info;
use std::time::Duration;

/// This example is used to illustrate how to implement the poll interface
/// and use it for inner loop notification
fn main() {
    env_logger::init();
    run()
}

struct SHandle;

#[async_trait]
impl ServiceHandle for SHandle {
    async fn handle_error(&mut self, _context: &mut ServiceContext, error: ServiceError) {
        info!("service error: {:?}", error);
    }
    async fn handle_event(&mut self, _context: &mut ServiceContext, event: ServiceEvent) {
        info!("service event: {:?}", event);
    }
}

struct PHandle {
    poll_recv: Receiver<()>,
}

#[async_trait]
impl ServiceProtocol for PHandle {
    async fn init(&mut self, _context: &mut ProtocolContext) {}

    async fn poll(&mut self, _context: &mut ProtocolContext) -> Option<()> {
        match self.poll_recv.next().await {
            Some(_) => {
                info!("get a trick");
                Some(())
            }
            None => None,
        }
    }
}

fn create_meta(id: ProtocolId, recv: Receiver<()>) -> ProtocolMeta {
    MetaBuilder::new()
        .id(id)
        .service_handle(move || {
            // All protocol use the same handle.
            // This is just an example. In the actual environment, this should be a different handle.
            let handle = Box::new(PHandle { poll_recv: recv });
            ProtocolHandle::Callback(handle)
        })
        .build()
}

fn create_server(recv: Receiver<()>) -> Service<SHandle, SecioKeyPair> {
    ServiceBuilder::default()
        .insert_protocol(create_meta(0.into(), recv))
        .handshake_type(SecioKeyPair::secp256k1_generated().into())
        .build(SHandle)
}

fn run() {
    let rt = tokio::runtime::Runtime::new().unwrap();

    rt.block_on(async {
        let (mut tx, rx) = channel(2);
        let mut service = create_server(rx);
        service
            .listen("/dns4/localhost/tcp/1337".parse().unwrap())
            .await
            .unwrap();
        tokio::spawn(async move {
            loop {
                tokio::time::sleep(Duration::from_secs(1)).await;
                tx.send(()).await.unwrap();
            }
        });
        service.run().await
    });
}