Refactor and rethink initialization

README.org

@@ -24,6 +24,24 @@ sudo zypper install gstreamer gstreamer-plugins-base gstreamer-plugins-good
 
 * Usage
 
+** Server
+
+The must be running.
+
+To install it, first build (see later in this document) and then copy it to your reMarkable.
+
+#+begin_src shell
+scp review-server.arm.static root@${REMARKABLE_IP}:review-server
+#+end_src
+
+And then start it manually.
+
+#+begin_src shell
+ssh root@${REMARKABLE_IP} 'RUST_LOG=trace ./review-server --port 6680'
+#+end_src
+
+TODO: Install it as service
+
 ** Client
 
 Options for the client can be set as command line options or environment variables.
@@ -38,10 +56,8 @@ Usage: review-client [OPTIONS]
 
 Options:
       --remarkable-ip <remarkable-ip>  IP of the reMarkable tablet (default: 10.11.99.1)
-      --ssh-port <ssh-port>            SSH Port used by the reMarkable tablet (default: 22)
-      --ssh-key <ssh-key>              Private SSH key file path
       --tcp-port <tcp-port>            TCP port for video stream (default: 6680)
-      --framerate <framerate>          Framerate (default: 120)
+      --framerate <framerate>          Framerate (default: 50)
       --dark-mode                      Dark mode - invert colors (default: false)
       --show-cursor                    Show cursor (default: false)
   -h, --help                           Print help
@@ -52,9 +68,8 @@ Options:
 
 Corresponding keys:
 - =REMARKABLE_IP=
-- =REMARKABLE_SSH_PORT=
-- =REMARKABLE_SSH_KEY_PATH=
 - =REMARKABLE_TCP_PORT=
+# - =REMARKABLE_SSH_KEY_PATH=
 - =REMARKABLE_FRAMERATE=
 - =REMARKABLE_DARK_MODE=
 - =REMARKABLE_SHOW_CURSOR=

client/src/config.rs

@@ -1,36 +1,31 @@
-use std::{
-    env::{self, VarError},
-    path::PathBuf,
-};
+use std::env::{self, VarError};
 
 use anyhow::{Context, Error};
 use clap::Parser;
-use review_server::config::ServerOptions;
 
 const DEFAULT_IP: &str = "10.11.99.1";
 const DEFAULT_SSH_PORT: u16 = 22;
 const DEFAULT_TCP_PORT: u16 = 6680;
 
+const DEFAULT_FRAMERATE: f32 = 50.;
+
 #[derive(Parser, Debug)]
 #[command(author, version)]
 pub struct CliOptions {
     /// IP of the reMarkable tablet (default: 10.11.99.1)
     #[arg(long, name = "remarkable-ip")]
     remarkable_ip: Option<String>,
 
-    /// SSH Port used by the reMarkable tablet (default: 22)
-    #[arg(long, name = "ssh-port")]
-    ssh_port: Option<u16>,
-
-    /// Private SSH key file path
-    #[arg(long, name = "ssh-key")]
-    ssh_key: Option<PathBuf>,
-
     /// TCP port for video stream (default: 6680)
     #[arg(long, name = "tcp-port")]
     tcp_port: Option<u16>,
 
-    /// Framerate (default: 120)
+    /*
+    /// Private SSH key file path
+    #[arg(long, name = "ssh-key")]
+    ssh_key: Option<PathBuf>,
+    */
+    /// Framerate (default: 50)
     #[arg(long, name = "framerate")]
     framerate: Option<f32>,
 
@@ -46,8 +41,7 @@ pub struct CliOptions {
 #[derive(Debug, Clone)]
 pub struct ClientOptions {
     pub remarkable_ip: String,
-    pub ssh_port: u16,
-    pub ssh_key: PathBuf,
+    //pub ssh_key: PathBuf,
     pub dark_mode: bool,
     pub tcp_port: u16,
     pub show_cursor: bool,
@@ -62,17 +56,7 @@ impl From<CliOptions> for ClientOptions {
                 "REMARKABLE_IP",
                 DEFAULT_IP.to_string(),
             ),
-            ssh_port: resolve_with(
-                value.ssh_port,
-                "REMARKABLE_SSH_PORT",
-                |string| {
-                    string
-                        .parse()
-                        .context("could not parse SSH port from environment")
-                },
-                DEFAULT_SSH_PORT,
-            ),
-            ssh_key: must_resolve_option(value.ssh_key, "REMARKABLE_SSH_KEY_PATH"),
+            //ssh_key: must_resolve_option(value.ssh_key, "REMARKABLE_SSH_KEY_PATH"),
             tcp_port: resolve_with(
                 value.tcp_port,
                 "REMARKABLE_TCP_PORT",
@@ -91,24 +75,14 @@ impl From<CliOptions> for ClientOptions {
                         .parse()
                         .context("could not parse framerate from environment")
                 },
-                120.,
+                DEFAULT_FRAMERATE,
             ),
             dark_mode: resolve_boolean_option(value.dark_mode, "REMARKABLE_DARK_MODE", false),
             show_cursor: resolve_boolean_option(value.show_cursor, "REMARKABLE_SHOW_CURSOR", false),
         }
     }
 }
 
