Code cleanup.

fleck/src/fleck_core/io.rs

@@ -28,13 +28,6 @@ pub trait FdHandler {
     fn dispatch(&self, api: &crate::Fleck, msg: super::msg::Message);
 }
 
-/*pub(crate) struct FdHandler {
-    pub(crate) ready_read: Rc<dyn Fn(&crate::Fleck)>,
-    pub(crate) ready_write: Rc<dyn Fn(&crate::Fleck)>,
-    pub(crate) dispatch: Rc<dyn Fn(&crate::Fleck, super::msg::Message)>,
-    owned: Rc<dyn std::any::Any>,
-}*/
-
 pub struct IOService {
     poll: Rc<RefCell<mio::Poll>>,
     next_token: Cell<mio::Token>,
@@ -51,7 +44,12 @@ impl Default for IOService {
     }
 }
 
-impl Service for Rc<IOService> { }
+impl Service for Rc<IOService> {
+    fn setup(&self, api: &crate::Fleck) {
+        api.create_channel::<super::ReceivePacketChannel>();
+        api.create_channel::<super::SendPacketChannel>();
+    }
+}
 
 impl IOService {
     pub fn register_handler(&self, reg: &InterestRegistration, handler: Rc<dyn FdHandler>) {

fleck/src/fleck_core/peer.rs

@@ -30,8 +30,6 @@ pub struct Peer {
 
 #[derive(Default)]
 pub struct PeerService {
-    // udp: RefCell<Vec<Rc<UdpSocket>>>,
-    // stream: RefCell<Vec<Rc<
 }
 
 impl Service for std::rc::Rc<PeerService> {

fleck/src/helper.rs

@@ -37,44 +37,3 @@ impl<'a, T> IntoWeak<T> for &'a std::rc::Weak<T> {
         self.clone()
     }
 }
-
-/*pub struct ClosureRef<'l, Target: 'l, Closure: Fn() -> &'l Target> {
-    closure: Closure,
-}
-
-impl<'l, Target: 'l, Closure: Fn() -> &'l Target> ClosureRef<'l, Target, Closure> {
-    pub fn new(c: Closure) -> Self {
-        Self { closure: c }
-    }
-}
-
-impl<'l, Target: 'l, Closure: Fn() -> &'l Target> AsRef<Target> for ClosureRef<'l, Target, Closure> {
-    fn as_ref(&self) -> &Target {
-        (self.closure)()
-    }
-}
-
-pub struct OwnedRef<'base, Base: 'base, Contained> {
-    own: Base,
-    get: Box<dyn Fn(&Base) -> &Contained>,
-    _ghost: std::marker::PhantomData<&'base ()>,
-}
-
-impl<'base, Base, Contained> OwnedRef<'base, Base, Contained> {
-    pub fn new<F: 'static + Fn(&Base) -> &Contained>(own: Base, get: F) -> Self {
-        Self { own, get: Box::new(get), _ghost: Default::default() }
-    }
-}
-
-impl<'base, Base, Contained> AsRef<Contained> for OwnedRef<'base, Base, Contained> {
-    fn as_ref(&self) -> &Contained {
-        (self.get)(&self.own)
-    }
-}
-
-impl<'base, Base, Contained> std::ops::Deref for OwnedRef<'base, Base, Contained> {
-    type Target = Contained;
-    fn deref(&self) -> &Self::Target {
-        (self.get)(&self.own)
-    }
-}*/

fleck/src/io.rs

@@ -1,323 +0,0 @@
-use std::{
-    cell::{Cell, RefCell},
-    collections::HashMap,
-    rc::Rc,
-};
-
-use mio::{Events, Interest, Poll, Token};
-
-use crate::{prelude::ChannelSpec, service::event::NoPriorityTag, Fleck};
-
-pub mod builder;
-// pub(super) mod stream;
-// pub(super) mod udp;
-
-use builder::DefaultServiceBuilder;
-
-lazy_static::lazy_static! {
-    pub static ref MULTICAST_ADDRESS : std::net::SocketAddrV4 =
-        std::net::SocketAddrV4::new(
-            std::net::Ipv4Addr::new(239, 0, 2, 35),
-            3535
-        );
-}
-
-pub const DEFAULT_PORT: u16 = 3535;
-
-pub(super) trait AbstractFdService: 'static {
-    fn ready_read(&self, api: &Fleck);
-    fn ready_write(&self, api: &Fleck);
-    fn token(&self) -> Token;
-}
-
-pub(super) struct IO {
-    poll: Rc<RefCell<mio::Poll>>,
-    services: RefCell<HashMap<Token, Rc<dyn AbstractFdService>>>,
-}
-
-impl IO {
-    pub(super) fn new() -> Self {
-        Self {
-            poll: Rc::new(RefCell::new(
-                mio::Poll::new().expect("couldn't create mio::Poll"),
-            )),
-            services: Default::default(),
-        }
-    }
-
-    pub(super) fn run(&self, api: &Fleck) {
-        let mut events = mio::Events::with_capacity(128);
-        loop {
-            log::trace!("Waiting for events...");
-            let pr = self.poll.borrow_mut().poll(&mut events, None);
-            log::trace!("Polled!");
-            if pr.is_err() {
-                continue;
-            }
-            log::trace!("Waking up for events");
-            let svcs = self.services.borrow();
-            for evt in &events {
-                let svc = svcs.get(&evt.token()).unwrap();
-                if evt.is_readable() {
-                    svc.ready_read(api);
-                }
-                if evt.is_writable() {
-                    svc.ready_write(api);
-                }
-            }
-
-            api.services().event_root().fire_all(api);
-        }
-    }
-
-    pub(super) fn add_service<'l>(&self, fleck: &'l Fleck) -> DefaultServiceBuilder<'l> {
-        let tok = Token(self.services.borrow().len() + 1);
-
-        builder::DefaultServiceBuilder::new(fleck, self.poll.clone(), tok)
-    }
-
-    /*pub(super) fn add_service<S: AbstractFdService>(&mut self, srv: S) {
-        let tok = Token(self.services.len() + 1);
-        self.services.insert(tok, Rc::new(srv));
-    }
-
-    pub(super) fn add_udp_service<CB: 'static + udp::UdpCallback>(&mut self, udp: mio::net::UdpSocket, on_read: CB) -> impl Fn(std::net::SocketAddr, Vec<u8>) {
-        let token = Token(self.services.len() + 1);
-        let srv = Rc::new(udp::UdpService {
-            socket: RefCell::new(udp),
-            token,
-            poll: self.poll.clone(),
-            send_queue: Default::default(),
-            write_registered: Cell::new(false),
-            read_callback: Rc::new(on_read),
-        });
-        self.services.insert(token, srv);
-
-        |addr, data| {
-
-        }
-        /*self.services.insert(token, Box::new(udp::UdpService::<Incoming, Outgoing, DataType> {
-            socket: RefCell::new(udp),
-            token,
-            poll: self.poll.clone(),
-            send_queue: Default::default(),
-            write_registered: std::cell::Cell::new(false),
-            _ghost: std::marker::PhantomData
-        }));*/
-    }*/
-}
-
-/*pub struct FleckIO {
-    poll: Rc<RefCell<mio::Poll>>,
-    services: HashMap<Token, Box<dyn AbstractFdServiceHelper>>,
-    // local: Vec<std::rc::Rc<dyn IOChannel>>,
-    // global: std::rc::Rc<SocketWrapperChannel>,
-
-    // minor: timerfd::TimerFd,
-    // major: timerfd::TimerFd,
-}*/
-
-/*
-struct SocketWrapperChannel {
-    poll: Rc<RefCell<mio::Poll>>,
-    socket: RefCell<mio::net::UdpSocket>,
-    queue: RefCell<VecDeque<(Vec<u8>, std::net::SocketAddr)>>,
-    write_registered: Cell<bool>,
-}
-
-impl SocketWrapperChannel {
-    fn new(poll: Rc<RefCell<mio::Poll>>, socket: mio::net::UdpSocket) -> Self {
-        Self {
-            poll,
-            socket: RefCell::new(socket),
-            queue: Default::default(),
-            write_registered: Cell::new(false),
-        }
-    }
-
-    fn read_packets(&self, feedback: &dyn IOFeedback) {
-    }
-
-    fn write_packets(&self) {
-        let mut queue = self.queue.borrow_mut();
-        while !queue.is_empty() {
-            let pkt = queue.front().unwrap();
-            match self.socket.borrow().send_to(&pkt.0, pkt.1) {
-                Ok(len) => {
-                    if len != pkt.0.len() {
-                        log::warn!(
-                            "Packet of {} bytes truncated to {} bytes when sending",
-                            pkt.0.len(),
-                            len
-                        );
-                    }
-                    queue.pop_front();
-                }
-                Err(e) => {
-                    log::info!("Error while sending packet: {:?}", e);
-                    break;
-                }
-            }
-        }
-
-        if queue.is_empty() {
-            self.poll
-                .borrow()
-                .registry()
-                .reregister(
-                    self.socket.borrow_mut().deref_mut(),
-                    FleckIO::GLOBAL,
-                    Interest::READABLE,
-                )
-                .expect("couldn't update interest?");
-            self.write_registered.set(false);
-        }
-    }
-}
-*/
-
-/*impl IOChannel for SocketWrapperChannel {
-    fn send_packet(&self, packet: &mut Packet) {
-        assert!(packet.data.is_some());
-        assert!(packet.addr.is_some());
-
-        self.queue
-            .borrow_mut()
-            .push_back((packet.data.clone().unwrap(), packet.addr.unwrap()));
-        if !self.write_registered.get() {
-            self.write_registered.set(true);
-            self.poll
-                .borrow()
-                .registry()
-                .reregister(
-                    self.socket.borrow_mut().deref_mut(),
-                    FleckIO::GLOBAL,
-                    Interest::READABLE.add(Interest::WRITABLE),
-                )
-                .expect("couldn't update interest?");
-        }
-    }
-}*/
-
-/*impl Default for FleckIO {
-    fn default() -> Self {
-        let poll = Poll::new().expect("couldn't create mio::Poll");
-
-        let poll = Rc::new(RefCell::new(poll));
-
-        Self {
-            poll,
-            ..Default::default()
-        }
-    }
-}
-*/
-
-/*
-impl FleckIO {
-    fn handle_event(&self, api: &Fleck, event: &mio::event::Event) {
-        /*match event.token() {
-            Self::GLOBAL => {
-                if event.is_readable() {
-                    self.global.read_packets(feedback);
-                }
-                if event.is_writable() {
-                    self.global.write_packets();
-                }
-            }
-            Self::MINOR_TICK => {
-                assert!(event.is_readable());
-                self.minor.read();
-                feedback.minor_tick();
-            }
-            Self::MAJOR_TICK => {
-                assert!(event.is_readable());
-                self.major.read();
-                feedback.major_tick();
-            }
-            _ => {
-                println!("unknown token!");
-            }
-        }*/
-    }
-
-    pub(crate) fn poll(&self, api: &Fleck) {
-        let mut events = Events::with_capacity(128);
-        loop {
-            let pr = self.poll.borrow_mut().poll(&mut events, None);
-            match pr {
-                Ok(()) => {
-                    for evt in &events {
-                        self.handle_event(api, evt);
-                    }
-                }
-                Err(_e) => {}
-            }
-        }
-    }
-
-    /*pub(crate) fn local(&self) -> std::rc::Rc<dyn IOChannel> {
-        // HACK: for now we just use the global
-        self.global.clone()
-    }
-
-    pub(crate) fn global(&self) -> std::rc::Rc<dyn IOChannel> {
-        self.global.clone()
-    }*/
-
-    const GLOBAL: Token = Token(0);
-    const MAJOR_TICK: Token = Token(1);
-    const MINOR_TICK: Token = Token(2);
-}
-*/
-
-/*
-
-       // minor tick registration
-       let mut minor = timerfd::TimerFd::new().unwrap();
-       minor.set_state(
-           timerfd::TimerState::Periodic {
-               current: std::time::Duration::new(1, 0),
-               interval: std::time::Duration::new(1, 0),
-           },
-           timerfd::SetTimeFlags::Default,
-       );
-       poll.registry()
-           .register(
-               &mut SourceFd(&minor.as_raw_fd()),
-               Self::MINOR_TICK,
-               Interest::READABLE,
-           )
-           .expect("couldn't register read interest for minor timer?");
-
-       // major tick registration
-       let mut major = timerfd::TimerFd::new().unwrap();
-       major.set_state(
-           timerfd::TimerState::Periodic {
-               current: std::time::Duration::new(1, 0),
-               interval: std::time::Duration::new(15, 0),
-           },
-           timerfd::SetTimeFlags::Default,
-       );
-       poll.registry()
-           .register(
-               &mut SourceFd(&major.as_raw_fd()),
-               Self::MAJOR_TICK,
-               Interest::READABLE,
-           )
-           .expect("couldn't register read interest for major timer?");
-
-       // global socket
-       let mut global_socket = mio::net::UdpSocket::bind(
-           std::net::SocketAddrV4::new(std::net::Ipv4Addr::UNSPECIFIED, DEFAULT_PORT).into(),
-       )
-       .expect("couldn't listen on UDP port?");
-       global_socket
-           .join_multicast_v4(&MULTICAST_ADDRESS.ip(), &std::net::Ipv4Addr::UNSPECIFIED)
-           .expect("couldn't join multicast group?");
-
-       poll.registry()
-           .register(&mut global_socket, Self::GLOBAL, Interest::READABLE)
-           .expect("couldn't register read interest in UDP port?");
-*
-* */

