- fix: fixed bug 'already borrowed'

- fix: fixed sequencing bugs, now single/multi branch examples work fine - channels: pk_store sequencing info now contains only MsgId instead of full TangleAddress - example: added pure channels example using BucketTransport to run without node - example: added extensive log to track sequencing info - impl: added/improved Debug/Display instances for some structs, mainly using hex::encode - todo: added TODO comments for HDF fields to fix types and wrap/unwrap operations - cargo fmt
iotaledger · Sep 18, 2020 · 7409f8e · 7409f8e
1 parent 1e997b2
commit 7409f8e
Show file tree

Hide file tree

Showing 11 changed files with 329 additions and 190 deletions.
diff --git a/examples/src/branching/multi_branch.rs b/examples/src/branching/multi_branch.rs
diff --git a/examples/src/branching/single_branch.rs b/examples/src/branching/single_branch.rs
@@ -5,22 +5,20 @@ use iota_streams::{
     },
     app_channels::{
         api::tangle::{
-            Address,
             Author,
             Subscriber,
             Transport,
         },
         message,
     },
     core::{
-        prelude::Vec,
+        print,
         println,
     },
     ddml::types::*,
 };
 
 use anyhow::{
-    anyhow,
     ensure,
     Result,
 };
@@ -41,7 +39,7 @@ where
     let multi_branching_flag = 0_u8;
     let encoding = "utf-8";
     let mut author = Author::new(seed, encoding, PAYLOAD_BYTES, multi_branching_flag == 1_u8);
-    println!("Author multi branching?: {}", author.is_multi_branching());
+    println!("  Author multi branching?: {}", author.is_multi_branching());
 
     let mut subscriberA = Subscriber::new("SUBSCRIBERA9SEED", encoding, PAYLOAD_BYTES);
     let mut subscriberB = Subscriber::new("SUBSCRIBERB9SEED", encoding, PAYLOAD_BYTES);
@@ -51,22 +49,16 @@ where
     let masked_payload = Bytes("MASKEDPAYLOAD".as_bytes().to_vec());
 
     println!("\nAnnounce Channel");
