store: Asyncify notification_listener

Author: lutter

store/postgres/src/chain_head_listener.rs

@@ -254,19 +254,21 @@ impl ChainHeadUpdateSender {
         }
     }
 
-    pub fn send(&self, hash: &str, number: i64) -> Result<(), StoreError> {
+    pub async fn send(&self, hash: &str, number: i64) -> Result<(), StoreError> {
         let msg = json! ({
             "network_name": &self.chain_name,
             "head_block_hash": hash,
             "head_block_number": number
         });
 
         let mut conn = self.pool.get()?;
-        self.sender.notify(
-            &mut conn,
-            CHANNEL_NAME.as_str(),
-            Some(&self.chain_name),
-            &msg,
-        )
+        self.sender
+            .notify(
+                &mut conn,
+                CHANNEL_NAME.as_str(),
+                Some(&self.chain_name),
+                &msg,
+            )
+            .await
     }
 }

store/postgres/src/chain_store.rs

@@ -2276,7 +2276,7 @@ impl ChainHeadStore for ChainStore {
         let number = ptr.number as i64; //block height
 
         //this will send an update via postgres, channel: chain_head_updates
-        self.chain_head_update_sender.send(&hash, number)?;
+        self.chain_head_update_sender.send(&hash, number).await?;
 
         pool.with_conn(async move |conn, _| {
             conn.transaction_async(|conn| {
@@ -2412,7 +2412,7 @@ impl ChainStoreTrait for ChainStore {
                 .await?
             };
         if let Some((hash, number)) = ptr {
-            self.chain_head_update_sender.send(&hash, number)?;
+            self.chain_head_update_sender.send(&hash, number).await?;
         }
 
         Ok(missing)

store/postgres/src/notification_listener.rs

@@ -404,7 +404,7 @@ impl NotificationSender {
     /// connection `conn` must be into the primary database as that's the
     /// only place where listeners connect. The `network` is only used for
     /// metrics gathering and does not affect how the notification is sent
-    pub fn notify(
+    pub async fn notify(
         &self,
         conn: &mut PgConnection,
         channel: &str,

store/postgres/src/primary.rs

@@ -1539,7 +1539,9 @@ impl Connection {
                 EVENT_TAP.lock().unwrap().push(event.clone());
             }
         }
-        sender.notify(&mut self.conn, "store_events", None, &v)
+        sender
+            .notify(&mut self.conn, "store_events", None, &v)
+            .await
     }
 
     /// Return the name of the node that has the fewest assignments out of the