fleck/src/io/builder.rs

@@ -1,459 +0,0 @@
-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};
-
-use crate::{prelude::{ChannelSpec, Service}, service::event::NoPriorityTag};
-use super::AbstractFdService;
-
-pub type NoChannel = ((), ());
-
-pub struct MulticastSpec {
-    group: std::net::IpAddr,
-    iface: std::net::IpAddr,
-}
-
-pub struct NoData;
-
-pub struct UdpData {
-    bind_addr: Option<std::net::SocketAddr>,
-    join_multicast: Option<MulticastSpec>,
-}
-
-pub struct StreamData {
-    fd: Option<RawFd>,
-}
-
-pub struct TimerData {
-    interval: Option<std::time::Duration>,
-}
-
-struct InterestRegistration {
-    poll: Rc<RefCell<mio::Poll>>,
-    write: Cell<bool>,
-    token: Token,
-}
-
-impl InterestRegistration {
-    fn subscribe_read(&self, source: &mut dyn mio::event::Source) {
-        self.poll.borrow_mut().registry().register(
-            source, self.token, mio::Interest::READABLE,
-        ).expect("couldn't subscribe?")
-    }
-
-    fn subscribe_write(&self, source: &mut dyn mio::event::Source) {
-        if self.write.get() { return }
-        self.poll.borrow_mut().registry().reregister(
-            source, self.token, mio::Interest::READABLE.add(mio::Interest::WRITABLE),
-        ).expect("couldn't subscribe?");
-        self.write.set(true);
-    }
-
-    fn unsubscribe_write(&self, source: &mut dyn mio::event::Source) {
-        self.poll.borrow_mut().registry().reregister(
-            source, self.token, mio::Interest::READABLE,
-        ).expect("couldn't subscribe?");
-        self.write.set(false);
-    }
-}
-
-pub type DefaultServiceBuilder<'l> = ServiceBuilder<'l, NoChannel, NoChannel, NoData>;
-
-// Source: where the service writes packets to (recv)
-// Sink: where the service reads packets from (send)
-pub struct ServiceBuilder<'l, SourceChannel: ChannelSpec, SinkChannel: ChannelSpec, Data> {
-    fleck: &'l crate::Fleck,
-    poll: Rc<RefCell<mio::Poll>>,
-    token: Token,
-
-    sink_priority: Option<SinkChannel::Priority>,
-
-    data: Data,
-
-    _ghost: std::marker::PhantomData<(SourceChannel, SinkChannel)>,
-}
-
-impl<'l, SourceChannel: ChannelSpec, SinkChannel: ChannelSpec>
-    ServiceBuilder<'l, SourceChannel, SinkChannel, NoData>
-{
-    pub(super) fn new(fleck: &'l crate::Fleck, poll: Rc<RefCell<mio::Poll>>, token: Token) -> Self {
-        Self {
-            fleck,
-            poll,
-            token,
-
-            sink_priority: None,
-
-            data: NoData,
-
-            _ghost: Default::default(),
-        }
-    }
-}
-
-impl<'l, SinkChannel: ChannelSpec, Data> ServiceBuilder<'l, NoChannel, SinkChannel, Data> {
-    pub fn source<SourceChannel: ChannelSpec<Priority = NoPriorityTag>>(
-        self,
-    ) -> ServiceBuilder<'l, SourceChannel, SinkChannel, Data> {
-        ServiceBuilder {
-            fleck: self.fleck,
-            poll: self.poll,
-            token: self.token,
-
-            sink_priority: self.sink_priority,
-
-            data: self.data,
-
-            _ghost: Default::default(),
-        }
-    }
-
-    pub fn source_with<SourceChannel: ChannelSpec>(
-        self,
-        p: SourceChannel::Priority,
-    ) -> ServiceBuilder<'l, SourceChannel, SinkChannel, Data>
-    where SourceChannel::Priority: PartialOrd
-    {
-        ServiceBuilder {
-            fleck: self.fleck,
-            poll: self.poll,
-            token: self.token,
-
-            sink_priority: self.sink_priority,
-
-            data: self.data,
-
-            _ghost: Default::default(),
-        }
-    }
-}
-
-impl<'l, SourceChannel: ChannelSpec, Data> ServiceBuilder<'l, SourceChannel, NoChannel, Data> {
-    pub fn sink<SinkChannel: ChannelSpec<Priority = NoPriorityTag>>(
-        self,
-    ) -> ServiceBuilder<'l, SourceChannel, SinkChannel, Data> {
-        ServiceBuilder {
-            fleck: self.fleck,
-            poll: self.poll,
-            token: self.token,
-
-            sink_priority: Some(NoPriorityTag),
-
-            data: self.data,
-
-            _ghost: Default::default(),
-        }
-    }
-
-    pub fn sink_with<SinkChannel: ChannelSpec>(
-        self,
-        p: SinkChannel::Priority,
-    ) -> ServiceBuilder<'l, SourceChannel, SinkChannel, Data>
-    where SinkChannel::Priority: PartialOrd {
-        ServiceBuilder {
-            fleck: self.fleck,
-            poll: self.poll,
-            token: self.token,
-
-            sink_priority: Some(p),
-
-            data: self.data,
-
-            _ghost: Default::default(),
-        }
-    }
-}
-
-impl<'l,
-        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> {
-        ServiceBuilder {
-            fleck: self.fleck,
-            poll: self.poll,
-            token: self.token,
-
-            sink_priority: self.sink_priority,
-
-            data: UdpData {
-                bind_addr: None,
-                join_multicast: None,
-            },
-
-            _ghost: Default::default(),
-        }
-    }
-}
-
-impl<'l,
-        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 {
-        self.data.bind_addr = Some(bind_addr);
-        self
-    }
-
-    pub fn set_multicast_group(
-        mut self,
-        group_addr: std::net::IpAddr,
-        interface_addr: std::net::IpAddr,
-    ) -> Self {
-        self.data.join_multicast = Some(MulticastSpec {
-            group: group_addr,
-            iface: interface_addr,
-        });
-        self
-    }
-
-    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 */
-impl<'l, SourceChannel: ChannelSpec<Data = ()>>
-    ServiceBuilder<'l, SourceChannel, ((), ()), NoData>
-{
-    pub fn timer(self) -> ServiceBuilder<'l, SourceChannel, ((), ()), TimerData> {
-        ServiceBuilder {
-            fleck: self.fleck,
-            poll: self.poll,
-            token: self.token,
-
-            sink_priority: self.sink_priority,
-
-            data: TimerData { interval: None },
-
-            _ghost: Default::default(),
-        }
-    }
-}
-
-struct Timer<Tag: 'static> {
-    token: Token,
-    timer: timerfd::TimerFd,
-    _ghost: std::marker::PhantomData<Tag>,
-}
-
-impl<Tag: 'static> Service for Rc<Timer<Tag>> {}
-impl<Tag: 'static> AbstractFdService for Timer<Tag> {
-    fn ready_read(&self, api: &crate::Fleck) {
-        self.timer.read();
-        api.queue::<(Tag, ())>(());
-    }
-
-    fn ready_write(&self, _api: &crate::Fleck) {
-        unreachable!();
-    }
-
-    fn token(&self) -> Token {
-        self.token
-    }
-}
-
-impl<'l, SourceChannel: ChannelSpec<Data = ()>>
-    ServiceBuilder<'l, SourceChannel, ((), ()), TimerData>
-{
-    pub fn interval(mut self, time: std::time::Duration) -> Self {
-        self.data.interval = Some(time);
-        self
-    }
-
-    pub fn build(self) {
-        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: raw_timer, _ghost: Default::default() });
-
-        self.poll.borrow().registry().register(
-            &mut SourceFd(&timer.timer.as_raw_fd()),
-            self.token,
-            mio::Interest::READABLE
-        ).expect("couldn't register timer read interest?");
-
-        self.fleck.io.borrow_mut().services.borrow_mut().insert(self.token, timer);
-    }
-}
-
-/* stream support */
-impl<'l, SourceChannel: ChannelSpec<Data = Vec<u8>>, SinkChannel: ChannelSpec<Data = Vec<u8>>>
-    ServiceBuilder<'l, SourceChannel, SinkChannel, NoData>
-{
-    pub fn stream(self) -> ServiceBuilder<'l, SourceChannel, SinkChannel, StreamData> {
-        ServiceBuilder {
-            fleck: self.fleck,
-            poll: self.poll,
-            token: self.token,
-
-            sink_priority: self.sink_priority,
-
-            data: StreamData { fd: None },
-
-            _ghost: Default::default(),
-        }
-    }
-}
-
-impl<'l, SourceChannel: ChannelSpec<Data = Vec<u8>>, SinkChannel: ChannelSpec<Data = Vec<u8>>>
-    ServiceBuilder<'l, SourceChannel, SinkChannel, StreamData>
-{
-    pub fn fd(mut self, fd: RawFd) -> Self {
-        self.data.fd = Some(fd);
-        self
-    }
-}
-
-impl<'l, SourceChannel: ChannelSpec<Data = Vec<u8>>, SinkTag: 'static, SinkPriority: 'static + Clone + PartialOrd>
-    ServiceBuilder<'l, SourceChannel, (SinkTag, Vec<u8>, SinkPriority), StreamData>
-{
-    pub fn build(self) {
-        let svc = Rc::new(StreamService::<SourceChannel, (SinkTag, Vec<u8>, SinkPriority)> {
-            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?")) }),
-            sink_priority: RefCell::new(Some(self.sink_priority.expect("no sink priority given!"))),
-            _ghost: Default::default(),
-        });
-
-        svc.interest.subscribe_read(&mut SourceFd(&svc.file.borrow().as_raw_fd()));
-
-        self.fleck.io.borrow_mut().services.borrow_mut().insert(self.token, svc.clone());
-        self.fleck.services.borrow_mut().add_service(svc.clone());
-    }
-}
-
-impl<'l, SourceChannel: ChannelSpec<Data = Vec<u8>>, SinkTag: 'static>
-    ServiceBuilder<'l, SourceChannel, (SinkTag, Vec<u8>), StreamData>
-{
-    pub fn build(self) {
-        let svc = Rc::new(StreamService::<SourceChannel, (SinkTag, Vec<u8>)> {
-            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?")) }),
-            sink_priority: RefCell::new(Some(self.sink_priority.expect("no sink priority given!"))),
-            _ghost: Default::default(),
-        });
-
-        svc.interest.subscribe_read(&mut SourceFd(&svc.file.borrow().as_raw_fd()));
-
-        self.fleck.io.borrow_mut().services.borrow_mut().insert(self.token, svc.clone());
-        self.fleck.services.borrow_mut().add_service(svc.clone());
-    }
-}
-
-pub struct StreamService<SourceChannel: ChannelSpec<Data = Vec<u8>>, SinkChannel: ChannelSpec<Data = Vec<u8>>> {
-    interest: InterestRegistration,
-    queue: RefCell<VecDeque<u8>>,
-    file: RefCell<std::fs::File>,
-    sink_priority: RefCell<Option<SinkChannel::Priority>>,
-    _ghost: std::marker::PhantomData<(SourceChannel, SinkChannel)>,
-}
-
-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);
-    }
-}
-
-impl<SourceChannel: ChannelSpec<Data = Vec<u8>>, SinkTag: 'static> Service for Rc<StreamService<SourceChannel, (SinkTag, Vec<u8>)>> {
-    fn register_channels(&self, eroot: &mut crate::prelude::EventRoot) {
-        eroot.channel::<(SinkTag, Vec<u8>)>().sub_opt(self, StreamService::sink_data);
-    }
-}
-
-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>> {
-        self.queue.borrow_mut().extend(data.drain(..));
-        self.interest.subscribe_write(&mut SourceFd(&self.file.borrow().as_raw_fd()));
-        None
-    }
-}
-
-impl<SourceChannel: ChannelSpec<Data = Vec<u8>>, SinkChannel: ChannelSpec<Data = Vec<u8>>> AbstractFdService for StreamService<SourceChannel, SinkChannel> {
-    fn ready_read(&self, api: &crate::Fleck) {
-        let mut buf = [0u8; 1024];
-        while let Ok(bytes) = self.file.borrow_mut().read(&mut buf) {
-            api.queue::<SourceChannel>(Vec::from(&buf[..bytes]));
-        }
-
-    }
-    fn ready_write(&self, _api: &crate::Fleck) {
-        let mut queue = self.queue.borrow_mut();
-        while !queue.is_empty() {
-            let (data, _) = queue.as_slices();
-            match self.file.borrow_mut().write(data) {
-                Ok(len) => {
-                    queue.drain(..len);
-                },
-                Err(_) => {
-                    return
-                }
-            }
-        }
-
-        self.interest.unsubscribe_write(&mut SourceFd(&self.file.borrow().as_raw_fd()));
-    }
-    fn token(&self) -> Token {
-        self.interest.token
-    }
-}

