server.rs

pubsub.rs

#[derive(Debug)]
pub enum Command {
    Publish(String),
    Retrieve
}

#[derive(Debug)]
pub enum Error {
    VerbMissing,
    UnknownVerb,
    UnexpectedPayload,
    MissingEOL,
    MissingPayload,
}

pub fn parse(input: &str) -> Result<Command, Error> {
    let mut input = input;
    if let Some(pos) = input.find('\n') {
        input = &input[0..pos]
    } else {
        return Err(Error::MissingEOL)
    }

    let words: Vec<&str> = input.splitn(2, ' ').collect();
    match words[0] {
        "" => Err(Error::VerbMissing),
        "PUBLISH" => {
            if words.len() > 1 {
                Ok(Command::Publish(String::from(words[1])))
            } else {
                Err(Error::MissingPayload)
            }
        },
        "RETRIEVE" => {
            if words.len() > 1 {
                Err(Error::UnexpectedPayload)
            } else {
                Ok(Command::Retrieve)
            }
        },
        _ => Err(Error::UnknownVerb)
    }

    //let mut words = input.splitn(2, ' ');
    //match (words.next(), words.next()) {
        //(Some(""), _) => Err(Error::VerbMissing),
        //(Some("PUBLISH"), Some(thing_else)) => Ok(Command::Publish(String::from(thing_else))),
        //(Some("RETRIEVE"), None) => Ok(Command::Retrieve),
        //_ => Err(Error::UnknownVerb)
    //}
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn fail_missing_eol() {
        assert!(match parse("") {
            Err(Error::MissingEOL) => true,
            _ => false
        })
    }

    #[test]
    fn fail_missing_verb() {
        assert!(match parse("\n") {
            Err(Error::VerbMissing) => true,
            _ => false
        })
    }

    #[test]
    fn fail_unknown_verb() {
        assert!(match parse("JEEJ\n") {
            Err(Error::UnknownVerb) => true,
            _ => false
        })
    }

    #[test]
    fn publish() {
        match parse("PUBLISH hello\n") {
            Ok(Command::Publish(payload)) => assert_eq!(payload, "hello"),
            _ => panic!()
        }
    }

    #[test]
    fn fail_publish_nospace() {
        assert!(match parse("PUBLISH\n") {
            Err(Error::MissingPayload) => true,
            _ => false
        });
    }

    #[test]
    fn publish_several_words() {
        match parse("PUBLISH hello world\n") {
            Ok(Command::Publish(payload)) => assert_eq!(payload, "hello world"),
            _ => panic!(false)
        }
    }

    #[test]
    fn publish_empty() {
        match parse("PUBLISH \n") {
            Ok(Command::Publish(payload)) => assert_eq!(payload, ""),
            _ => panic!()
        }
    }

    #[test]
    fn retrieve() {
        assert!(match parse("RETRIEVE\n") {
            Ok(Command::Retrieve) => true,
            _ => false
        })
    }

    #[test]
    fn fail_retrieve_unknown() {
        assert!(match parse("RETRIEVE word\n") {
            Err(Error::UnexpectedPayload) => true,
            _ => false
        })
    }
}

run.sh

#!/bin/bash

while true;
do
    echo "PUBLISH $1" | nc 127.0.0.1 1337
    echo "RETRIEVE" | nc 127.0.0.1 1337
done

server.rs

use std::io::prelude::*;
use std::net::{TcpListener, TcpStream};
use std::collections::VecDeque;

use std::thread;
use std::sync::{Arc, Mutex};

extern crate redisish;

use redisish::{ parse, Error, Command };

type StringQueue = VecDeque<String>;
type SharedStringQueue = Arc<Mutex<StringQueue>>;

fn create_shared_string_q() -> SharedStringQueue {
    Arc::new(Mutex::new(StringQueue::new()))
}

//impl SharedStringQueue {
    //fn new() -> Self {
        //Arc::new(Mutex::new(StringQueue::new())
    //}
//}

fn handle_client(stream: &mut TcpStream, queue: &SharedStringQueue) {
    println!("Handling client");

    let mut buf = String::new();

    let read_chars = stream.read_to_string(&mut buf).unwrap();
    println!("Read {} bytes", read_chars);

    let mut result = match parse(&buf) {
        Ok(command) => {
            match command {
                Command::Publish(payload) => {
                    println!("The socket said: {}", payload);
                    {
                        let mut q = queue.lock().unwrap();
                        q.push_back(payload);
                    }
                    String::from("great success!")
                },
                Command::Retrieve => {
                    let last_msg = {
                        let mut q = queue.lock().unwrap();
                        q.pop_front()
                    };
                    match last_msg {
                        Some(head) => head,
                        None => String::from("ERROR: nothing to retrieve!")
                    }
                }
            }
        },

        Err(err) => {
            String::from(match err {
                Error::VerbMissing => "Verb is missing",
                Error::UnknownVerb => "This verb is unknown",
                Error::UnexpectedPayload => "Unexpected payload for RETRIEVE",
                Error::MissingEOL => "Missing an EOL",
                Error::MissingPayload => "Missing payload for PUBLISH"
            })
        }
    };

    result = result + "\n";
    stream.write(&result.into_bytes()).unwrap();
}

fn main() {
    let mut args = std::env::args();

    // Pop program name
    args.next();

    let host = args.next().unwrap_or("127.0.0.1".to_string());
    let port = args.next().unwrap_or("1337".to_string());
    let binding = format!("{}:{}", host, port);

    println!("Starting server on {}...", binding);
    let listener = TcpListener::bind(binding).unwrap();

    let queue = create_shared_string_q();
    for stream in listener.incoming() {
        let mut clone = queue.clone();
        thread::spawn(move || {
            handle_client(&mut stream.unwrap(), &mut clone);
        }).join().unwrap();
    }
}