Merge pull request #61 from EspressoSystems/rm/hook-identifiers

Message hook identifiers

cdn-broker/src/tasks/broker/handler.rs

@@ -15,7 +15,7 @@ use cdn_proto::{
     discovery::BrokerIdentifier,
     error::{Error, Result},
     message::{Message, Topic},
-    util::mnemonic,
+    util::{hash, mnemonic},
     verify_broker,
 };
 use tokio::{spawn, time::timeout};
@@ -123,8 +123,9 @@ impl<Def: RunDef> Inner<Def> {
         broker_identifier: &BrokerIdentifier,
         connection: Connection,
     ) -> Result<()> {
-        // Clone the hook
+        // Clone the hook and set its identifier
         let mut local_message_hook = self.broker_message_hook.clone();
+        local_message_hook.set_identifier(hash(broker_identifier));
 
         loop {
             // Receive a message from the broker

cdn-broker/src/tasks/user/handler.rs

@@ -13,7 +13,7 @@ use std::time::Duration;
 use cdn_proto::connection::{protocols::Connection, UserPublicKey};
 use cdn_proto::def::{HookResult, MessageHookDef, RunDef, Topic as _};
 use cdn_proto::error::{Error, Result};
-use cdn_proto::util::mnemonic;
+use cdn_proto::util::{hash, mnemonic};
 use cdn_proto::{connection::auth::broker::BrokerAuth, message::Message};
 use tokio::spawn;
 use tokio::time::timeout;
@@ -99,6 +99,7 @@ impl<Def: RunDef> Inner<Def> {
     ) -> Result<()> {
         // Clone the hook
         let mut local_message_hook = self.user_message_hook.clone();
+        local_message_hook.set_identifier(hash(public_key));
 
         loop {
             // Receive a message from the user

cdn-proto/src/def.rs

@@ -85,6 +85,10 @@ pub trait MessageHookDef: Send + Sync + 'static + Clone {
     fn on_message_received(&mut self, _message: &mut Message) -> AnyhowResult<HookResult> {
         Ok(HookResult::ProcessMessage)
     }
+
+    /// Set a unique identifier for the hook. This can be included with the hook and can be
+    /// used to deterministically identify message producers.
+    fn set_identifier(&mut self, _identifier: u64) {}
 }
 
 /// The no-op hook

cdn-proto/src/util.rs

@@ -7,12 +7,18 @@ use std::{
 
 use tokio::task::{JoinError, JoinHandle};
 
-/// A function for generating a cute little user mnemonic from a hash
+/// A function for generating a cute little user mnemonic from a hashable value
 #[must_use]
 pub fn mnemonic<H: Hash>(bytes: H) -> String {
+    mnemonic::to_string(hash(bytes).to_le_bytes())
+}
+
+/// A helper function for generating a 64-bit hash from a value
+#[must_use]
+pub fn hash<H: Hash>(bytes: H) -> u64 {
     let mut state = std::collections::hash_map::DefaultHasher::new();
     bytes.hash(&mut state);
-    mnemonic::to_string(state.finish().to_le_bytes())
+    state.finish()
 }
 
 /// A wrapper for a `JoinHandle` that will abort the task if dropped