-    let (announcement_address, announcement_tag) = {
-        let msg = &author.announce()?;
+    let announcement_link = {
+        let msg = author.announce()?;
+        println!("  msg => <{}> {:?}", msg.link.msgid, msg);
+        print!("  Author     : {}", author);
         transport.send_message_with_options(&msg, send_opt)?;
-        (msg.link.appinst.as_ref().to_vec(), msg.link.msgid.as_ref().to_vec())
+        msg.link
     };
-    println!("    Author channel address: {:?}", author.channel_address());
-
-    let mut v1 = Vec::<u8>::new();
-    v1.extend_from_slice(&announcement_address);
-
-    let mut v2 = Vec::<u8>::new();
-    v2.extend_from_slice(&announcement_tag);
+    println!("  Author channel address: {}", author.channel_address().unwrap());
 
-    let announcement_link = Address::from_str(&hex::encode(announcement_address), &hex::encode(announcement_tag))
-        .map_err(|_| anyhow!("bad address"))?;
-    println!("    Announcement link at: {}", &announcement_link);
+    println!("\nHandle Announce Channel");
     {
         let msg = transport.recv_message_with_options(&announcement_link, recv_opt)?;
         let preparsed = msg.parse_header()?;
@@ -76,18 +68,19 @@ where
         );
 
         subscriberA.unwrap_announcement(preparsed.clone())?;
-        println!("    SubscriberA channel address: {:?}", subscriberA.channel_address());
+        print!("  SubscriberA: {}", subscriberA);
         ensure!(
             (author.channel_address() == subscriberA.channel_address()),
             "SubscriberA channel address does not match Author channel address"
         );
         subscriberB.unwrap_announcement(preparsed.clone())?;
-        println!("    SubscriberB channel address: {:?}", subscriberB.channel_address());
+        print!("  SubscriberB: {}", subscriberB);
         ensure!(
             subscriberA.channel_address() == subscriberB.channel_address(),
             "SubscriberB channel address does not match Author channel address"
         );
         subscriberC.unwrap_announcement(preparsed)?;
+        print!("  SubscriberC: {}", subscriberC);
         ensure!(
             subscriberA.channel_address() == subscriberC.channel_address(),
             "SubscriberC channel address does not match Author channel address"
@@ -108,11 +101,13 @@ where
     println!("\nSubscribe A");
     let subscribeA_link = {
         let msg = subscriberA.subscribe(&announcement_link)?;
+        println!("  msg => <{}> {:?}", msg.link.msgid, msg);
+        print!("  SubscriberA: {}", subscriberA);
         transport.send_message_with_options(&msg, send_opt)?;
-        println!("    Subscribe at {}", msg.link.msgid);
         msg.link
     };
 
+    println!("\nHandle Subscribe A");
     {
         let msg = transport.recv_message_with_options(&subscribeA_link, recv_opt)?;
         let preparsed = msg.parse_header()?;
@@ -122,16 +117,19 @@ where
             preparsed.header.content_type
         );
         author.unwrap_subscribe(preparsed)?;
+        print!("  Author     : {}", author);
     }
 
     println!("\nSubscribe B");
     let subscribeB_link = {
         let msg = subscriberB.subscribe(&announcement_link)?;
+        println!("  msg => <{}> {:?}", msg.link.msgid, msg);
+        print!("  SubscriberB: {}", subscriberB);
         transport.send_message_with_options(&msg, send_opt)?;
-        println!("    Subscribe at {}", msg.link.msgid);
         msg.link
     };
 
+    println!("\nHandle Subscribe B");
     {
         let msg = transport.recv_message_with_options(&subscribeB_link, recv_opt)?;
         let preparsed = msg.parse_header()?;
@@ -141,16 +139,20 @@ where
             preparsed.header.content_type
         );
         author.unwrap_subscribe(preparsed)?;
+        print!("  Author     : {}", author);
     }
 
     println!("\nShare keyload for everyone [SubscriberA, SubscriberB]");
     let previous_msg_link = {
-        let msg = author.share_keyload_for_everyone(&announcement_link)?;
-        transport.send_message_with_options(&msg.0, send_opt)?;
-        println!("Keyload message at {}", &msg.0.link.msgid);
-        msg.0.link
+        let (msg, seq) = author.share_keyload_for_everyone(&announcement_link)?;
+        println!("  msg => <{}> {:?}", msg.link.msgid, msg);
+        assert!(seq.is_none());
+        print!("  Author     : {}", author);
+        transport.send_message_with_options(&msg, send_opt)?;
+        msg.link
     };
 
+    println!("\nHandle Keyload");
     {
         let msg = transport.recv_message_with_options(&previous_msg_link, recv_opt)?;
         let preparsed = msg.parse_header()?;
@@ -161,20 +163,27 @@ where
         );
 
         let resultC = subscriberC.unwrap_keyload(preparsed.clone());
+        print!("  SubscriberC: {}", subscriberC);
         ensure!(resultC.is_err(), "SubscriberC should not be able to unwrap the keyload");
 
         subscriberA.unwrap_keyload(preparsed.clone())?;
-        subscriberB.unwrap_keyload(preparsed)?
+        print!("  SubscriberA: {}", subscriberA);
+        subscriberB.unwrap_keyload(preparsed)?;
+        print!("  SubscriberB: {}", subscriberB);
     }
 
     println!("\nSigned packet");
     let previous_msg_link = {
-        let msg = author.sign_packet(&previous_msg_link, &public_payload, &masked_payload)?;
-        transport.send_message_with_options(&msg.0, send_opt)?;
-        println!("Signed packet at {}", &msg.0.link.msgid);
-        msg.0.link
+        print!("  Author     : {}", author);
+        let (msg, seq) = author.sign_packet(&previous_msg_link, &public_payload, &masked_payload)?;
+        println!("  msg => <{}> {:?}", msg.link.msgid, msg);
+        assert!(seq.is_none());
+        print!("  Author     : {}", author);
+        transport.send_message_with_options(&msg, send_opt)?;
+        msg.link
     };
 
+    println!("\nHandle Signed packet");
     {
         let msg = transport.recv_message_with_options(&previous_msg_link, recv_opt)?;
         let preparsed = msg.parse_header()?;
@@ -185,6 +194,7 @@ where
         );
 
         let (_signer_pk, unwrapped_public, unwrapped_masked) = subscriberA.unwrap_signed_packet(preparsed)?;
