Add priority broadcast type and insert handler

Author: spacekookie

crates/corro-agent/src/agent/handlers.rs

@@ -108,7 +108,7 @@ pub fn spawn_incoming_connection_handlers(
 
         // Spawn handler tasks for this connection
         spawn_foca_handler(&agent, &tripwire, &conn);
-        uni::spawn_unipayload_handler(&tripwire, &conn, agent.tx_changes().clone());
+        uni::spawn_unipayload_handler(&agent, &bookie, &tripwire, &conn, agent.tx_changes().clone());
         bi::spawn_bipayload_handler(&agent, &bookie, &tripwire, &conn);
     });
 }

crates/corro-agent/src/agent/uni.rs

@@ -1,4 +1,5 @@
 use corro_types::{
+    agent::{Agent, Bookie},
     broadcast::{BroadcastV1, ChangeSource, ChangeV1, UniPayload, UniPayloadV1},
     channel::CorroSender,
 };
@@ -12,15 +13,22 @@ use tripwire::Tripwire;
 /// Spawn a task that accepts unidirectional broadcast streams, then
 /// spawns another task for each incoming stream to handle.
 pub fn spawn_unipayload_handler(
+    agent: &Agent,
+    bookie: &Bookie,
     tripwire: &Tripwire,
     conn: &quinn::Connection,
     changes_tx: CorroSender<(ChangeV1, ChangeSource)>,
 ) {
+    let agent = agent.clone();
+    let bookie = bookie.clone();
     tokio::spawn({
         let conn = conn.clone();
         let mut tripwire = tripwire.clone();
+
         async move {
             loop {
+                let agent = agent.clone();
+                let bookie = bookie.clone();
                 let rx = tokio::select! {
                     rx_res = conn.accept_uni() => match rx_res {
                         Ok(rx) => rx,
@@ -48,6 +56,9 @@ pub fn spawn_unipayload_handler(
                         let mut framed = FramedRead::new(rx, LengthDelimitedCodec::new());
 
                         loop {
+                            let agent = agent.clone();
+                            let bookie = bookie.clone();
+
                             match StreamExt::next(&mut framed).await {
                                 Some(Ok(b)) => {
                                     counter!("corro.peer.stream.bytes.recv.total", "type" => "uni")
@@ -70,6 +81,23 @@ pub fn spawn_unipayload_handler(
                                                         return;
                                                     }
                                                 }
+                                                UniPayload::V1(
+                                                    UniPayloadV1::PriorityBroadcast(
+                                                        BroadcastV1::Change(change),
+                                                    ),
+                                                ) => {
+                                                    // fixme: queue with priority?
+                                                    tokio::spawn(async move {
+                                                        crate::change::process_priority_change(
+                                                            agent,
+                                                            bookie,
+                                                            change,
+                                                            ChangeSource::PriorityBroadcast,
+                                                        )
+                                                        .await
+                                                        .unwrap();
+                                                    });
+                                                }
                                             }
                                         }
                                         Err(e) => {

crates/corro-agent/src/api/peer.rs

@@ -1549,10 +1549,7 @@ mod tests {
     use tokio::sync::mpsc;
     use tripwire::Tripwire;
 
-    use crate::{
-        agent::{process_multiple_changes, setup},
-        api::public::api_v1_db_schema,
-    };
+    use crate::{agent::setup, api::public::api_v1_db_schema, change::process_multiple_changes};
 
     use super::*;