Reworking application structure

Cargo.toml

@@ -7,6 +7,9 @@ edition = "2021"
 
 [dependencies]
 argh = "0.1.12"
+bincode = "1.3.3"
+crc32fast = "1.3.2"
 crossterm = "0.27.0"
 interprocess = "1.2.1"
 ratatui = "0.23.0"
+serde = "1.0.196"

src/client.rs

@@ -1,4 +1,3 @@
-use argh::FromArgs;
 use std::{error::Error, time::Duration};
 
 use crate::CliArgs;

src/client/app.rs

@@ -1,226 +1,56 @@
-use std::{ops::Deref, time::Duration};
+use std::time::Duration;
 
-const FILE_PATHS: [&str; 0] = [];
+use interprocess::local_socket::LocalSocketStream;
-
+use rmp::protocol::{Message, MessageType};
-#[derive(Debug, Default, Clone, Eq, PartialEq, Hash)]
-pub struct ListState {
-    offset: usize,
-    selected: Option<usize>,
-}
-
-impl ListState {
-    pub fn offset(&self) -> usize {
-        self.offset
-    }
-
-    pub fn offset_mut(&mut self) -> &mut usize {
-        &mut self.offset
-    }
-
-    pub fn with_selected(mut self, selected: Option<usize>) -> Self {
-        self.selected = selected;
-        self
-    }
-
-    pub fn with_offset(mut self, offset: usize) -> Self {
-        self.offset = offset;
-        self
-    }
-
-    pub fn selected(&self) -> Option<usize> {
-        self.selected
-    }
-
-    pub fn select(&mut self, index: Option<usize>) {
-        self.selected = index;
-        if index.is_none() {
-            self.offset = 0;
-        }
-    }
-}
-
-pub struct StatefulList<T> {
-    pub state: ListState,
-    pub items: Vec<T>,
-}
-
-impl<T> StatefulList<T>
-where
-    T: Deref + Eq,
-{
-    pub fn with_items(items: Vec<T>) -> StatefulList<T> {
-        StatefulList {
-            state: ListState::default(),
-            items,
-        }
-    }
-
-    pub fn next(&mut self) {
-        let i = match self.state.selected() {
-            Some(i) => {
-                if i >= self.items.len() - 1 {
-                    0
-                } else {
-                    i + 1
-                }
-            }
-            None => 0,
-        };
-        self.state.select(Some(i));
-    }
-
-    pub fn previous(&mut self) {
-        let i = match self.state.selected() {
-            Some(i) => {
-                if i == 0 {
-                    self.items.len() - 1
-                } else {
-                    i - 1
-                }
-            }
-            None => 0,
-        };
-        self.state.select(Some(i));
-    }
-
-    pub fn index_of(&self, item: T) -> Option<usize> {
-        self.items.iter().position(|element| *element == item)
-    }
-
-    pub fn current(&self) -> Option<&T> {
-        self.state
-            .selected
-            .map_or(None, |index| self.items.get(index))
-    }
-
-    pub fn current_mut(&mut self) -> Option<&mut T> {
-        self.state
-            .selected
-            .map_or(None, |index| self.items.get_mut(index))
-    }
-}
-
-// TODO: Implement this
-#[derive(Default, Clone)]
-pub struct TrackMetadata {
-    pub duration: Option<Duration>,
-}
-
-pub struct PlayerState {
-    pub currently_playing: Option<String>,
-    pub is_playing: bool,
-    pub time_ratio: f64,
-}
-
-impl Default for PlayerState {
-    fn default() -> Self {
-        Self {
-            currently_playing: None,
-            is_playing: false,
-            time_ratio: 0.0,
-        }
-    }
-}
-
-impl PlayerState {
-    pub fn play(&mut self, path: &str) {
-        self.currently_playing = Some(path.to_string());
-        self.is_playing = true;
-        self.time_ratio = 0.0;
-        // TODO: Handle audio
-    }
-
-    pub fn toggle_pause(&mut self) {
-        self.is_playing = !self.is_playing;
-    }
-}
 
 pub struct App {
+    pub socket: Option<LocalSocketStream>,
     pub title: String,
-    pub playlist: StatefulList<String>,
-    pub player_state: PlayerState,
     pub should_quit: bool,
     pub enhanced_graphics: bool,
 }
 
 impl App {
     pub fn new(title: &str, enhanced_graphics: bool) -> Self {
-        let mut playlist = StatefulList::<String> {
-            state: ListState::default(),
-            items: FILE_PATHS.iter().map(|path| path.to_string()).collect(),
-        };
-        playlist.state.selected = if playlist.items.len() > 0 {
-            Some(0)
-        } else {
-            None
-        };
         Self {
+            socket: None,
             title: title.to_string(),
-            playlist,
-            player_state: PlayerState::default(),
             should_quit: false,
             enhanced_graphics,
         }
     }
 
-    pub fn play_next(&mut self) {
+    pub fn connect(&mut self) -> Result<(), ()> {
-        let current = if let Some(currently_playing) = self.player_state.currently_playing.clone() {
+        let path = rmp::os::get_socket_path().map_err(|_| ())?;
-            if let Some(current_index) = self
+        let socket = LocalSocketStream::connect(path).map_err(|_| ())?;
-                .playlist
+        self.socket = Some(socket);
-                .items
+        Ok(())
-                .iter()
-                .position(|path| *path == currently_playing)
-            {
-                current_index
-            } else {
-                return;
-            }
-        } else {
-            return;
-        };
-        let track = if current >= self.playlist.items.len() - 1 {
-            self.playlist.items[0].as_str()
-        } else {
-            self.playlist.items[current + 1].as_str()
-        };
-        self.player_state.play(track);
     }
 
-    pub fn play_previous(&mut self) {
+    pub fn connected(&self) -> bool {
-        let current = if let Some(currently_playing) = self.player_state.currently_playing.clone() {
+        self.socket.is_some()
-            if let Some(current_index) = self
-                .playlist
-                .items
-                .iter()
-                .position(|path| *path == currently_playing)
-            {
-                current_index
-            } else {
-                return;
-            }
-        } else {
-            return;
-        };
-        let track = if current <= 0 {
-            self.playlist.items[self.playlist.items.len() - 1].as_str()
-        } else {
-            self.playlist.items[current - 1].as_str()
-        };
-        self.player_state.play(track);
     }
 
-    pub fn replay_current(&mut self) {
+    pub fn toggle_shuffle(&mut self) {}
-        if let Some(current) = self.playlist.current() {
+
-            self.player_state.play(current)
+    pub fn toggle_next(&mut self) {}
-        }
+
+    pub fn toggle_repeat(&mut self) {}
+
+    pub fn fetch_state(&mut self) {
+        let mut socket = self.socket.as_mut().unwrap();
+        Message::new(MessageType::FetchState, None)
+            .send(&mut socket)
+            .unwrap();
     }
 
     pub fn on_key(&mut self, key: char) {
         match key {
-            'b' => self.play_previous(),
+            's' => self.toggle_shuffle(),
-            'n' => self.play_next(),
+            'n' => self.toggle_next(),
+            'r' => self.toggle_repeat(),
+            ' ' => self.fetch_state(),
             'q' => self.should_quit = true,
-            ' ' => self.player_state.toggle_pause(),
             _ => (),
         }
     }
@@ -229,33 +59,13 @@ impl App {
 
     pub fn on_right(&mut self) {}
 
-    pub fn on_up(&mut self) {
+    pub fn on_up(&mut self) {}
-        if self.playlist.items.len() > 0 {
-            self.playlist.previous()
-        }
-    }
 
-    pub fn on_down(&mut self) {
+    pub fn on_down(&mut self) {}
-        if self.playlist.items.len() > 0 {
-            self.playlist.next()
-        }
-    }
 
-    pub fn on_enter(&mut self) {
+    pub fn on_enter(&mut self) {}
-        if let Some(path) = self.playlist.current() {
-            self.player_state.play(path)
-        }
-    }
 
     pub fn on_tab(&mut self) {}
 
-    pub fn on_tick(&mut self, duration: Duration) {
+    pub fn on_tick(&mut self, duration: Duration) {}
-        let time_mod = if self.player_state.is_playing {
-            0.25
-        } else {
-            0.0
-        };
-        self.player_state.time_ratio =
-            (self.player_state.time_ratio + duration.as_secs_f64() * time_mod) % 1.0
-    }
 }

src/client/crossterm.rs

@@ -50,6 +50,14 @@ fn run_app<B: Backend>(
     loop {
         terminal.draw(|f| ui::draw(f, &mut app))?;
 
+        if !app.connected() {
+            match app.connect() {
+                Ok(_) => (),
+                Err(_) => (),
+            }
+            continue;
+        }
+
         let timeout = tick_rate
             .checked_sub(last_tick.elapsed())
             .unwrap_or_else(|| Duration::from_secs(0));

src/client/ui.rs

@@ -1,54 +1,47 @@
 use ratatui::{prelude::*, widgets::*};
-use std::path::Path;
 
 use super::app::App;
 
 pub fn draw<B: Backend>(f: &mut Frame<B>, app: &mut App) {
-    let chunks = Layout::default()
+    if app.connected() {
-        .constraints([Constraint::Min(3), Constraint::Length(2)].as_ref())
+        let chunks = Layout::default()
-        .split(f.size());
+            .constraints([Constraint::Min(3), Constraint::Length(2)].as_ref())
-    draw_playlist(f, app, chunks[0]);
+            .split(f.size());
-    draw_player(f, app, chunks[1]);
+        draw_playlist(f, app, chunks[0]);
+        draw_player(f, app, chunks[1]);
+    } else {
+        draw_no_connection(f);
+    }
 }
 
 static PRIMARY_COLOR: Color = Color::Rgb(200, 150, 70);
 static SECONDARY_COLOR: Color = Color::Rgb(200, 200, 200);
 
+fn draw_no_connection<B: Backend>(f: &mut Frame<B>) {
+    let message = "Not connected";
+    let width = message.len() as u16 + 4;
+    let height = 3;
+
+    let x = (f.size().width as i16 - width as i16).max(0) as u16 / 2;
+    let y = (f.size().height as i16 - height as i16).max(0) as u16 / 2;
+
+    let area = Rect {
+        x,
+        y,
+        width: u16::min(width, f.size().width - x),
+        height: u16::min(height, f.size().height - y),
+    };
+    f.render_widget(
+        Paragraph::new(message)
+            .fg(SECONDARY_COLOR)
+            .block(Block::default().borders(Borders::ALL).fg(PRIMARY_COLOR))
+            .alignment(Alignment::Center),
+        area,
+    );
+}
+
 fn draw_playlist<B: Backend>(f: &mut Frame<B>, app: &App, area: Rect) {
-    let tracks: Vec<_> = app
+    let playlist = List::new(vec![])
-        .playlist
-        .items
-        .iter()
-        .enumerate()
-        .map(|(index, path)| {
-            let selected = app
-                .playlist
-                .state
-                .selected()
-                .map_or(false, |selected| index == selected);
-            let playing = app
-                .player_state
-                .currently_playing
-                .clone()
-                .map_or(false, |currently_playing| currently_playing == *path);
-            let mut style = Style::default();
-            match (selected, playing) {
-                (true, false) => {
-                    style.fg = Some(Color::Black);
-                    style.bg = Some(PRIMARY_COLOR);
-                }
-                (false, true) => style.fg = Some(PRIMARY_COLOR),
-                (true, true) => {
-                    style.fg = None;
-                    style.bg = Some(PRIMARY_COLOR);
-                }
-                (_, _) => (),
-            }
-            let content = Span::from(format_track_name(path));
-            ListItem::new(content).set_style(style)
-        })
-        .collect();
-    let playlist = List::new(tracks)
         .block(
             Block::default()
                 .borders(Borders::ALL)
@@ -59,22 +52,11 @@ fn draw_playlist<B: Backend>(f: &mut Frame<B>, app: &App, area: Rect) {
 }
 
 fn draw_player<B: Backend>(f: &mut Frame<B>, app: &App, area: Rect) {
-    let title_content = match app.player_state.currently_playing.as_ref() {
+    let title_content = vec![
-        Some(playing) => {
+        Span::from("[ "),
-            let symbol = match app.player_state.is_playing {
+        Span::from("??").fg(PRIMARY_COLOR),
-                true => Span::from("||").fg(PRIMARY_COLOR),
+        Span::from(" ]"),
-                false => Span::from("|>").fg(Color::Black).bg(PRIMARY_COLOR),
+    ];
-            };
-            vec![
-                Span::from("[ "),
-                symbol,
-                Span::from(" "),
-                Span::from(format_track_name(playing.as_str())),
-                Span::from(" ]"),
-            ]
-        }
-        None => vec![Span::from(" Nothing selected ")],
-    };
 
     let player = Gauge::default()
         .block(
@@ -85,19 +67,8 @@ fn draw_player<B: Backend>(f: &mut Frame<B>, app: &App, area: Rect) {
                 .title_position(block::Position::Top),
         )
         .gauge_style(Style::default().fg(PRIMARY_COLOR))
-        .ratio(app.player_state.time_ratio)
+        .ratio(0.25)
-        .label(if app.player_state.is_playing {
+        .label("[ PAUSED ]");
-            ""
-        } else {
-            "[ PAUSED ]"
-        });
 
     f.render_widget(player, area)
 }
-
-fn format_track_name(path: &str) -> String {
-    match Path::new(path).file_stem() {
-        Some(file_name) => file_name.to_string_lossy().to_string(),
-        None => path.to_string(),
-    }
-}

src/lib.rs

@@ -0,0 +1,269 @@
+use std::path::PathBuf;
+
+use serde::{Deserialize, Serialize};
+
+#[derive(Serialize, Deserialize, Debug, Default)]
+pub struct ServerState {
+    pub playlist_params: PlaylistParams,
+    pub directory_playlist: Option<DirectoryPlaylist>,
+    pub queue_playlist: QueuePlaylist,
+}
+
+pub type PlaylistElement = PathBuf;
+
+pub enum PlaylistType {
+    Directory,
+    Queue,
+}
+
+#[derive(Serialize, Deserialize, Debug, Default)]
+pub struct Playlist {
+    pub items: Vec<PlaylistElement>,
+    pub current: Option<usize>,
+}
+
+impl Playlist {}
+
+#[derive(Serialize, Deserialize, Debug)]
+pub struct DirectoryPlaylist {
+    pub directory: PathBuf,
+    pub playlist: Playlist,
+}
+
+#[derive(Serialize, Deserialize, Debug, Default)]
+pub struct QueuePlaylist {
+    pub playlist: Playlist,
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+pub struct PlaylistParams {
+    pub shuffle: bool,
+    pub next: bool,
+    pub repeat: bool,
+}
+
+impl Default for PlaylistParams {
+    fn default() -> Self {
+        Self {
+            shuffle: false,
+            next: true,
+            repeat: true,
+        }
+    }
+}
+
+/// Protocol for client-server communication
+pub mod protocol {
+    use std::io::{Read, Write};
+
+    use interprocess::local_socket::LocalSocketStream;
+    use serde::{Deserialize, Serialize};
+
+    /// Prefix messages with this header
+    pub const HEADER_MAGIC: [u8; 4] = [0xCA, 0xFE, 0xBA, 0xBE];
+    /// Maximum allowed body size
+    pub const MAX_BODY_LENGTH: usize = 10 * 1024 * 1024;
+
+    #[derive(Debug)]
+    pub enum MessageError {
+        HeaderMismatch,
+        BodySizeLimit,
+        ChecksumMismatch,
+        ReadError,
+        DeserializationError,
+    }
+
+    #[derive(Debug, PartialEq, Eq, Hash, Clone, Copy, Deserialize, Serialize)]
+    pub enum MessageType {
+        FetchState = 0,
+        FetchStateAck,
+    }
+
+    #[derive(Debug, Serialize, Deserialize)]
+    pub struct Message {
+        pub message_type: MessageType,
+        pub body: Option<Vec<u8>>,
+    }
+
+    impl Message {
+        pub fn new(message_type: MessageType, body: Option<&[u8]>) -> Self {
+            Self {
+                message_type,
+                body: body.map(|b| Vec::from(b)),
+            }
+        }
+
+        /// Message format (values are in little-endian):
+        /// offset | size | explanation
+        /// -------+------+-----------
+        /// 0x00   |  u32 | HEADER_MAGIC
+        /// 0x04   |  u32 | Body checksum
+        /// 0x08   |  u32 | Body length
+        /// 0x12   |    ? | Body
+        fn as_bytes(&self) -> Vec<u8> {
+            let magic = &HEADER_MAGIC[..];
+            let body = &bincode::serialize(self).unwrap();
+            let checksum = &crc32fast::hash(&body).to_le_bytes();
+            let body_length = &(body.len() as u32).to_le_bytes();
+            [magic, checksum, body_length, body].concat()
+        }
+
+        pub fn send(&self, stream: &mut LocalSocketStream) -> Result<(), std::io::Error> {
+            let bytes = self.as_bytes();
+            stream.write_all(&bytes)?;
+            Ok(())
+        }
+    }
+
+    pub fn parse_stream(stream: &mut LocalSocketStream) -> Result<Message, MessageError> {
+        let mut magic_buffer = vec![0; HEADER_MAGIC.len()];
+        if let Err(_) = stream.read_exact(&mut magic_buffer) {
+            return Err(MessageError::ReadError);
+        }
+        if magic_buffer != HEADER_MAGIC {
+            return Err(MessageError::HeaderMismatch);
+        }
+
+        let mut checksum_buffer = [0; 4];
+        if let Err(_) = stream.read_exact(&mut checksum_buffer) {
+            return Err(MessageError::ReadError);
+        }
+        let expected_checksum = u32::from_le_bytes(checksum_buffer);
+
+        let mut body_length_buffer = [0; 4];
+        if let Err(_) = stream.read_exact(&mut body_length_buffer) {
+            return Err(MessageError::ReadError);
+        }
+        let expected_body_length = u32::from_le_bytes(body_length_buffer) as usize;
+
+        if expected_body_length > MAX_BODY_LENGTH {
+            return Err(MessageError::BodySizeLimit);
+        }
+
+        let mut body_buffer = vec![0; expected_body_length];
+        if let Err(_) = stream.read_exact(&mut body_buffer) {
+            return Err(MessageError::ReadError);
+        }
+
+        if crc32fast::hash(&body_buffer) != expected_checksum {
+            return Err(MessageError::ChecksumMismatch);
+        }
+
+        println!("Stream data:\n\t{magic_buffer:?}\n\t{checksum_buffer:?}\n\t{body_length_buffer:?}\n\t{body_buffer:?}");
+        bincode::deserialize(&body_buffer).map_err(|_| MessageError::DeserializationError)
+    }
+}
+
+pub mod server {
+    use std::{fmt::Debug, path::PathBuf};
+
+    #[derive(Debug)]
+    pub enum ServerError {
+        Other(String),
+        Io(std::io::Error),
+        AlreadyStarted,
+        MissingRuntimeDir(PathBuf),
+    }
+
+    impl ServerError {
+        pub fn from_debuggable(err: impl Debug) -> Self {
+            Self::Other(format!("Unexpected error: {err:?}").to_string())
+        }
+    }
+}
+
+#[cfg(target_family = "unix")]
+pub mod os {
+    use std::{
+        fs,
+        path::{Path, PathBuf},
+        process::{id, Command, Stdio},
+    };
+
+    use super::server::ServerError;
+
+    pub fn reserve_pid() -> Result<(), ServerError> {
+        let pid_path = get_pid_path()?;
+        is_running()?;
+
+        fs::write(&pid_path, id().to_string()).map_err(|err| ServerError::Io(err))?;
+        Command::new("chmod")
+            .args(&["600", &pid_path.to_string_lossy()])
+            .output()
+            .map_err(|err| ServerError::Io(err))?;
+        Ok(())
+    }
+
+    pub fn is_running() -> Result<bool, ServerError> {
+        let pid_path = get_pid_path()?;
+
+        match fs::read(&pid_path) {
+            Ok(old_pid) => {
+                let old_pid =
+                    String::from_utf8(old_pid).map_err(|err| ServerError::from_debuggable(err))?;
+                let old_pid = old_pid.trim();
+                Ok(Command::new("ps")
+                    .args(&["-p", old_pid])
+                    .stdout(Stdio::null())
+                    .stderr(Stdio::null())
+                    .status()
+                    .map_err(|err| ServerError::Io(err))?
+                    .success())
+            }
+            _ => Ok(false),
+        }
+    }
+
+    pub fn run_in_background() -> Result<(), ServerError> {
+        let this = std::env::args().next().unwrap();
+        Command::new(this)
+            .stdout(Stdio::null())
+            .stderr(Stdio::null())
+            .args(&["-s"])
+            .spawn()
+            .map_err(|err| ServerError::Io(err))?;
+        Ok(())
+    }
+
+    pub fn kill() -> Result<(), ServerError> {
+        let pid_path = get_pid_path()?;
+        let socket_path = get_socket_path()?;
+        let pid = String::from_utf8(fs::read(&pid_path).map_err(|err| ServerError::Io(err))?)
+            .map_err(|err| ServerError::from_debuggable(err))?;
+        let pid = pid.trim();
+        Command::new("kill")
+            .arg(pid)
+            .spawn()
+            .map_err(|err| ServerError::Io(err))?;
+        Command::new("rm")
+            .args(&[
+                "-f",
+                &pid_path.to_string_lossy(),
+                &socket_path.to_string_lossy(),
+            ])
+            .spawn()
+            .map_err(|err| ServerError::Io(err))?;
+        Ok(())
+    }
+
+    pub fn get_socket_path() -> Result<PathBuf, ServerError> {
+        Ok(get_runtime_dir()?.join("rmp.socket"))
+    }
+
+    fn get_runtime_dir() -> Result<PathBuf, ServerError> {
+        let uid = String::from_utf8(
+            Command::new("id")
+                .arg("-u")
+                .output()
+                .map_err(|err| ServerError::Io(err))?
+                .stdout,
+        )
+        .map_err(|err| ServerError::from_debuggable(err))?;
+        let dir = Path::new("/run/user").join(uid.trim().to_string());
+        Ok(dir)
+    }
+
+    fn get_pid_path() -> Result<PathBuf, ServerError> {
+        Ok(get_runtime_dir()?.join("rmp.pid"))
+    }
+}

src/main.rs

@@ -10,6 +10,18 @@ pub struct CliArgs {
     #[argh(switch, short = 's')]
     server: bool,
 
+    /// randomize next track?
+    #[argh(option, default = "false")]
+    shuffle: bool,
+
+    /// change track after the current one is finished?
+    #[argh(option, default = "true")]
+    next: bool,
+
+    /// repeat the playlist (or track if 'next' is disabled) after it's finished
+    #[argh(option, default = "true")]
+    repeat: bool,
+
     /// kill server
     #[argh(switch, short = 'x')]
     exit: bool,
@@ -43,5 +55,5 @@ fn main() {
     if !args.client_only && !server::is_running().unwrap() {
         server::run_in_background();
     }
-    client::run(args).map(|_| server::kill()).unwrap();
+    client::run(args).unwrap();
 }

src/server.rs

@@ -1,32 +1,21 @@
-use interprocess::local_socket::LocalSocketListener;
+use interprocess::local_socket::{LocalSocketListener, LocalSocketStream};
-use std::{fmt::Debug, fs, path::PathBuf, process};
+use rmp::{os, server::ServerError, PlaylistParams, ServerState};
+use std::{fs, path::PathBuf};
 
 use crate::CliArgs;
 
 pub mod audio_backend;
 
-#[derive(Debug)]
-pub enum ServerError {
-    Other(String),
-    Io(std::io::Error),
-    AlreadyStarted,
-    MissingRuntimeDir(PathBuf),
-}
-
-impl ServerError {
-    fn from_debuggable(err: impl Debug) -> Self {
-        Self::Other(format!("Unexpected error: {err:?}").to_string())
-    }
-}
-
 pub struct Server {
     pub socket: LocalSocketListener,
+    pub state: ServerState,
 }
 
 impl Server {
-    pub fn new(socket_path: PathBuf) -> Result<Self, ServerError> {
+    pub fn new(socket_path: PathBuf, state: ServerState) -> Result<Self, ServerError> {
         Ok(Self {
             socket: LocalSocketListener::bind(socket_path).map_err(|err| ServerError::Io(err))?,
+            state,
         })
     }
 }
@@ -34,15 +23,22 @@ impl Server {
 pub fn run(args: CliArgs) -> Result<(), ServerError> {
     if let Err(err) = os::reserve_pid() {
         let exit_code = handle_error(err);
-        process::exit(exit_code);
+        std::process::exit(exit_code);
     }
-    serve()
+    serve(ServerState {
+        playlist_params: PlaylistParams {
+            shuffle: args.shuffle,
+            next: args.next,
+            repeat: args.repeat,
+        },
+        ..Default::default()
+    })
 }
 
 pub fn kill() {
     if let Err(err) = os::kill() {
         let exit_code = handle_error(err);
-        process::exit(exit_code);
+        std::process::exit(exit_code);
     }
 }
 
@@ -51,7 +47,7 @@ pub fn is_running() -> Result<bool, ServerError> {
         Ok(is_running) => Ok(is_running),
         Err(err) => {
             let exit_code = handle_error(err);
-            process::exit(exit_code);
+            std::process::exit(exit_code);
         }
     }
 }
@@ -59,32 +55,57 @@ pub fn is_running() -> Result<bool, ServerError> {
 pub fn run_in_background() {
     if let Err(err) = os::run_in_background() {
         let exit_code = handle_error(err);
-        process::exit(exit_code);
+        std::process::exit(exit_code);
     }
 }
 
-fn serve() -> Result<(), ServerError> {
+fn serve(state: ServerState) -> Result<(), ServerError> {
     let socket_path = os::get_socket_path()?;
     if socket_path.exists() {
         fs::remove_file(&socket_path).map_err(|err| ServerError::Io(err))?;
     }
-    let server = Server::new(socket_path)?;
+    println!("state: {state:?}");
-    println!("Waiting for messages...");
+    let server = Server::new(socket_path, state)?;
+    println!("Waiting for connections...");
     for message in server.socket.incoming() {
         match message {
             Ok(stream) => {
-                println!("\tstream: {:?}", stream);
+                let thread_builder = std::thread::Builder::new().name("session_handler".into());
+                thread_builder
+                    .spawn(move || session_handler(stream))
+                    .unwrap();
             }
             Err(err) => {
                 let exit_code = handle_error(ServerError::Io(err));
-                process::exit(exit_code);
+                std::process::exit(exit_code);
             }
-        }
+        };
     }
     println!("Reached the end of an infinite loop?");
     Ok(())
 }
 
+fn session_handler(mut stream: LocalSocketStream) {
+    let thread_id = std::thread::current().id();
+    println!("session created: {thread_id:?}");
+    loop {
+        match rmp::protocol::parse_stream(&mut stream) {
+            Ok(body) => {
+                println!("Message: {body:?}")
+            }
+            Err(error) => match error {
+                rmp::protocol::MessageError::ReadError => {
+                    println!("session terminated: {thread_id:?}");
+                    return;
+                }
+                error => {
+                    eprintln!("Message error in {thread_id:?}: {error:?}")
+                }
+            },
+        }
+    }
+}
+
 fn handle_error(err: ServerError) -> i32 {
     match &err {
         ServerError::Other(msg) => {
@@ -105,99 +126,3 @@ fn handle_error(err: ServerError) -> i32 {
         }
     }
 }
-
-#[cfg(target_family = "unix")]
-mod os {
-    use std::{
-        fs,
-        path::{Path, PathBuf},
-        process::{id, Command, Stdio},
-    };
-
-    use super::ServerError;
-
-    pub fn reserve_pid() -> Result<(), ServerError> {
-        let pid_path = get_pid_path()?;
-        is_running()?;
-
-        fs::write(&pid_path, id().to_string()).map_err(|err| ServerError::Io(err))?;
-        Command::new("chmod")
-            .args(&["600", &pid_path.to_string_lossy()])
-            .output()
-            .map_err(|err| ServerError::Io(err))?;
-        Ok(())
-    }
-
-    pub fn is_running() -> Result<bool, ServerError> {
-        let pid_path = get_pid_path()?;
-
-        match fs::read(&pid_path) {
-            Ok(old_pid) => {
-                let old_pid =
-                    String::from_utf8(old_pid).map_err(|err| ServerError::from_debuggable(err))?;
-                let old_pid = old_pid.trim();
-                Ok(Command::new("ps")
-                    .args(&["-p", old_pid])
-                    .stdout(Stdio::null())
-                    .stderr(Stdio::null())
-                    .status()
-                    .map_err(|err| ServerError::Io(err))?
-                    .success())
-            }
-            _ => Ok(false),
-        }
-    }
-
-    pub fn run_in_background() -> Result<(), ServerError> {
-        let this = std::env::args().next().unwrap();
-        Command::new(this)
-            .stdout(Stdio::null())
-            .stderr(Stdio::null())
-            .args(&["-s"])
-            .spawn()
-            .map_err(|err| ServerError::Io(err))?;
-        Ok(())
-    }
-
-    pub fn kill() -> Result<(), ServerError> {
-        let pid_path = get_pid_path()?;
-        let socket_path = get_socket_path()?;
-        let pid = String::from_utf8(fs::read(&pid_path).map_err(|err| ServerError::Io(err))?)
-            .map_err(|err| ServerError::from_debuggable(err))?;
-        let pid = pid.trim();
-        Command::new("kill")
-            .arg(pid)
-            .spawn()
-            .map_err(|err| ServerError::Io(err))?;
-        Command::new("rm")
-            .args(&[
-                "-f",
-                &pid_path.to_string_lossy(),
-                &socket_path.to_string_lossy(),
-            ])
-            .spawn()
-            .map_err(|err| ServerError::Io(err))?;
-        Ok(())
-    }
-
-    pub fn get_socket_path() -> Result<PathBuf, ServerError> {
-        Ok(get_runtime_dir()?.join("rmp.socket"))
-    }
-
-    fn get_runtime_dir() -> Result<PathBuf, ServerError> {
-        let uid = String::from_utf8(
-            Command::new("id")
-                .arg("-u")
-                .output()
-                .map_err(|err| ServerError::Io(err))?
-                .stdout,
-        )
-        .map_err(|err| ServerError::from_debuggable(err))?;
-        let dir = Path::new("/run/user").join(uid.trim().to_string());
-        Ok(dir)
-    }
-
-    fn get_pid_path() -> Result<PathBuf, ServerError> {
-        Ok(get_runtime_dir()?.join("rmp.pid"))
-    }
-}