WIP.

fleck/examples/simple_node.rs

@@ -1,9 +1,14 @@
-use fleck::prelude::*;
+use fleck::{prelude::*, fleck_core};
 
 fn main() {
     pretty_env_logger::init_timed();
 
     let fleck = fleck::Fleck::new();
 
+    struct UDPSource;
+    struct UDPSink;
+    fleck.add_io()
+        .sink_with::<fleck_core::SendPacketChannel>(fleck_core::SendOrder::Send);
+
     fleck.run();
 }

fleck/src/fleck_core.rs

@@ -1,5 +1,8 @@
 // pub(crate) mod lowlevel;
 // mod nodes;
+// pub mod vchannel;
+pub mod peer;
+pub mod msg;
 
 // pub use nodes::Node;
 // pub use nodes::Nodes;
@@ -11,23 +14,23 @@ pub mod channel_tags {
     pub struct MajorTickTag {}
 }
 
-#[derive(PartialEq, PartialOrd)]
+#[derive(PartialEq, PartialOrd, Clone)]
 pub enum SendOrder {
     Route,
     Sign,
     Encrypt,
     Send,
 }
-// pub type SendPacketChannel = (channel_tags::SendPacketTag, crate::io::Packet, SendOrder);
+pub type SendPacketChannel = (channel_tags::SendPacketTag, msg::Message, SendOrder);
 
-#[derive(PartialEq, PartialOrd)]
+#[derive(PartialEq, PartialOrd, Clone)]
 pub enum ReceiveOrder {
     Decryption,
     Verify,
     Parse,
     Dispatch,
 }
-// pub type ReceivePacketChannel = (channel_tags::ReceivePacketTag, crate::io::Packet, ReceiveOrder);
+pub type ReceivePacketChannel = (channel_tags::ReceivePacketTag, msg::Message, ReceiveOrder);
 
 /// Minor ticks (roughly once per second): for retries, status updates, etc.
 pub type MinorTickChannel = (channel_tags::MinorTickTag, ());

fleck/src/msg.rs → fleck/src/fleck_core/msg.rs

@@ -53,6 +53,10 @@ pub struct Message {
     #[serde(skip_serializing)]
     #[serde(skip_deserializing)]
     pub(crate) parsed: Option<Box<dyn Any>>,
+
+    #[serde(skip_serializing)]
+    #[serde(skip_deserializing)]
+    pub(crate) peer: Option<super::peer::Peer>,
 }
 
 impl Message {
@@ -64,6 +68,7 @@ impl Message {
             data: bincode::serialize(&from).expect("couldn't serialize message"),
             saved_params: Some(SavedMessageParams::save::<M>()),
             parsed: Some(Box::new(from)),
+            peer: None
         }
     }
 