fleck/src/io/stream.rs

@@ -1,128 +0,0 @@
-use std::cell::Cell;
-use std::cell::RefCell;
-use std::collections::VecDeque;
-use std::fs::File;
-use std::io::Read;
-use std::io::Write;
-use std::os::unix::prelude::AsRawFd;
-use std::rc::Rc;
-
-use mio::unix::SourceFd;
-use mio::Interest;
-use mio::Token;
-
-use super::AbstractFdService;
-use crate::prelude::*;
-use crate::service::event::NoPriorityTag;
-use crate::Fleck;
-
-pub trait StreamData: 'static + Into<Vec<u8>> + From<Vec<u8>> {}
-pub struct StreamService<
-    Incoming: ChannelSpec<Priority = NoPriorityTag, Data = DataType>,
-    Outgoing: ChannelSpec<Data = DataType>,
-    DataType: StreamData,
-> {
-    stream: RefCell<File>,
-    token: Token,
-    poll: Rc<RefCell<mio::Poll>>,
-    send_queue: RefCell<VecDeque<u8>>,
-    write_registered: Cell<bool>,
-    _ghost: std::marker::PhantomData<(Incoming, Outgoing, DataType)>,
-}
-
-impl<
-        Incoming: ChannelSpec<Priority = NoPriorityTag, Data = DataType>,
-        Outgoing: ChannelSpec<Priority = NoPriorityTag, Data = DataType>,
-        DataType: StreamData,
-    > Service for Rc<StreamService<Incoming, Outgoing, DataType>>
-{
-    fn register_channels(&self, eroot: &mut crate::prelude::EventRoot) {
-        eroot
-            .channel::<Outgoing>()
-            .sub_opt(self, StreamService::handle_outgoing);
-    }
-}
-
-impl<
-        Incoming: ChannelSpec<Priority = NoPriorityTag, Data = DataType>,
-        Outgoing: ChannelSpec<Priority = NoPriorityTag, Data = DataType>,
-        DataType: StreamData,
-    > StreamService<Incoming, Outgoing, DataType>
-{
-    fn handle_outgoing(&self, _api: &Fleck, packet: DataType) -> Option<DataType> {
-        self.send_queue.borrow_mut().extend(packet.into().iter());
-
-        self.register_write_interest();
-
-        None
-    }
-
-    fn clear_write_interest(&self) {
-        self.poll
-            .borrow()
-            .registry()
-            .reregister(
-                &mut SourceFd(&self.stream.borrow().as_raw_fd()),
-                self.token,
-                Interest::READABLE,
-            )
-            .expect("couldn't update interest?");
-        self.write_registered.set(false);
-    }
-
-    fn register_write_interest(&self) {
-        if self.write_registered.get() {
-            return;
-        }
-        self.poll
-            .borrow()
-            .registry()
-            .reregister(
-                &mut SourceFd(&self.stream.borrow().as_raw_fd()),
-                self.token,
-                Interest::READABLE.add(Interest::WRITABLE),
-            )
-            .expect("couldn't update interest?");
-        self.write_registered.set(true);
-    }
-}
-
-impl<
-        Incoming: ChannelSpec<Priority = NoPriorityTag, Data = DataType>,
-        Outgoing: ChannelSpec<Priority = NoPriorityTag, Data = DataType>,
-        DataType: StreamData,
-    > AbstractFdService for StreamService<Incoming, Outgoing, DataType>
-{
-    fn ready_read(&self, api: &Fleck) {
-        let mut buf = [0u8; 2048];
-        while let Ok(len) = self.stream.borrow_mut().read(&mut buf) {
-            api.queue::<Incoming>(Vec::from(&buf[..len]).into());
-        }
-    }
-
-    fn ready_write(&self, _api: &Fleck) {
-        let mut queue = self.send_queue.borrow_mut();
-        while !queue.is_empty() {
-            let pkt = queue.front_mut().unwrap();
-
-            let (data, _) = queue.as_slices();
-
-            match self.stream.borrow_mut().write(data) {
-                Ok(len) => {
-                    queue.drain(0..len);
-                }
-                Err(e) => {
-                    break;
-                }
-            }
-        }
-
-        if queue.is_empty() {
-            self.clear_write_interest()
-        }
-    }
-
-    fn token(&self) -> Token {
-        self.token
-    }
-}