+        print!("  SubscriberA: {}", subscriberA);
         ensure!(public_payload == unwrapped_public, "Public payloads do not match");
         ensure!(masked_payload == unwrapped_masked, "Masked payloads do not match");
     }
@@ -194,12 +204,15 @@ where
 
     println!("\nTagged packet 1 - SubscriberA");
     let previous_msg_link = {
-        let msg = subscriberA.tag_packet(&previous_msg_link, &public_payload, &masked_payload)?;
-        transport.send_message_with_options(&msg.0, send_opt)?;
-        println!("Tagged packet at {}", &msg.0.link.msgid);
-        msg.0.link
+        let (msg, seq) = subscriberA.tag_packet(&previous_msg_link, &public_payload, &masked_payload)?;
+        println!("  msg => <{}> {:?}", msg.link.msgid, msg);
+        assert!(seq.is_none());
+        print!("  SubscriberA: {}", subscriberA);
+        transport.send_message_with_options(&msg, send_opt)?;
+        msg.link
     };
 
+    println!("\nHandle Tagged packet 1");
     {
         let msg = transport.recv_message_with_options(&previous_msg_link, recv_opt)?;
         let preparsed = msg.parse_header()?;
@@ -210,10 +223,12 @@ where
         );
 
         let (unwrapped_public, unwrapped_masked) = author.unwrap_tagged_packet(preparsed.clone())?;
+        print!("  Author     : {}", author);
         ensure!(public_payload == unwrapped_public, "Public payloads do not match");
         ensure!(masked_payload == unwrapped_masked, "Masked payloads do not match");
 
         let resultC = subscriberC.unwrap_tagged_packet(preparsed);
