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
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137

// Copyright 2015-2020 Parity Technologies (UK) Ltd.
// This file is part of OpenEthereum.

// OpenEthereum is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.

// OpenEthereum 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 General Public License for more details.

// You should have received a copy of the GNU General Public License
// along with OpenEthereum.  If not, see <http://www.gnu.org/licenses/>.

//! A map of subscribers.

use ethereum_types::H64;
use jsonrpc_pubsub::{
    typed::{Sink, Subscriber},
    SubscriptionId,
};
use std::{collections::HashMap, ops, str};

#[derive(Debug, Clone, Hash, Eq, PartialEq)]
pub struct Id(H64);
impl str::FromStr for Id {
    type Err = String;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        if s.starts_with("0x") {
            Ok(Id(s[2..].parse().map_err(|e| format!("{}", e))?))
        } else {
            Err("The id must start with 0x".into())
        }
    }
}
impl Id {
    // TODO: replace `format!` see [#10412](https://github.com/openethereum/openethereum/issues/10412)
    pub fn as_string(&self) -> String {
        format!("{:?}", self.0)
    }
}

#[cfg(not(test))]
mod random {
    use rand::rngs::OsRng;
    pub type Rng = rand::rngs::OsRng;
    pub fn new() -> Rng {
        OsRng
    }
}

#[cfg(test)]
mod random {
    extern crate rand_xorshift;
    use self::rand_xorshift::XorShiftRng;
    use rand::SeedableRng;
    const RNG_SEED: [u8; 16] = [0u8; 16];
    pub type Rng = XorShiftRng;
    pub fn new() -> Rng {
        Rng::from_seed(RNG_SEED)
    }
}

pub struct Subscribers<T> {
    rand: random::Rng,
    subscriptions: HashMap<Id, T>,
}

impl<T> Default for Subscribers<T> {
    fn default() -> Self {
        Subscribers {
            rand: random::new(),
            subscriptions: HashMap::new(),
        }
    }
}

impl<T> Subscribers<T> {
    fn next_id(&mut self) -> Id {
        let data = H64::random_using(&mut self.rand);
        Id(data)
    }

    /// Insert new subscription and return assigned id.
    pub fn insert(&mut self, val: T) -> SubscriptionId {
        let id = self.next_id();
        debug!(target: "pubsub", "Adding subscription id={:?}", id);
        let s = id.as_string();
        self.subscriptions.insert(id, val);
        SubscriptionId::String(s)
    }

    /// Removes subscription with given id and returns it (if any).
    pub fn remove(&mut self, id: &SubscriptionId) -> Option<T> {
        trace!(target: "pubsub", "Removing subscription id={:?}", id);
        match *id {
            SubscriptionId::String(ref id) => match id.parse() {
                Ok(id) => self.subscriptions.remove(&id),
                Err(_) => None,
            },
            _ => None,
        }
    }
}

impl<T> Subscribers<Sink<T>> {
    /// Assigns id and adds a subscriber to the list.
    pub fn push(&mut self, sub: Subscriber<T>) {
        let id = self.next_id();
        if let Ok(sink) = sub.assign_id(SubscriptionId::String(id.as_string())) {
            debug!(target: "pubsub", "Adding subscription id={:?}", id);
            self.subscriptions.insert(id, sink);
        }
    }
}

impl<T, V> Subscribers<(Sink<T>, V)> {
    /// Assigns id and adds a subscriber to the list.
    pub fn push(&mut self, sub: Subscriber<T>, val: V) {
        let id = self.next_id();
        if let Ok(sink) = sub.assign_id(SubscriptionId::String(id.as_string())) {
            debug!(target: "pubsub", "Adding subscription id={:?}", id);
            self.subscriptions.insert(id, (sink, val));
        }
    }
}

impl<T> ops::Deref for Subscribers<T> {
    type Target = HashMap<Id, T>;

    fn deref(&self) -> &Self::Target {
        &self.subscriptions
    }
}