feat: Migrate to dumb agents (#198)

BREAKING CHANGE: handshake, chainsync, localstate and blockfetch mini-protocols changed the API surface

Co-authored-by: jmhrpr <harper.jme@gmail.com>

examples/n2n-miniprotocols/src/main.rs

@@ -1,61 +1,29 @@
 use pallas::network::{
-    miniprotocols::{blockfetch, chainsync, handshake, run_agent, Point, MAINNET_MAGIC},
-    multiplexer::{agents::ChannelBuffer, bearers::Bearer, StdChannel, StdPlexer},
+    miniprotocols::{blockfetch, chainsync, handshake, Point, MAINNET_MAGIC},
+    multiplexer::{bearers::Bearer, StdChannel, StdPlexer},
 };
 
 #[derive(Debug)]
 struct LoggingObserver;
 
-impl blockfetch::Observer for LoggingObserver {
-    fn on_block_received(&mut self, body: Vec<u8>) -> Result<(), Box<dyn std::error::Error>> {
-        log::trace!("block received: {}", hex::encode(&body));
-        Ok(())
+fn do_handshake(channel: StdChannel) {
+    let mut client = handshake::N2NClient::new(channel);
+
+    let confirmation = client
+        .handshake(handshake::n2n::VersionTable::v7_and_above(MAINNET_MAGIC))
+        .unwrap();
+
+    match confirmation {
+        handshake::Confirmation::Accepted(v, _) => {
+            log::info!("hand-shake accepted, using version {}", v)
+        }
+        handshake::Confirmation::Rejected(x) => {
+            log::info!("hand-shake rejected with reason {:?}", x)
+        }
     }
 }
 
-impl chainsync::Observer<chainsync::HeaderContent> for LoggingObserver {
-    fn on_roll_forward(
-        &mut self,
-        _content: chainsync::HeaderContent,
-        tip: &chainsync::Tip,
-    ) -> Result<chainsync::Continuation, Box<dyn std::error::Error>> {
-        log::info!("asked to roll forward, tip at {:?}", tip);
-
-        Ok(chainsync::Continuation::Proceed)
-    }
-
-    fn on_intersect_found(
-        &mut self,
-        point: &Point,
-        tip: &chainsync::Tip,
-    ) -> Result<chainsync::Continuation, Box<dyn std::error::Error>> {
-        log::debug!("intersect was found {:?} (tip: {:?})", point, tip);
-
-        Ok(chainsync::Continuation::Proceed)
-    }
-
-    fn on_rollback(
-        &mut self,
-        point: &Point,
-    ) -> Result<chainsync::Continuation, Box<dyn std::error::Error>> {
-        log::debug!("asked to roll back {:?}", point);
-
-        Ok(chainsync::Continuation::Proceed)
-    }
-
-    fn on_tip_reached(&mut self) -> Result<chainsync::Continuation, Box<dyn std::error::Error>> {
-        log::debug!("tip was reached");
-
-        Ok(chainsync::Continuation::Proceed)
-    }
-}
-
-fn do_handshake(mut channel: ChannelBuffer<StdChannel>) {
-    let versions = handshake::n2n::VersionTable::v4_and_above(MAINNET_MAGIC);
-    let _last = run_agent(handshake::Initiator::initial(versions), &mut channel).unwrap();
-}
-
-fn do_blockfetch(mut channel: ChannelBuffer<StdChannel>) {
+fn do_blockfetch(channel: StdChannel) {
     let range = (
         Point::Specific(
             43847831,
@@ -69,29 +37,38 @@ fn do_blockfetch(mut channel: ChannelBuffer<StdChannel>) {
         ),
     );
 
-    let agent = run_agent(
-        blockfetch::BatchClient::initial(range, LoggingObserver {}),
-        &mut channel,
-    );
+    let mut client = blockfetch::Client::new(channel);
 
-    println!("{:?}", agent);
+    let blocks = client.fetch_range(range).unwrap();
+
+    for block in blocks {
+        log::info!("received block of size: {}", block.len());
+    }
 }
 
-fn do_chainsync(mut channel: ChannelBuffer<StdChannel>) {
+fn do_chainsync(channel: StdChannel) {
     let known_points = vec![Point::Specific(
         43847831u64,
         hex::decode("15b9eeee849dd6386d3770b0745e0450190f7560e5159b1b3ab13b14b2684a45").unwrap(),
     )];
 
-    let agent = run_agent(
-        chainsync::Consumer::<chainsync::HeaderContent, _>::initial(
-            Some(known_points),
-            LoggingObserver {},
-        ),
-        &mut channel,
-    );
+    let mut client = chainsync::N2NClient::new(channel);
 
-    println!("{:?}", agent);
+    let (point, _) = client.find_intersect(known_points).unwrap();
+
+    log::info!("intersected point is {:?}", point);
+
+    for _ in 0..10 {
+        let next = client.request_next().unwrap();
+
+        match next {
+            chainsync::NextResponse::RollForward(h, _) => {
+                log::info!("rolling forward, header size: {}", h.cbor.len())
+            }
+            chainsync::NextResponse::RollBackward(x, _) => log::info!("rollback to {:?}", x),
+            chainsync::NextResponse::Await => log::info!("tip of chaing reached"),
+        };
+    }
 }
 
 fn main() {
@@ -106,9 +83,9 @@ fn main() {
     // setup the multiplexer by specifying the bearer and the IDs of the
     // miniprotocols to use
     let mut plexer = StdPlexer::new(bearer);
-    let channel0 = plexer.use_channel(0).into();
-    let channel3 = plexer.use_channel(3).into();
-    let channel2 = plexer.use_channel(2).into();
+    let channel0 = plexer.use_channel(0);
+    let channel3 = plexer.use_channel(3);
+    let channel2 = plexer.use_channel(2);
 
     plexer.muxer.spawn();
     plexer.demuxer.spawn();