+        print!("  SubscriberC: {}", subscriberC);
         ensure!(
             resultC.is_err(),
             "Subscriber C should not be able to access this message"
@@ -222,12 +237,15 @@ where
 
     println!("\nTagged packet 2 - SubscriberA");
     let previous_msg_link = {
-        let msg = subscriberA.tag_packet(&previous_msg_link, &public_payload, &masked_payload)?;
-        transport.send_message_with_options(&msg.0, send_opt)?;
-        println!("Tagged packet at {}", &msg.0.link.msgid);
-        msg.0.link
+        let (msg, seq) = subscriberA.tag_packet(&previous_msg_link, &public_payload, &masked_payload)?;
+        println!("  msg => <{}> {:?}", msg.link.msgid, msg);
+        assert!(seq.is_none());
+        print!("  SubscriberA: {}", subscriberA);
+        transport.send_message_with_options(&msg, send_opt)?;
+        msg.link
     };
 
+    println!("\nHandle Tagged packet 2");
     {
         let msg = transport.recv_message_with_options(&previous_msg_link, recv_opt)?;
         let preparsed = msg.parse_header()?;
@@ -240,12 +258,15 @@ where
 
     println!("\nTagged packet 3 - SubscriberA");
     let previous_msg_link = {
-        let msg = subscriberA.tag_packet(&previous_msg_link, &public_payload, &masked_payload)?;
-        transport.send_message_with_options(&msg.0, send_opt)?;
-        println!("Tagged packet at {}", &msg.0.link.msgid);
-        msg.0.link
+        let (msg, seq) = subscriberA.tag_packet(&previous_msg_link, &public_payload, &masked_payload)?;
+        println!("  msg => <{}> {:?}", msg.link.msgid, msg);
+        assert!(seq.is_none());
+        print!("  SubscriberA: {}", subscriberA);
+        transport.send_message_with_options(&msg, send_opt)?;
+        msg.link
     };
 
+    println!("\nHandle Tagged packet 3");
     {
         let msg = transport.recv_message_with_options(&previous_msg_link, recv_opt)?;
         let preparsed = msg.parse_header()?;
@@ -261,12 +282,15 @@ where
 
     println!("\nTagged packet 4 - SubscriberB");
     let previous_msg_link = {
-        let msg = subscriberB.tag_packet(&previous_msg_link, &public_payload, &masked_payload)?;
-        transport.send_message_with_options(&msg.0, send_opt)?;
-        println!("Tagged packet at {}", &msg.0.link.msgid);
-        msg.0.link
+        let (msg, seq) = subscriberB.tag_packet(&previous_msg_link, &public_payload, &masked_payload)?;
+        println!("  msg => <{}> {:?}", msg.link.msgid, msg);
+        assert!(seq.is_none());
+        print!("  SubscriberB: {}", subscriberB);
+        transport.send_message_with_options(&msg, send_opt)?;
+        msg.link
     };
 
+    println!("\nHandle Tagged packet 4");
     {
         let msg = transport.recv_message_with_options(&previous_msg_link, recv_opt)?;
         let preparsed = msg.parse_header()?;
@@ -277,10 +301,12 @@ where
         );
 
         let (unwrapped_public, unwrapped_masked) = subscriberA.unwrap_tagged_packet(preparsed.clone())?;
+        print!("  SubscriberA: {}", subscriberA);
         ensure!(public_payload == unwrapped_public, "Public payloads do not match");
         ensure!(masked_payload == unwrapped_masked, "Masked payloads do not match");
 
         let resultC = subscriberC.unwrap_tagged_packet(preparsed);
+        print!("  SubscriberC: {}", subscriberC);
         ensure!(
             resultC.is_err(),
             "Subscriber C should not be able to access this message"
@@ -292,12 +318,15 @@ where
 
     println!("\nSigned packet");
     let previous_msg_link = {
-        let msg = author.sign_packet(&previous_msg_link, &public_payload, &masked_payload)?;
-        transport.send_message_with_options(&msg.0, send_opt)?;
-        println!("Signed packet at {}", &msg.0.link.msgid);
-        msg.0.link
+        let (msg, seq) = author.sign_packet(&previous_msg_link, &public_payload, &masked_payload)?;
+        println!("  msg => <{}> {:?}", msg.link.msgid, msg);
+        assert!(seq.is_none());
+        print!("  Author     : {}", author);
+        transport.send_message_with_options(&msg, send_opt)?;
+        msg.link
     };
 
+    println!("\nHandle Signed packet");
     {
         let msg = transport.recv_message_with_options(&previous_msg_link, recv_opt)?;
         let preparsed = msg.parse_header()?;
@@ -308,10 +337,12 @@ where
         );
 
         let (_signer_pk, unwrapped_public, unwrapped_masked) = subscriberA.unwrap_signed_packet(preparsed.clone())?;
+        print!("  SubscriberA: {}", subscriberA);
         ensure!(public_payload == unwrapped_public, "Public payloads do not match");
         ensure!(masked_payload == unwrapped_masked, "Masked payloads do not match");
 
         let (_signer_pk, unwrapped_public, unwrapped_masked) = subscriberB.unwrap_signed_packet(preparsed)?;
+        print!("  SubscriberB: {}", subscriberB);
         ensure!(public_payload == unwrapped_public, "Public payloads do not match");
         ensure!(masked_payload == unwrapped_masked, "Masked payloads do not match");
     }

diff --git a/examples/src/branching/utils.rs b/examples/src/branching/utils.rs
@@ -1,10 +1,13 @@
 use iota_streams::{
     app::message::HasLink as _,
     app_channels::{
-        api::tangle::{
-            Author,
-            Subscriber,
-            Transport,
+        api::{
+            tangle::{
+                Author,
+                Subscriber,
+                Transport,
+            },
+            SequencingState,
         },
         message,
     },
@@ -24,7 +27,7 @@ pub fn s_fetch_next_messages<T: Transport>(
         let ids = subscriber.gen_next_msg_ids(multi_branching);
         exists = false;
 
-        for (pk, (next_id, seq_num)) in ids.iter() {
+        for (pk, SequencingState(next_id, seq_num)) in ids.iter() {
             let msg = transport.recv_message_with_options(&next_id, recv_opt).ok();
             if msg.is_none() {
                 continue;
@@ -90,7 +93,7 @@ pub fn a_fetch_next_messages<T: Transport>(
     while exists {
         let ids = author.gen_next_msg_ids(multi_branching);
         exists = false;
-        for (pk, (next_id, seq_num)) in ids.iter() {
+        for (pk, SequencingState(next_id, seq_num)) in ids.iter() {
             let msg = transport.recv_message_with_options(&next_id, recv_opt).ok();
             if msg.is_none() {
                 continue;