fleck/src/io/udp.rs

@@ -1,110 +0,0 @@
-use std::cell::{Cell, RefCell};
-use std::collections::VecDeque;
-use std::net::SocketAddr;
-use std::ops::DerefMut;
-use std::rc::Rc;
-
-use mio::net::UdpSocket;
-use mio::Interest;
-use mio::Token;
-
-use super::AbstractFdService;
-use crate::Fleck;
-
-pub trait UdpCallback {
-    fn invoke(&self, addr: SocketAddr, data: Vec<u8>);
-}
-
-impl<T: Fn(SocketAddr, Vec<u8>)> UdpCallback for T {
-    fn invoke(&self, addr: SocketAddr, data: Vec<u8>) {
-        (self)(addr, data);
-    }
-}
-
-pub struct UdpService {
-    pub(super) socket: RefCell<UdpSocket>,
-    pub(super) token: Token,
-    pub(super) poll: Rc<RefCell<mio::Poll>>,
-    pub(super) send_queue: RefCell<VecDeque<(SocketAddr, Vec<u8>)>>,
-    pub(super) write_registered: Cell<bool>,
-    pub(super) read_callback: Rc<dyn UdpCallback>,
-}
-
-impl UdpService {
-    fn clear_write_interest(&self) {
-        self.poll
-            .borrow()
-            .registry()
-            .reregister(
-                self.socket.borrow_mut().deref_mut(),
-                self.token,
-                Interest::READABLE,
-            )
-            .expect("couldn't update interest?");
-        self.write_registered.set(false);
-    }
-
-    fn register_write_interest(&self) {
-        if self.write_registered.get() {
-            return;
-        }
-        self.poll
-            .borrow()
-            .registry()
-            .reregister(
-                self.socket.borrow_mut().deref_mut(),
-                self.token,
-                Interest::READABLE.add(Interest::WRITABLE),
-            )
-            .expect("couldn't update interest?");
-        self.write_registered.set(true);
-    }
-
-    /*fn handle_outgoing(&self, _api: &Fleck, packet: DataType) -> Option<DataType> {
-        self.send_queue.borrow_mut().push_back(packet.into());
-
-        self.register_write_interest();
-
-        None
-    }*/
-}
-
-impl AbstractFdService for UdpService {
-    fn ready_read(&self, _api: &Fleck) {
-        let mut buf = [0u8; 2048];
-        while let Ok((len, addr)) = self.socket.borrow_mut().recv_from(&mut buf) {
-            (self.read_callback)(addr, Vec::from(&buf[..len]));
-        }
-    }
-
-    fn ready_write(&self, _api: &Fleck) {
-        let mut queue = self.send_queue.borrow_mut();
-        while !queue.is_empty() {
-            let pkt = queue.front().unwrap();
-            match self.socket.borrow_mut().send_to(&pkt.1, pkt.0) {
-                Ok(len) => {
-                    if len != pkt.1.len() {
-                        log::warn!(
-                            "Packet of {} bytes truncated to {} bytes when sending",
-                            pkt.1.len(),
-                            len
-                        );
-                    }
-                    queue.pop_front();
-                }
-                Err(e) => {
-                    log::info!("Error while sending packet: {:?}", e);
-                    break;
-                }
-            }
-        }
-
-        if queue.is_empty() {
-            self.clear_write_interest()
-        }
-    }
-
-    fn token(&self) -> Token {
-        self.token
-    }
-}