@@ -112,7 +117,7 @@ pub(crate) struct ParsePacket {}
 impl ParsePacket {}
 
 impl Service for std::rc::Rc<ParsePacket> {
-    fn register_channels(&self, eroot: &mut super::EventRoot) {
+    fn register_channels(&self, eroot: &mut crate::EventRoot) {
         /*eroot
         .channel::<fleck_core::ReceivePacketChannel>()
         // .with::<order::Preprocessing>()

fleck/src/fleck_core/peer.rs

@@ -0,0 +1,90 @@
+//! Peer: someone we can communicate with directly
+
+use std::{rc::Rc, collections::HashMap, cell::RefCell};
+
+use crate::prelude::*;
+
+use super::msg::Message;
+
+
+#[derive(Clone)]
+pub struct Peer {
+    inner: Rc<dyn InnerPeer>
+}
+
+impl std::fmt::Debug for Peer {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str("Peer")?;
+        Ok(())
+    }
+}
+
+trait InnerPeer {
+    fn dispatch(&self, api: &crate::Fleck, msg: super::msg::Message);
+}
+
+struct UdpPeer<SourceTag: 'static, SinkTag: 'static> {
+    _ghost: std::marker::PhantomData<(SourceTag, SinkTag)>,
+}
+
+impl<SourceTag: 'static, SinkTag: 'static> UdpPeer<SourceTag, SinkTag> {
+    fn forward(&self, api: crate::Fleck, msg: (std::net::SocketAddr, Vec<u8>)) -> Option<(std::net::SocketAddr, Vec<u8>)> {
+        None
+    }
+}
+
+impl<SourceTag: 'static, SinkTag: 'static> InnerPeer for UdpPeer<SourceTag, SinkTag> {
+    fn dispatch(&self, api: &crate::Fleck, msg: fleck_core::msg::Message) {
+        
+        api.queue::<(SinkTag, (std::net::SocketAddr, Vec<u8>))>(todo!());
+    }
+}
+
+struct UdpSocket {
+    peers: HashMap<std::net::SocketAddr, Rc<Peer>>,
+}
+
+impl UdpSocket {
+    fn handle_incoming(&self, ) {
+        
+    }
+
+    fn handle_outgoing(&self, ) {
+        
+    }
+}
+
+#[derive(Default)]
+pub struct PeerService {
+    udp: RefCell<Vec<Rc<UdpSocket>>>,
+}
+
+impl Service for std::rc::Rc<PeerService> {
+    fn register_channels(&self, eroot: &mut EventRoot) {
+        eroot.channel::<fleck_core::SendPacketChannel>().sub_opt(fleck_core::SendOrder::Send, self, PeerService::dispatch);
+    }
+}
+
+impl PeerService {
+    fn dispatch(&self, api: &crate::Fleck, msg: super::msg::Message) -> Option<super::msg::Message> {
+        /*let peer = match &msg.peer {
+            Some(peer) => peer.clone(),
+            None => {
+                log::error!("Packet got to peer dispatch with no peer?");
+                return None
+            }
+        };
+
+        peer.inner.dispatch(api, msg);*/
+
+        None
+    }
+
+    pub fn link_udp_socket<SourceTag: 'static, SinkTag: 'static>(&self) {
+        self.udp.borrow_mut().push(
+            Rc::new(UdpSocket {
+                peers: Default::default()
+            })
+        );
+    }
+}

fleck/src/fleck_core/vchannel.rs

@@ -0,0 +1,50 @@
+use std::{collections::HashSet, rc::Rc};
+
+use crate::prelude::*;
+
+pub struct VChannel {
+    
+}
+
+struct VChannelService<Tag: 'static> {
+    _ghost: std::marker::PhantomData<Tag>,
+}
+
+impl<Tag: 'static> Default for VChannelService<Tag> {
+    fn default() -> Self {
+        Self {
+            _ghost: Default::default(),
+        }
+    }
+}
+
+impl<Tag: 'static> Service for Rc<VChannelService<Tag>> {
+    fn register_channels(&self, eroot: &mut EventRoot) {
+        
+    }
+}
+
+#[derive(Default)]
+pub struct VChannelDispatch {
+    vchannels: HashSet<Rc<VChannel>>,
+}
+
+impl Service for std::rc::Rc<VChannelDispatch> {
+    fn register_channels(&self, eroot: &mut EventRoot) {
+        
+    }
+}
+
+impl VChannelDispatch {
+    pub fn link_udp_channel<SourceTag: 'static, SinkTag: 'static>(&self, api: &crate::Fleck) {
+        api.add_service::<VChannelService<(SourceTag, SinkTag)>>();
+    }
+
+    fn stream_receive(&self, api: &crate::Fleck, data: Vec<u8>) -> Option<Vec<u8>> {
+        None
+    }
+
+    fn udp_receive(&self, api: &crate::Fleck, data: (std::net::SocketAddr, Vec<u8>)) -> Option<(std::net::SocketAddr, Vec<u8>)> {
+        None
+    }
+}

fleck/src/io/builder.rs

@@ -1,4 +1,4 @@
-use std::{cell::{RefCell, Cell}, os::unix::prelude::{OwnedFd, AsRawFd, FromRawFd, RawFd}, rc::Rc, collections::VecDeque, io::{Read, Write}};
+use std::{cell::{RefCell, Cell}, os::unix::prelude::{OwnedFd, AsRawFd, FromRawFd, RawFd}, rc::Rc, collections::VecDeque, io::{Read, Write}, ops::DerefMut};
 
 use mio::{Token, unix::SourceFd};
 
