refactor to use cross-service sdk in pub-sub

examples/examples/manual_node.rs

@@ -74,7 +74,8 @@ async fn main() {
     });
 
     let spreads_layer_router = LayersSpreadRouterSyncBehavior::new(router.clone());
-    let (key_value, key_value_sdk) = key_value::KeyValueBehavior::new(args.node_id, timer.clone(), 10000);
+    let key_value_sdk = key_value::KeyValueSdk::new();
+    let key_value = key_value::KeyValueBehavior::new(args.node_id, 10000, Some(Box::new(key_value_sdk.clone())));
 
     if let Some(addr) = args.redis_addr {
         let mut redis_server = RedisServer::new(addr, key_value_sdk);

packages/core/utils/src/hashmap.rs

@@ -165,9 +165,8 @@ mod tests {
         let mut map = HashMap::new();
         map.insert("key1", "value1");
         map.insert("key2", "value2");
-        let mut iter = map.iter();
-        assert_eq!(iter.next(), Some((&"key1", &"value1")));
-        assert_eq!(iter.next(), Some((&"key2", &"value2")));
-        assert_eq!(iter.next(), None);
+        let mut slots = map.iter().collect::<Vec<_>>();
+        slots.sort_by_key(|a| a.0);
+        assert_eq!(slots, vec![(&"key1", &"value1"), (&"key2", &"value2")]);
     }
 }

packages/network/src/plane.rs

@@ -189,7 +189,7 @@
     fn pop_actions(&mut self, now_ms: u64) {
         while let Some(action) = self.internal.pop_action() {
             match action {
-                PlaneInternalAction::SpawnConnection(spwd_conn)  => {
+                PlaneInternalAction::SpawnConnection(spwd_conn) => {
                     let SpawnedConnection { outgoing, sender, receiver, handlers } = spwd_conn;
                     let internal_tx = self.internal_tx.clone();
 
@@ -293,7 +293,7 @@
                     NetworkBehaviorAction::CloseNode(node) => {
                         self.bus.close_node(node);
                     }
                     NetworkBehaviorAction::ToSdkService(_, _) => {}
                 },
             }
         }

packages/network/src/plane/bus_impl.rs

@@ -456,5 +456,5 @@ mod tests {
         let conn = Arc::new(sender);
         let _rx = bus.add_conn(conn).expect("Should have rx");
         assert!(bus.to_net_conn(ConnId::from_in(1, 1), TransportMsg::build_unreliable(1, RouteRule::ToService(2), 1, &[1u8])).is_some());
-}
+    }
 }

packages/network/src/plane/internal.rs

@@ -10,7 +10,7 @@
 
 use super::NetworkPlaneInternalEvent;
 
 #[derive(Debug, Eq, PartialEq)]
 pub enum PlaneInternalError {
     InvalidServiceId(u8),
 }
@@ -257,7 +257,7 @@
             if let Some((to_behaviour, to_context)) = &mut self.behaviors[to as usize] {
                 to_behaviour.on_sdk_msg(to_context, now_ms, from, msg);
             } else {
                 debug_assert!(false, "service not found {}", to);
             }
         }
     }
@@ -273,7 +273,7 @@
     use crate::{
         behaviour::MockNetworkBehavior,
         msg::TransportMsg,
-        transport::{ConnectionReceiver, ConnectionRejectReason, MockConnectionAcceptor, MockConnectionReceiver, MockConnectionSender, OutgoingConnectionError},
+        transport::{ConnectionRejectReason, MockConnectionAcceptor, MockConnectionReceiver, MockConnectionSender, OutgoingConnectionError},
     };
 
     type BE = ();

packages/runner/src/lib.rs

@@ -10,8 +10,8 @@ pub use network::{
     convert_enum,
 };
 pub use pub_sub::{
-    ChannelIdentify, ChannelSourceHashmapReal, ChannelUuid, Consumer, ConsumerRaw, ConsumerSingle, Feedback, FeedbackType, LocalPubId, LocalSubId, NumberInfo, Publisher, PublisherRaw, PubsubSdk,
-    PubsubServiceBehaviour, PubsubServiceBehaviourEvent, PubsubServiceHandlerEvent,
+    ChannelIdentify, ChannelUuid, Consumer, ConsumerRaw, ConsumerSingle, Feedback, FeedbackType, LocalPubId, LocalSubId, NumberInfo, Publisher, PublisherRaw, PubsubSdk, PubsubServiceBehaviour,
+    PubsubServiceBehaviourEvent, PubsubServiceHandlerEvent,
 };
 pub use transport_udp::UdpTransport;
 pub use utils::{