PixSpace/src/tracker.rs

use bevy::prelude::*;

extern crate bytes;
extern crate vrpn;

// use futures_lite::future;
use rand::Rng;
use std::time::{Duration, Instant};

use bytes::{Bytes, BytesMut};
use std::{
    io::{self, Read},
    net::{SocketAddr, TcpStream},
};

use vrpn::{
    codec::peek_u32,
    constants::MAGIC_DATA,
    cookie::check_ver_nonfile_compatible,
    handler::{HandlerCode, TypedHandler},
    message::MessageSize,
    sync_io::{read_cookie, write_cookie, EndpointSyncTcp},
    tracker::PoseReport,
    unbuffer, CookieData, Message, Result, SequencedGenericMessage, TypeDispatcher, Unbuffer,
};

#[derive(Debug)]
struct TrackerHandler {}
impl TypedHandler for TrackerHandler {
    type Item = PoseReport;
    fn handle_typed(&mut self, msg: &Message<PoseReport>) -> Result<HandlerCode> {
        println!("{:?}\n   {:?}", msg.header, msg.body);
        Ok(HandlerCode::ContinueProcessing)
    }
}

#[derive(Component)]
pub struct Tracker {
    stream: TcpStream, // endpoint: EndpointSyncTcp,
                       // dispatcher: TypeDispatcher
                       // stream: ConnectionIpStream,
                       // connection: Arc<ConnectionIp>
}

// #[derive(Debug)]
// struct TrackerHandler {}

// impl TypedHandler for TrackerHandler {
//     type Item = PoseReport;
//     fn handle_typed(&mut self, msg: &Message<PoseReport>) -> Result<HandlerCode> {
//         info!("{:?}\n   {:?}", msg.header, msg.body);
//         Ok(HandlerCode::ContinueProcessing)
//     }
// }

impl Tracker {
    pub fn from_info(server_ip: &str, port: u16) -> Self {
        // let server = format!("tcp://{}:{}", server_ip, port)
        //     .parse::<ServerInfo>()
        //     .unwrap();

        // let connection = ConnectionIp::new_client(server, None, None).expect("msg");

        // let sender = connection
        //     .register_sender(StaticSenderName(b"Tracker0"))
        //     .expect("should be able to register sender");

        // let ping_client = ping::Client::new(sender, Arc::clone(&connection))
        //     .expect("should be able to create ping client");

        let addr: SocketAddr = format!("{}:{}", server_ip.to_string(), port)
            .parse()
            .unwrap();

        let mut stream = TcpStream::connect(addr).unwrap();
        stream.set_nodelay(true).unwrap();

        // We first write our cookie, then read and check the server's cookie, before the loop.
        write_cookie(&mut stream, CookieData::from(MAGIC_DATA)).unwrap();
        let cookie_buf = read_cookie(&mut stream).unwrap();

        let mut cookie_buf = Bytes::from(&cookie_buf[..]);

        CookieData::unbuffer_ref(&mut cookie_buf)
            .and_then(|msg| check_ver_nonfile_compatible(msg.version))
            .unwrap();

        Tracker { stream }
    }

    pub fn update(&mut self) -> vrpn::Result<()> {
        let mut buf = BytesMut::new();

        // Read the message header and padding
        buf.resize(24, 0);
        self.stream.read_exact(buf.as_mut())?;

        // Peek the size field, to compute the MessageSize.

        // let buf = buf.freeze();

        let total_len = peek_u32(&buf.clone().freeze())?.unwrap();
        // let total_len = peek_u32(&buf)?.unwrap();
        let size = MessageSize::from_length_field(total_len);

        // Read the body of the message
        let mut body_buf = BytesMut::new();
        body_buf.resize(size.padded_body_size(), 0);
        self.stream.read_exact(body_buf.as_mut())?;

        // Combine the body with the header
        buf.extend_from_slice(&body_buf[..]);
        let mut buf = buf.clone().freeze();

        // eprintln!("{:?}", Message::from(unbuffered));
        // info!("{:?}", Message::from(unbuffered));

        let pose = PoseReport::unbuffer_ref(&mut body_buf.clone().freeze())?;
        info!("{:?}", pose);

        // Unbuffer the message.
        // let unbuffered = SequencedGenericMessage::unbuffer_ref(&mut buf)?;

        // let message = Message::from(unbuffered);
        // info!("{:?}",message.header);

        Ok(())
    }
}

pub fn setup_tracker(mut commands: Commands) {
    // commands.spawn(Tracker::from_info("127.0.0.1", 3883));
    commands.spawn(Tracker::from_info("212.201.64.122", 3883));
}

pub fn update_tracker(mut query: Query<&mut Tracker>) {
    for mut tracker in query.iter_mut() {
        tracker.update();

        // tracker.connection.poll_endpoints();
        // tracker
    }
}

pub fn setup_threaded_tracker(mut commands: Commands) {
    // let (tx, rx) = bounded::<u32>(10);
    std::thread::spawn(move || {
        let addr: SocketAddr = "127.0.0.1:3883".parse().unwrap();
        let mut stream = TcpStream::connect(addr).unwrap();
        stream.set_nodelay(true).unwrap();

        // We first write our cookie, then read and check the server's cookie, before the loop.
        write_cookie(&mut stream, CookieData::from(MAGIC_DATA)).unwrap();
        let cookie_buf = read_cookie(&mut stream).unwrap();
        let mut cookie_buf = Bytes::from(&cookie_buf[..]);

        CookieData::unbuffer_ref(&mut cookie_buf)
            .and_then(|msg| check_ver_nonfile_compatible(msg.version)).unwrap();

        let mut endpoint = EndpointSyncTcp::new(stream);

        let mut dispatcher = TypeDispatcher::new();
        let _ = dispatcher.add_typed_handler(Box::new(TrackerHandler {}), None).unwrap();

        loop {

            // vrpn implementation is rather verbose - we need to see what specific errors 
            // should be handled properly!
            endpoint.poll_endpoint(&mut dispatcher).unwrap_or_else(|err| {
                info!("error from tracker thread {:?}",err);
            })
        }
    });
}