@@ -24,7 +24,7 @@ pub struct StreamData {
 }
 
 pub struct TimerData {
-    timer: Option<timerfd::TimerState>,
+    interval: Option<std::time::Duration>,
 }
 
 struct InterestRegistration {
@@ -65,7 +65,6 @@ pub struct ServiceBuilder<'l, SourceChannel: ChannelSpec, SinkChannel: ChannelSp
     poll: Rc<RefCell<mio::Poll>>,
     token: Token,
 
-    source_priority: Option<SourceChannel::Priority>,
     sink_priority: Option<SinkChannel::Priority>,
 
     data: Data,
@@ -82,7 +81,6 @@ impl<'l, SourceChannel: ChannelSpec, SinkChannel: ChannelSpec>
             poll,
             token,
 
-            source_priority: None,
             sink_priority: None,
 
             data: NoData,
@@ -101,7 +99,6 @@ impl<'l, SinkChannel: ChannelSpec, Data> ServiceBuilder<'l, NoChannel, SinkChann
             poll: self.poll,
             token: self.token,
 
-            source_priority: Some(NoPriorityTag),
             sink_priority: self.sink_priority,
 
             data: self.data,
@@ -110,16 +107,17 @@ impl<'l, SinkChannel: ChannelSpec, Data> ServiceBuilder<'l, NoChannel, SinkChann
         }
     }
 
-    pub fn source_with<Priority: PartialOrd, SourceChannel: ChannelSpec<Priority = Priority>>(
+    pub fn source_with<SourceChannel: ChannelSpec>(
         self,
-        p: Priority,
-    ) -> ServiceBuilder<'l, SourceChannel, SinkChannel, Data> {
+        p: SourceChannel::Priority,
+    ) -> ServiceBuilder<'l, SourceChannel, SinkChannel, Data>
+    where SourceChannel::Priority: PartialOrd
+    {
         ServiceBuilder {
             fleck: self.fleck,
             poll: self.poll,
             token: self.token,
 
-            source_priority: Some(p),
             sink_priority: self.sink_priority,
 
             data: self.data,
@@ -138,7 +136,6 @@ impl<'l, SourceChannel: ChannelSpec, Data> ServiceBuilder<'l, SourceChannel, NoC
             poll: self.poll,
             token: self.token,
 
-            source_priority: self.source_priority,
             sink_priority: Some(NoPriorityTag),
 
             data: self.data,
@@ -147,16 +144,16 @@ impl<'l, SourceChannel: ChannelSpec, Data> ServiceBuilder<'l, SourceChannel, NoC
         }
     }
 