-impl Into<ServerOptions> for ClientOptions {
-    fn into(self) -> ServerOptions {
-        ServerOptions {
-            port: self.tcp_port,
-            show_cursor: self.show_cursor,
-            framerate: self.framerate,
-        }
-    }
-}
-
 fn resolve_option<T: From<String>>(cli_value: Option<T>, variable_name: &str, default: T) -> T {
     resolve_with(
         cli_value,

client/src/connection/mod.rs

@@ -0,0 +1,59 @@
+pub mod video;
+
+use anyhow::{Context, Error};
+use futures::{SinkExt, StreamExt};
+use review_server::{
+    config::{StreamConfig, device::DeviceConfig},
+    version::VersionInfo,
+};
+use tokio::net::TcpStream;
+use tokio_util::codec::{Framed, LengthDelimitedCodec};
+use tracing::info;
+
+use crate::config::ClientOptions;
+
+#[derive(Debug)]
+pub struct Connection {
+    framed: Framed<TcpStream, LengthDelimitedCodec>,
+}
+
+impl Connection {
+    pub async fn new(client_options: ClientOptions) -> Result<Self, Error> {
+        info!("setting up TCP connection");
+        let stream = TcpStream::connect(format!(
+            "{}:{}",
+            client_options.remarkable_ip, client_options.tcp_port
+        ))
+        .await
+        .context("could not connect to TCP stream")?;
+
+        let framed = Framed::new(stream, LengthDelimitedCodec::new());
+
+        Ok(Connection { framed })
+    }
+
+    pub async fn receive_version_info(&mut self) -> Result<VersionInfo, Error> {
+        let msg = self
+            .framed
+            .next()
+            .await
+            .context("connection was dropped before version info was communicated")?
+            .context("could not receive version info message")?;
+
+        let version_info =
+            bson::deserialize_from_slice(&msg).context("could not deserialize version info")?;
+
+        Ok(version_info)
+    }
+
+    pub async fn send_stream_config(&mut self, stream_config: StreamConfig) -> Result<(), Error> {
+        let msg =
+            bson::serialize_to_vec(&stream_config).context("could not serialize stream config")?;
+
+        self.framed
+            .send(msg.into())
+            .await
+            .context("could not send stream config")
+            .map(|_| ())
+    }
+}

client/src/connection/video.rs

@@ -0,0 +1,51 @@
+use anyhow::{Context, Error};
+use futures::stream::StreamExt;
+use lz4_flex::decompress_size_prepended;
+use tracing::debug;
+
+use crate::display::Display;
+
+use super::Connection;
+use review_server::config::device::VideoConfig;
+
+#[derive(Debug)]
+pub struct VideoConnection {
+    conn: Connection,
+    display: Display,
+}
+
+impl VideoConnection {
+    pub fn new(conn: Connection, video_config: VideoConfig) -> Result<Self, Error> {
+        let display = Display::new(video_config).context("could not initialize display")?;
+
+        Ok(Self { conn, display })
+    }
+
+    pub async fn run(&mut self) -> Result<(), Error> {
+        loop {
+            debug!("attempting to read data from TCP stream");
+
+            let compressed_frame = self
+                .conn
+                .framed
+                .next()
+                .await
+                .context("TCP stream was closed")?
+                .context("could not read from TCP stream")?;
+
+            debug!(
+                "read one compressed frame from TCP stream ({} bytes)",
+                compressed_frame.len(),
+            );
+
+            let frame = decompress_size_prepended(&compressed_frame)
+                .context("could not decompress received frame")?;
+
+            debug!("decompressed: {} bytes", frame.len());
+
+            self.display
+                .push_frame(frame)
+                .context("could not push frame to display")?;
+        }
+    }
+}

client/src/display/mod.rs

@@ -3,92 +3,46 @@ use super::config::*;
 use std::{io::Read as _, thread::sleep, time::Duration};
 
 use anyhow::{Context, Error};
-use futures::stream::StreamExt;
 use gstreamer_app::AppSrc;
 use gstreamer_video::VideoFormat;
-use lz4_flex::decompress_size_prepended;
-use review_server::config::{CommunicatedConfig, PixelFormat};
-use tokio::net::TcpStream;
-use tokio_util::codec::{Framed, LengthDelimitedCodec};
+use review_server::config::device::{PixelFormat, VideoConfig};
 use tracing::{debug, info};
 
 use gstreamer::{Pipeline, prelude::*};
 
-pub async fn gstreamer_thread(opts: ClientOptions) -> Result<(), Error> {
-    gstreamer::init().context("could not init gstreamer")?;
-
-    sleep(Duration::from_millis(100));
-
-    info!("setting up TCP connection");
-    let stream = TcpStream::connect(format!("{}:{}", opts.remarkable_ip, opts.tcp_port))
-        .await
-        .context("could not connect to TCP stream")?;
-
-    let mut framed_stream = Framed::new(stream, LengthDelimitedCodec::new());
-    let communicated_config = get_communicated_config(&mut framed_stream)
-        .await
-        .context("could not get communicated config from TCP stream")?;
-
-    debug!("received communicated config: {:?}", &communicated_config);
-
-    let (pipeline, appsrc) =
-        build_pipeline(&communicated_config).context("could not build gstreamer pipeline")?;
-    pipeline
-        .set_state(gstreamer::State::Playing)
-        .context("could not start playing gstreamer pipeline")?;
-
-    loop {
-        debug!("attempting to read data from TCP stream");
-
-        let compressed_frame = framed_stream
-            .next()
-            .await
-            .context("TCP stream was closed")?
-            .context("could not read from TCP stream")?;
+#[derive(Debug)]
+pub struct Display {
+    pipeline: Pipeline,
+    appsrc: AppSrc,
+}
 
-        debug!(
-            "read one compressed frame from TCP stream ({} bytes)",
-            compressed_frame.len(),
-        );
+impl Display {
+    pub fn new(video_config: VideoConfig) -> Result<Self, Error> {
+        gstreamer::init().context("could not init gstreamer")?;
 
-        let frame = decompress_size_prepended(&compressed_frame)
-            .context("could not decompress received frame")?;
+        let (pipeline, appsrc) =
+            build_pipeline(&video_config).context("could not build gstreamer pipeline")?;
+        pipeline
+            .set_state(gstreamer::State::Playing)
+            .context("could not start playing gstreamer pipeline")?;
 
-        debug!("decompressed: {} bytes", frame.len());
+        Ok(Self { pipeline, appsrc })
+    }
 
+    pub fn push_frame(&mut self, frame: Vec<u8>) -> Result<(), Error> {
         let buffer = gstreamer::Buffer::from_mut_slice(frame);
-        appsrc
+        self.appsrc
             .push_buffer(buffer)
-            .context("could not push buffer to app source")?;
+            .context("could not push buffer to app source")
+            .map(|_| ())
     }
 }
 
-async fn get_communicated_config(
-    framed_stream: &mut Framed<TcpStream, LengthDelimitedCodec>,
-) -> Result<CommunicatedConfig, Error> {
-    let config_bytes = framed_stream
-        .next()
-        .await
-        .context("received None as config bytes")?
-        .context("could not receive config bytes")?;
-
-    bson::deserialize_from_slice(&config_bytes).context("could not deserialize config from bytes")
-}
-
-fn to_video_format(pixel_format: &PixelFormat) -> VideoFormat {
-    match pixel_format {
-        PixelFormat::Rgb565le => todo!("not sure what the video format for RGB 565 LE is"),
-        PixelFormat::Gray8 => VideoFormat::Gray8,
-        PixelFormat::Gray16be => VideoFormat::Gray16Be,
-        PixelFormat::Bgra => VideoFormat::Bgra,
-    }
-}
-
-fn build_pipeline(communicated_config: &CommunicatedConfig) -> Result<(Pipeline, AppSrc), Error> {
+fn build_pipeline(video_config: &VideoConfig) -> Result<(Pipeline, AppSrc), Error> {
     let video_info = gstreamer_video::VideoInfo::builder(
-        to_video_format(&communicated_config.video_config.pixel_format),
-        communicated_config.video_config.width as u32,
-        communicated_config.video_config.height as u32,
+        to_video_format(&video_config.pixel_format),
+        video_config.width as u32,
+        video_config.height as u32,
     )
     .build()
     .context("could not build video info")?;
@@ -115,3 +69,12 @@ fn build_pipeline(communicated_config: &CommunicatedConfig) -> Result<(Pipeline,
 
     Ok((pipeline, appsrc))
 }
+
+fn to_video_format(pixel_format: &PixelFormat) -> VideoFormat {
+    match pixel_format {
+        PixelFormat::Rgb565le => VideoFormat::Rgb16, // TODO: not sure
+        PixelFormat::Gray8 => VideoFormat::Gray8,
+        PixelFormat::Gray16be => VideoFormat::Gray16Be,
+        PixelFormat::Bgra => VideoFormat::Bgra,
+    }
+}