Intra-kernel compatability

Cargo.toml

@@ -22,9 +22,9 @@ citadel-internal-service-connector = { path = "./citadel-internal-service-connec
 citadel-internal-service-macros = { path = "./citadel-internal-service-macros", default-features = false, version = "0.1.0" }
 
 # Avarok deps
-citadel_sdk = { git = "https://github.com/Avarok-Cybersecurity/Citadel-Protocol" }
-citadel_types = { git = "https://github.com/Avarok-Cybersecurity/Citadel-Protocol" }
-citadel_logging = { git = "https://github.com/Avarok-Cybersecurity/Citadel-Protocol", default-features = false }
+citadel_sdk = { git = "https://github.com/Avarok-Cybersecurity/Citadel-Protocol/" }
+citadel_types = { git = "https://github.com/Avarok-Cybersecurity/Citadel-Protocol/" }
+citadel_logging = { git = "https://github.com/Avarok-Cybersecurity/Citadel-Protocol/" }
 
 # Standard deps
 serde = { version = "1.0.104", features = ["derive"] }
@@ -41,4 +41,5 @@ uuid = { version="1.3.3", features = [
 anyhow = "1.0.71"
 async-recursion = { version = "1.0.4" }
 parking_lot = { version = "0.12.1" }
-structopt = { version = "0.3.26" }
+structopt = { version = "0.3.26" }
+lazy_static = "1.4.0"

citadel-internal-service-types/src/lib.rs

@@ -115,6 +115,7 @@ pub struct DeleteVirtualFileFailure {
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct PeerConnectSuccess {
     pub cid: u64,
+    pub peer_cid: u64,
     pub request_id: Option<Uuid>,
 }
 
@@ -557,7 +558,7 @@ pub struct FileTransferStatusNotification {
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct FileTransferTickNotification {
     pub cid: u64,
-    pub peer_cid: u64,
+    pub peer_cid: Option<u64>,
     pub status: ObjectTransferStatus,
 }
 

citadel-internal-service/Cargo.toml

@@ -6,6 +6,7 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [features]
+default = []
 vendored = ["citadel_sdk/vendored"]
 
 [dependencies]

citadel-internal-service/src/kernel/mod.rs

@@ -4,6 +4,7 @@ use citadel_internal_service_connector::util::wrap_tcp_conn;
 use citadel_internal_service_types::*;
 use citadel_logging::{error, info, warn};
 use citadel_sdk::prefabs::ClientServerRemote;
+use citadel_sdk::prelude::remote_specialization::PeerRemote;
 use citadel_sdk::prelude::VirtualTargetType;
 use citadel_sdk::prelude::*;
 use futures::stream::{SplitSink, StreamExt};
@@ -56,7 +57,7 @@ pub struct Connection {
 #[allow(dead_code)]
 struct PeerConnection {
     sink: PeerChannelSendHalf,
-    remote: SymmetricIdentifierHandle,
+    remote: PeerRemote,
     handler_map: HashMap<u64, Option<ObjectTransferHandler>>,
     associated_tcp_connection: Uuid,
 }
@@ -88,7 +89,7 @@ impl Connection {
         &mut self,
         peer_cid: u64,
         sink: PeerChannelSendHalf,
-        remote: SymmetricIdentifierHandle,
+        remote: PeerRemote,
     ) {
         self.peers.insert(
             peer_cid,
@@ -333,8 +334,8 @@ fn handle_connection(
                             }
                         }
                     }
-                    Err(_) => {
-                        warn!(target: "citadel", "Bad message from client");
+                    Err(err) => {
+                        warn!(target: "citadel", "Bad message from client: {err:?}");
                     }
                 }
             }
@@ -356,7 +357,7 @@ fn handle_connection(
 fn spawn_tick_updater(
     object_transfer_handler: ObjectTransferHandler,
     implicated_cid: u64,
-    peer_cid: u64,
+    peer_cid: Option<u64>,
     server_connection_map: &mut HashMap<u64, Connection>,
     tcp_connection_map: Arc<Mutex<HashMap<Uuid, UnboundedSender<InternalServiceResponse>>>>,
 ) {
@@ -377,7 +378,7 @@ fn spawn_tick_updater(
                         );
                         match entry.send(message.clone()) {
                             Ok(_res) => {
-                                info!(target: "citadel", "File Transfer Status Tick Sent");
+                                info!(target: "citadel", "File Transfer Status Tick Sent {status:?}");
                             }
                             Err(err) => {
                                 warn!(target: "citadel", "File Transfer Status Tick Not Sent: {err:?}");

citadel-internal-service/src/kernel/requests/file/respond_file_transfer.rs

@@ -32,7 +32,7 @@ pub async fn handle(
                     spawn_tick_updater(
                         owned_handler,
                         cid,
-                        peer_cid,
+                        Some(peer_cid),
                         &mut server_connection_map,
                         this.tcp_connection_map.clone(),
                     );

citadel-internal-service/src/kernel/requests/group/request_join.rs

@@ -5,7 +5,7 @@ use citadel_internal_service_types::{
     InternalServiceResponse,
 };
 use citadel_sdk::prelude::{
-    GroupBroadcast, GroupBroadcastCommand, GroupEvent, NodeRequest, NodeResult,
+    GroupBroadcast, GroupBroadcastCommand, GroupEvent, NodeRequest, NodeResult, TargetLockedRemote,
 };
 use futures::StreamExt;
 use uuid::Uuid;
@@ -38,7 +38,11 @@ pub async fn handle(
                 implicated_cid: cid,
                 command: group_request,
             });
-            match peer_remote.send_callback_subscription(request).await {
+            match peer_remote
+                .remote()
+                .send_callback_subscription(request)
+                .await
+            {
                 Ok(mut subscription) => {
                     let mut result = Err("Group Request Join Failed".to_string());
                     while let Some(evt) = subscription.next().await {

citadel-internal-service/src/kernel/requests/group/respond_request.rs

@@ -5,7 +5,8 @@ use citadel_internal_service_types::{
     InternalServiceResponse,
 };
 use citadel_sdk::prelude::{
-    GroupBroadcast, GroupBroadcastCommand, GroupChannelCreated, GroupEvent, NodeRequest, NodeResult,
+    GroupBroadcast, GroupBroadcastCommand, GroupChannelCreated, GroupEvent, NodeRequest,
+    NodeResult, TargetLockedRemote,
 };
 use futures::StreamExt;
 use uuid::Uuid;
@@ -52,7 +53,11 @@ pub async fn handle(
             let peer_remote = peer_connection.remote.clone();
             drop(server_connection_map);
 
-            match peer_remote.send_callback_subscription(request).await {
+            match peer_remote
+                .remote()
+                .send_callback_subscription(request)
+                .await
+            {
                 Ok(mut subscription) => {
                     let mut result = false;
                     if invitation {

citadel-internal-service/src/kernel/requests/peer/connect.rs

@@ -38,6 +38,7 @@ pub async fn handle(
     if already_connected {
         let response = InternalServiceResponse::PeerConnectSuccess(PeerConnectSuccess {
             cid,
+            peer_cid,
             request_id: Some(request_id),
         });
 
@@ -67,11 +68,7 @@ pub async fn handle(
                         .await
                         .get_mut(&cid)
                         .unwrap()
-                        .add_peer_connection(
-                            peer_cid,
-                            sink,
-                            symmetric_identifier_handle_ref.into_owned(),
-                        );
+                        .add_peer_connection(peer_cid, sink, peer_connect_success.remote);
 
                     let hm_for_conn = this.tcp_connection_map.clone();
 
@@ -101,6 +98,7 @@ pub async fn handle(
 
                     InternalServiceResponse::PeerConnectSuccess(PeerConnectSuccess {
                         cid,
+                        peer_cid,
                         request_id: Some(request_id),
                     })
                 }

citadel-internal-service/src/kernel/responses/object_transfer_handle.rs

@@ -41,7 +41,7 @@ pub async fn handle(
                 spawn_tick_updater(
                     object_transfer_handler,
                     implicated_cid,
-                    peer_cid,
+                    Some(peer_cid),
                     &mut server_connection_map,
                     this.tcp_connection_map.clone(),
                 );
@@ -74,7 +74,7 @@ pub async fn handle(
         spawn_tick_updater(
             object_transfer_handler,
             implicated_cid,
-            peer_cid,
+            Some(peer_cid),
             &mut server_connection_map,
             this.tcp_connection_map.clone(),
         );