-    pub fn sink_with<Priority: PartialOrd, SinkChannel: ChannelSpec<Priority = Priority>>(
+    pub fn sink_with<SinkChannel: ChannelSpec>(
         self,
-        p: Priority,
-    ) -> ServiceBuilder<'l, SourceChannel, SinkChannel, Data> {
+        p: SinkChannel::Priority,
+    ) -> ServiceBuilder<'l, SourceChannel, SinkChannel, Data>
+    where SinkChannel::Priority: PartialOrd {
         ServiceBuilder {
             fleck: self.fleck,
             poll: self.poll,
             token: self.token,
 
-            source_priority: self.source_priority,
             sink_priority: Some(p),
 
             data: self.data,
@@ -167,9 +164,8 @@ impl<'l, SourceChannel: ChannelSpec, Data> ServiceBuilder<'l, SourceChannel, NoC
 }
 
 impl<'l,
-        PacketType: 'static + From<(std::net::SocketAddr, Vec<u8>)> + Into<(std::net::SocketAddr, Vec<u8>)>,
-        SourceChannel: ChannelSpec<Data = PacketType>,
-        SinkChannel: ChannelSpec<Data = PacketType>,
+        SourceChannel: ChannelSpec<Data = (std::net::SocketAddr, Vec<u8>)>,
+        SinkChannel: ChannelSpec<Data = (std::net::SocketAddr, Vec<u8>)>,
     > ServiceBuilder<'l, SourceChannel, SinkChannel, NoData>
 {
     pub fn udp(self) -> ServiceBuilder<'l, SourceChannel, SinkChannel, UdpData> {
@@ -178,7 +174,6 @@ impl<'l,
             poll: self.poll,
             token: self.token,
 
-            source_priority: self.source_priority,
             sink_priority: self.sink_priority,
 
             data: UdpData {
@@ -192,9 +187,8 @@ impl<'l,
 }
 
 impl<'l,
-        PacketType: 'static + From<(std::net::SocketAddr, Vec<u8>)> + Into<(std::net::SocketAddr, Vec<u8>)>,
-        SourceChannel: ChannelSpec<Data = PacketType>,
-        SinkChannel: ChannelSpec<Data = PacketType>,
+        SourceChannel: ChannelSpec<Data = (std::net::SocketAddr, Vec<u8>)>,
+        SinkChannel: ChannelSpec<Data = (std::net::SocketAddr, Vec<u8>)>,
     > ServiceBuilder<'l, SourceChannel, SinkChannel, UdpData>
 {
     pub fn set_bind_address(mut self, bind_addr: std::net::SocketAddr) -> Self {
@@ -215,8 +209,62 @@ impl<'l,
     }
 
     pub fn build(self) {
+        let socket = mio::net::UdpSocket::bind(
+            self.data.bind_addr.expect("UDP socket not given a bind address!")
+        ).expect("couldn't bind to UDP socket");
+
+        if let Some(mcast) = self.data.join_multicast {
+            use std::net::IpAddr;
+            match (mcast.group, mcast.iface) {
+                (IpAddr::V4(group), IpAddr::V4(iface)) => {
+                  socket.join_multicast_v4(&group, &iface).expect("couldn't join multicast group");
+                },
+                (IpAddr::V6(_group), IpAddr::V6(_iface)) => { todo!() },
+                _ => panic!("Multicast specification mixes ipv4 and ipv6 addresses!"),
+            }
+        }
+
+        let svc = Rc::new(UdpService::<SourceChannel, SinkChannel> {
+            interest: InterestRegistration { poll: self.poll, write: Cell::new(false), token: self.token },
+            queue: Default::default(),
+            // file: RefCell::new(unsafe { std::fs::File::from_raw_fd(self.data.fd.expect("Building stream IO service with no FD?")) }),
+            socket: RefCell::new(socket),
+            sink_priority: RefCell::new(Some(self.sink_priority.expect("no sink priority given!"))),
+            _ghost: Default::default(),
+        });
+
+        svc.interest.subscribe_read(svc.socket.borrow_mut().deref_mut());
+
+        self.fleck.io.borrow_mut().services.borrow_mut().insert(self.token, svc.clone());
+        self.fleck.services.borrow_mut().add_service(svc.clone());
+        
+    }
+}
+
+struct UdpService<SourceChannel: ChannelSpec, SinkChannel: ChannelSpec> {
+    interest: InterestRegistration,
+    queue: VecDeque<(std::net::SocketAddr, Vec<u8>)>,
+    socket: RefCell<mio::net::UdpSocket>,
+    sink_priority: RefCell<Option<SinkChannel::Priority>>,
+    _ghost: std::marker::PhantomData<(SourceChannel, SinkChannel)>,
+}
+
+impl<SourceChannel: ChannelSpec, SinkChannel: ChannelSpec> Service for Rc<UdpService<SourceChannel, SinkChannel>> {
+    
+}
+
+impl<SourceChannel: ChannelSpec, SinkChannel: ChannelSpec> AbstractFdService for UdpService<SourceChannel, SinkChannel> {
+    fn ready_read(&self, api: &crate::Fleck) {
+        
+    }
+
+    fn ready_write(&self, api: &crate::Fleck) {
         
     }
+
+    fn token(&self) -> Token {
+        self.interest.token
+    }
 }
 
 /* timerfd support */
@@ -229,10 +277,9 @@ impl<'l, SourceChannel: ChannelSpec<Data = ()>>
             poll: self.poll,
             token: self.token,
 
-            source_priority: self.source_priority,
             sink_priority: self.sink_priority,
 
-            data: TimerData { timer: None },
+            data: TimerData { interval: None },
 
             _ghost: Default::default(),
         }
@@ -252,7 +299,7 @@ impl<Tag: 'static> AbstractFdService for Timer<Tag> {
         api.queue::<(Tag, ())>(());
     }
 
