ubisync-lib/src/messages/mod.rs

@@ -1,6 +1,6 @@
 use serde::{Deserialize, Serialize};
 
-use crate::types::{ElementContent, ElementId, Family, MessageId, PotId};
+use crate::types::{ElementContent, ElementId, Family, FamilyId, MessageId, PotId};
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct Message {

ubisync-lib/src/types/peer_id.rs

@@ -1,7 +1,7 @@
 use std::hash::Hash;
 
-use anyhow::{anyhow, Error};
+use anyhow::{anyhow, bail, Error};
-use i2p::net::{I2pAddr, I2pSocketAddr};
+use i2p::net::{I2pAddr, I2pSocketAddr, ToI2pSocketAddrs};
 use regex::Regex;
 use serde::{Deserialize, Serialize};
 
@@ -52,11 +52,7 @@ impl From<PeerId> for I2pSocketAddr {
 
 impl From<I2pSocketAddr> for PeerId {
     fn from(value: I2pSocketAddr) -> Self {
-        Self::try_from_b32(&value.dest().string(), Some(value.port())).unwrap_or(
+        PeerId { addr: value }
-            Self::try_from_dest(&value.dest().string(), Some(value.port())).expect(&format!(
-                "I2pSocketAddr was neither in b32 nor in b64 format: {value:?}"
-            )),
-        )
     }
 }
 

ubisync/src/comm/message_processor.rs

@@ -1,7 +1,7 @@
 use tracing::debug;
 
 use ubisync_lib::peer::Peer;
-use ubisync_lib::types::{ContentUpdateStrategy, PeerId};
+use ubisync_lib::types::{ContentUpdateStrategy, Family, PeerId};
 
 use ubisync_lib::messages::{Message, MessageContent};
 
@@ -25,9 +25,15 @@ pub fn handle(state: &CommState, peer: &PeerId, message: Message) {
         MessageContent::JoinFamily => state.request_family_join(peer.to_owned()),
         MessageContent::AddedToFamily { family } => {
             if state.has_family_join_request(peer.to_owned()) {
+                debug!("Own join request was accepted, setting family");
                 state
                     .set_own_family(family.to_owned())
                     .expect("State failed");
+                
+                debug!("New own family: {:?}", state.get_family_of_peer(state.own_peer_id().unwrap()))
+            }
+            else {
+                debug!("Got AddedToFamily message, but no family join request was found")
             }
         }
         MessageContent::LeaveFamily => state.remove_peer_from_family(peer.to_owned()),

ubisync/src/comm/mod.rs

@@ -1,6 +1,7 @@
 mod conflict_resolution;
 pub mod message_processor;
 
+use i2p::sam::StreamForward;
 use tracing::{debug, error, warn};
 use ubisync_lib::messages::Message;
 use ubisync_lib::types::PeerId;
@@ -38,7 +39,7 @@ impl CommHandle {
         }
 
         let listener = listener_builder.build().unwrap();
-        let own_peer_id: PeerId = (&listener).local_addr().map_err(|e| anyhow!(e))?.into();
+        let mut own_peer_id: PeerId = (&listener).local_addr().map_err(|e| anyhow!(e))?.into();
 
         Ok(CommHandle {
             state: Arc::new(state),

ubisync/src/lib.rs

@@ -1,4 +1,4 @@
-use std::sync::Arc;
+use std::{future::Future, sync::Arc};
 
 use anyhow::bail;
 use api::{v0::app::App, Api, ApiBuilder};

ubisync/src/state/comm_state.rs

@@ -6,8 +6,7 @@ use ubisync_lib::{
     api::events::AppEvent,
     peer::Peer,
     types::{
-        ContentUpdateStrategy, Element, ElementContent, ElementId, Family, FamilyId, MessageId,
+        ContentUpdateStrategy, Element, ElementContent, ElementId, Family, FamilyId, MessageId, PeerId, PotId
-        PeerId, PotId,
     },
 };
 
@@ -102,13 +101,12 @@ impl CommState {
     }
 
     pub fn request_family_join(&self, peer: PeerId) {
-        let _ = self
+        self.state
-            .state
             .emit_node_event(UbisyncNodeEvent::FamilyJoinRequest { joiner: peer });
     }
 
     pub fn remove_peer_from_family(&self, peer: PeerId) {
-        let _ = self.db().remove_peer_from_family(peer);
+        self.db().remove_peer_from_family(peer);
     }
 
     pub fn has_family_join_request(&self, peer: PeerId) -> bool {

ubisync/src/state/database/mod.rs

@@ -32,7 +32,7 @@ impl StateDB {
     }
 
     pub fn add_family_join_request(&self, peer: PeerId) {
-        let _ = self.db.set_key(peer.to_string(), &"").execute();
+        self.db.set_key(peer.to_string(), &"").execute();
         debug!(
             "Added join request: {:?}",
             self.db.get_key(peer.to_string()).query()

ubisync/src/state/mod.rs

@@ -13,7 +13,7 @@ use ubisync_lib::{
     types::{AppId, Element, ElementContent, ElementId, Family, FamilyId, PeerId, PotId, Tag},
 };
 
-use anyhow::Error;
+use anyhow::{anyhow, Error};
 use tracing::{debug, warn};
 
 mod api_state;

ubisync/tests/api.rs

@@ -41,7 +41,7 @@ async fn two_nodes_element_creation() {
     let app_id2 = api_client2.app_id();
     ubi2.set_node_event_callback(
         move |ev, node| {
-            if let UbisyncNodeEvent::NewPot { id, app_type: _ } = ev {
+            if let UbisyncNodeEvent::NewPot { id, app_type } = ev {
                 debug!("callback called");
                 node.add_pot_member(id, app_id2.clone()).unwrap();
             }
@@ -105,7 +105,7 @@ async fn two_nodes_api_event() {
     ubi1.set_node_event_callback(
         move |ev, node| {
             debug!("callback called");
-            if let UbisyncNodeEvent::NewPot { id, app_type: _ } = ev {
+            if let UbisyncNodeEvent::NewPot { id, app_type } = ev {
                 node.add_pot_member(id, app_id1.clone()).unwrap();
             }
         },