fleck/src/lib.rs

@@ -1,17 +1,12 @@
-use std::cell::RefCell;
-use std::collections::HashMap;
-use std::os::unix::prelude::{AsRawFd, OwnedFd};
 use std::rc::Rc;
 
 // mod crypto;
 pub mod fleck_core;
 mod helper;
-// mod io;
 pub mod service;
 
 pub mod prelude {
     pub use crate::helper::{AsAny, IntoWeak};
-    // pub use crate::fleck_core::msg::{Message, MessageParams};
     pub use crate::service::{ChannelSpec, EventRoot, Service};
     pub use crate::fleck_core;
 }
@@ -25,34 +20,15 @@ pub struct Fleck {
 impl Fleck {
     pub fn new() -> Rc<Self> {
         let res = Rc::new(Self {
-            // io: RefCell::new(io::IO::new()),
             services: Default::default(),
         });
 
         res.add_service::<fleck_core::io::IOService>();
         res.add_service::<fleck_core::CoreInitService>();
 
-        // res.register_core_services();
-        // res.timer_setup();
-
         res
     }
 
-    /*fn timer_setup(&self) {
-        // minor tick registration
-        self.add_io()
-            .source::<(fleck_core::MinorTickChannel, ())>()
-            .timer()
-            .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();
-    }*/
-
     fn register_core_services(&self) {
         // core I/O service
         // initial outgoing/incoming/minor/major channels
@@ -85,19 +61,23 @@ impl Fleck {
     {
         self.services.give_service::<S>(self);
     }
+    /// Access a pre-existing service
     pub fn with_service<S: 'static, R, F: FnOnce(&S) -> R>(&self, f: F) -> R
     where
         Rc<S>: Service,
     {
         self.services.with_service(f)
     }
+    /// Create an event channel from a specification
     pub fn create_channel<CS: service::ChannelSpec>(&self) -> Rc<service::event::Channel<CS::Tag, Self, CS::Data, CS::Priority>> {
         self.services.event_root().create_channel::<CS>();
         self.channel::<CS>()
     }
+    /// Get an existing channel
     pub fn channel<CS: service::ChannelSpec>(&self) -> Rc<service::event::Channel<CS::Tag, Self, CS::Data, CS::Priority>> {
         self.services.event_root().channel::<CS>()
     }
+    /// Queue a message onto a channel
     pub fn queue<CS: service::ChannelSpec>(&self, data: CS::Data) {
         self.services
             .event_root()

fleck/src/service.rs

@@ -79,31 +79,3 @@ impl ServiceStack {
         &self.message_registry
     }
 }
-
-/*
-impl ServiceStack {
-    pub(crate) fn process_incoming(&self, api: &crate::Fleck, packet: Packet) {
-        self.eroot
-            .channel::<fleck_core::ReceivePacketChannel>()
-            .queue(packet);
-        self.eroot.fire_all(api);
-    }
-
-    pub(crate) fn process_minor_tick(&self, api: &crate::Fleck) {
-        self.eroot.channel::<fleck_core::MinorTickChannel>().queue(());
-        self.eroot.fire_all(api);
-    }
-
-    pub(crate) fn process_major_tick(&self, api: &crate::Fleck) {
-        self.eroot.channel::<fleck_core::MajorTickChannel>().queue(());
-        self.eroot.fire_all(api);
-    }
-
-    pub(crate) fn create_io_channels(&mut self) {
-        self.eroot.create_channel::<fleck_core::SendPacketChannel>();
-        self.eroot.create_channel::<fleck_core::ReceivePacketChannel>();
-        self.eroot.create_channel::<fleck_core::MinorTickChannel>();
-        self.eroot.create_channel::<fleck_core::MajorTickChannel>();
-    }
-}
-*/

fleck/src/service/event.rs

@@ -48,8 +48,6 @@ impl<Host: 'static, Context: 'static + ?Sized, Data: 'static> EventSub<Context,
 }
 
 pub struct Channel<Tag: 'static, Context: 'static + ?Sized, Data: 'static, Priority: 'static> {
-    /*subs: RefCell<super::priority::TotalOrder<Rc<dyn EventSub<Context, Data>>>>,
-    order_cache: RefCell<Option<Vec<Rc<dyn EventSub<Context, Data>>>>>,*/
     subs: RefCell<Vec<(Priority, Rc<dyn EventSub<Context, Data>>)>>,
 
     queue: RefCell<VecDeque<Data>>,
@@ -236,8 +234,6 @@ impl<Context: 'static + ?Sized> Default for EventRoot<Context> {
     }
 }
 
-// pub type ChannelRef<'a, CS: ChannelSpec, Context> = crate::helper::OwnedRef<'a, HashMap<std::any::TypeId, Rc<dyn std::any::Any>>, Channel<CS::Tag, Context, CS::Data, CS::Priority>>;
-
 impl<Context: 'static + ?Sized> EventRoot<Context> {
     pub fn create_channel<CS: ChannelSpec>(&self) {
         let tid = std::any::TypeId::of::<Channel<CS::Tag, Context, CS::Data, CS::Priority>>();