-    fn ready_write(&self, api: &crate::Fleck) {
+    fn ready_write(&self, _api: &crate::Fleck) {
         unreachable!();
     }
 
@@ -264,16 +311,19 @@ impl<Tag: 'static> AbstractFdService for Timer<Tag> {
 impl<'l, SourceChannel: ChannelSpec<Data = ()>>
     ServiceBuilder<'l, SourceChannel, ((), ()), TimerData>
 {
-    pub fn setup(mut self, state: timerfd::TimerState) -> Self {
-        self.data.timer = Some(state);
+    pub fn interval(mut self, time: std::time::Duration) -> Self {
+        self.data.interval = Some(time);
         self
     }
 
     pub fn build(self) {
-        let mut timer = timerfd::TimerFd::new_custom(timerfd::ClockId::Monotonic, true, false).expect("couldn't create timer");
-        timer.set_state(self.data.timer.expect("no setup done for timer?"), timerfd::SetTimeFlags::Default);
+        let mut raw_timer = timerfd::TimerFd::new_custom(timerfd::ClockId::Monotonic, true, false).expect("couldn't create timer");
+        raw_timer.set_state(timerfd::TimerState::Periodic {
+            current: std::time::Duration::from_millis(1),
+            interval: self.data.interval.expect("no interval set for timer"),
+        }, timerfd::SetTimeFlags::Default);
 
-        let timer = Rc::new(Timer::<SourceChannel::Tag> { token: self.token, timer, _ghost: Default::default() });
+        let timer = Rc::new(Timer::<SourceChannel::Tag> { token: self.token, timer: raw_timer, _ghost: Default::default() });
 
         self.poll.borrow().registry().register(
             &mut SourceFd(&timer.timer.as_raw_fd()),
@@ -295,7 +345,6 @@ impl<'l, SourceChannel: ChannelSpec<Data = Vec<u8>>, SinkChannel: ChannelSpec<Da
             poll: self.poll,
             token: self.token,
 
-            source_priority: self.source_priority,
             sink_priority: self.sink_priority,
 
             data: StreamData { fd: None },
@@ -360,12 +409,6 @@ pub struct StreamService<SourceChannel: ChannelSpec<Data = Vec<u8>>, SinkChannel
     _ghost: std::marker::PhantomData<(SourceChannel, SinkChannel)>,
 }
 
-/*impl<SourceChannel: ChannelSpec<Data = Vec<u8>>, SinkChannel: ChannelSpec<Data = Vec<u8>, Priority = SinkPriority>, SinkPriority: 'static + PartialOrd + Clone> Service for Rc<StreamService<SourceChannel, SinkChannel>> {
-    fn register_channels(&self, eroot: &mut crate::prelude::EventRoot) {
-        eroot.channel::<SinkChannel>().sub_opt(self.sink_priority.clone(), self, StreamService::sink_data);
-    }
-}*/
-
 impl<SourceChannel: ChannelSpec<Data = Vec<u8>>, SinkTag: 'static, SinkPriority: 'static + Clone + PartialOrd> Service for Rc<StreamService<SourceChannel, (SinkTag, Vec<u8>, SinkPriority)>> {
     fn register_channels(&self, eroot: &mut crate::prelude::EventRoot) {
         eroot.channel::<(SinkTag, Vec<u8>, SinkPriority)>().sub_opt(self.sink_priority.borrow_mut().take().expect("registering channel with no priority?"), self, StreamService::sink_data);
@@ -379,7 +422,7 @@ impl<SourceChannel: ChannelSpec<Data = Vec<u8>>, SinkTag: 'static> Service for R
 }
 
 impl<SourceChannel: ChannelSpec<Data = Vec<u8>>, SinkChannel: ChannelSpec<Data = Vec<u8>>> StreamService<SourceChannel, SinkChannel> {
-    fn sink_data(&self, api: &crate::Fleck, mut data: Vec<u8>) -> Option<Vec<u8>> {
+    fn sink_data(&self, _api: &crate::Fleck, mut data: Vec<u8>) -> Option<Vec<u8>> {
         self.queue.borrow_mut().extend(data.drain(..));
         self.interest.subscribe_write(&mut SourceFd(&self.file.borrow().as_raw_fd()));
         None

fleck/src/lib.rs

@@ -3,7 +3,6 @@ use std::collections::HashMap;
 use std::os::unix::prelude::{AsRawFd, OwnedFd};
 use std::rc::Rc;
 
-mod msg;
 // mod crypto;
 pub mod fleck_core;
 mod helper;
@@ -12,34 +11,13 @@ pub mod service;
 
 pub mod prelude {
     pub use crate::helper::{AsAny, IntoWeak};
-    pub use crate::msg::{Message, MessageParams};
+    // pub use crate::fleck_core::msg::{Message, MessageParams};
     pub use crate::service::{ChannelSpec, EventRoot, Service};
-    // pub use crate::fleck_core;
+    pub use crate::fleck_core;
 }
 
 use prelude::*;
 
-struct SimpleTag;
-
-#[derive(Default)]
-struct SimpleListener<T: 'static>(std::marker::PhantomData<T>);
-
-impl<T: 'static> Service for Rc<SimpleListener<T>> {
-    fn register_channels(&self, eroot: &mut EventRoot) {
-        eroot.create_channel::<(SimpleTag, T)>();
-        eroot
-            .channel::<(SimpleTag, T)>()
-            .sub_opt(self, SimpleListener::<T>::tick);
-    }
-}
-
-impl<T: 'static> SimpleListener<T> {
-    fn tick(&self, _: &Fleck, _: T) -> Option<T> {
-        log::trace!("Ticked!");
-        None
-    }
-}
-
 pub struct Fleck {
     io: RefCell<io::IO>,
     services: RefCell<service::ServiceStack>,
@@ -52,67 +30,44 @@ impl Fleck {
             services: Default::default(),
         });
 
+        res.timer_setup();
         res.register_core_services();
-        res.set_up_timers();
 
         res
     }
 
-    fn set_up_timers(&self) {
+    fn timer_setup(&self) {
         // minor tick registration
-        let mut minor =
-            timerfd::TimerFd::new_custom(timerfd::ClockId::Monotonic, true, false).unwrap();
-        minor.set_state(
-            timerfd::TimerState::Periodic {
-                current: std::time::Duration::new(1, 0),
-                interval: std::time::Duration::new(1, 0),
-            },
-            timerfd::SetTimeFlags::Default,
-        );
-
-        self.add_service::<SimpleListener<Vec<u8>>>();
-        self.add_service::<SimpleListener<()>>();
-
-        /*self.add_io()
-            .sink::<(SimpleTag, Vec<u8>)>()
-            .source::<(SimpleTag, Vec<u8>)>()
-            .stream()
-            .fd(minor.as_raw_fd())
-            .build();*/
-
         self.add_io()
-            .source::<(SimpleTag, ())>()
+            .source::<(fleck_core::MinorTickChannel, ())>()
             .timer()
-            .setup(timerfd::TimerState::Periodic { current: std::time::Duration::new(1, 0), interval: std::time::Duration::new(1, 0) })
+            .interval(std::time::Duration::new(1, 0))
+            .build();
+        // major tick registration
+        self.add_io()
+            .source::<(fleck_core::MajorTickChannel, ())>()
+            .timer()
+            .interval(std::time::Duration::new(15, 0))
             .build();
-
-        Box::leak(Box::new(minor));
     }
 
     fn register_core_services(&self) {
         // initial outgoing/incoming/minor/major channels
         // self.services.borrow_mut().create_io_channels();
 
-        /*
         // Node registration
-        self.add_service::<fleck_core::Nodes>();
+        // self.add_service::<fleck_core::Nodes>();
         // Local node discovery
-        self.add_service::<fleck_core::lowlevel::LocalDiscovery>();
+        // self.add_service::<fleck_core::lowlevel::LocalDiscovery>();
         // Actually sending packets
-        self.add_service::<fleck_core::lowlevel::SendPacket>();
+        // self.add_service::<fleck_core::lowlevel::SendPacket>();
         // Parsing incoming packets
-        self.add_service::<msg::ParsePacket>();
+        self.add_service::<fleck_core::msg::ParsePacket>();
         // Signing packets
-        self.add_service::<crypto::SignPacket>();
-        */
+        // self.add_service::<crypto::SignPacket>();
     }
 
     pub fn run(&self) {
-        /*let io = io::IO::new();
-
-        // TODO: insert/build global socket, minor, major timerfds
-
-        io.run(self);*/
         self.io.borrow().run(self);
     }
 }

fleck/src/service.rs

@@ -7,12 +7,14 @@ pub(crate) mod event;
 
 pub use event::ChannelSpec;
 
+use crate::fleck_core;
+
 pub type EventRoot = event::EventRoot<crate::Fleck>;
 
 #[allow(unused_variables)]
 pub trait Service: std::any::Any + crate::helper::AsAny {
     fn register_channels(&self, eroot: &mut EventRoot) {}
-    fn register_msgs(&self, registry: &mut crate::msg::MessageRegistry) {}
+    fn register_msgs(&self, registry: &mut fleck_core::msg::MessageRegistry) {}
 }
 
 trait ServiceExt {
@@ -32,7 +34,7 @@ pub struct ServiceStack {
     service_map: HashMap<TypeId, Box<dyn Service>>,
 
     eroot: EventRoot,
-    message_registry: crate::msg::MessageRegistry,
+    message_registry: fleck_core::msg::MessageRegistry,
 }
 
 impl ServiceStack {
@@ -75,7 +77,7 @@ impl ServiceStack {
         &self.eroot
     }
 
-    pub fn message_registry(&self) -> &crate::msg::MessageRegistry {
+    pub fn message_registry(&self) -> &fleck_core::msg::MessageRegistry {
         &self.message_registry
     }
 }