signer-daemon 0.2.1

Signer daemon package.
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
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
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115

use std::sync::Arc;

use crate::{model::crdt::crdt::CrdtDeltaBox, swallow_error::SwallowError};

use super::SignerDaemon;

#[derive(Debug, Clone)]
#[allow(dead_code)]
pub(crate) enum SignerDaemonEvent {
  InsertDelta(CrdtDeltaBox),
  SyncEvent(String),
  SyncEventAll,
  SyncUser,
  ApplyEvent,
  UpdateConnections,
  ForceReapplyAllEvent,
}

impl SignerDaemon {
  pub(crate) fn emit(&self, event: SignerDaemonEvent) -> anyhow::Result<()> {
    self.event_sender.send(event)?;
    Ok(())
  }

  pub async fn wait_apply(&self) -> anyhow::Result<()> {
    let mut rx = self.event_receiver.subscribe();
    loop {
      match rx.recv().await? {
        SignerDaemonEvent::ApplyEvent => break,
        _ => continue,
      }
    }

    Ok(())
  }

  pub(crate) fn start_watch_event(&self) {
    let daemon = Arc::new(self.clone());
    let mut rx = self.event_sender.subscribe();

    tokio::spawn(async move {
      loop {
        let ev = match rx.recv().await {
          Ok(ev) => ev,
          Err(_) => break,
        };

        match &ev {
          SignerDaemonEvent::InsertDelta(delta) => {
            delta
              .insert(&daemon)
              .await
              .swallow_error("insert delta failed");
            daemon
              .emit(SignerDaemonEvent::ApplyEvent)
              .expect("emit apply event failed");
            daemon
              .emit(SignerDaemonEvent::SyncEventAll)
              .expect("emit sync event all failed");
          }
          SignerDaemonEvent::SyncEvent(addr) => {
            daemon
              .crdt_pull(&addr)
              .await
              .swallow_error("crdt pull failed");
            daemon
              .crdt_push(&addr)
              .await
              .swallow_error("crdt push failed");
            daemon
              .emit(SignerDaemonEvent::ApplyEvent)
              .expect("emit apply event failed");
          }
          SignerDaemonEvent::SyncEventAll => {
            for (addr, conn) in daemon.connections.lock().await.iter() {
              conn.sync_crdt_event().await.swallow_error(&format!(
                "sync event all failed: sync crdt event from {} failed",
                addr
              ));
            }
          }
          SignerDaemonEvent::SyncUser => {
            daemon
              .crdt_sync_user_public()
              .await
              .swallow_error("crdt sync user public failed");
          }
          SignerDaemonEvent::ApplyEvent => {
            daemon
              .crdt_apply_all()
              .await
              .swallow_error("crdt apply all failed");
            daemon
              .emit(SignerDaemonEvent::SyncUser)
              .expect("emit sync user failed");
          }
          SignerDaemonEvent::UpdateConnections => {
            daemon
              .update_connections()
              .await
              .swallow_error("update connections failed");
          }
          SignerDaemonEvent::ForceReapplyAllEvent => {
            daemon
              .crdt_apply_all()
              .await
              .swallow_error("crdt apply all failed");
          }
        };

        let _ = daemon.event_receiver.send(ev);
      }
    });
  }
}