draft new api version

src/client/response/header.rs

@@ -8,7 +8,11 @@ pub use meta::Meta;
 pub use mime::Mime;
 pub use status::Status;
 
-use glib::Bytes;
+use gio::{
+    prelude::{IOStreamExt, InputStreamExt},
+    Cancellable, SocketConnection,
+};
+use glib::{Bytes, Priority};
 
 pub struct Header {
     status: Status,
@@ -21,39 +25,58 @@ pub struct Header {
 
 impl Header {
     // Constructors
-    pub fn from_response(bytes: &Bytes) -> Result<Self, Error> {
-        // Get header slice of bytes
-        let end = Self::end(bytes)?;
 
-        let bytes = Bytes::from(match bytes.get(..end) {
-            Some(buffer) => buffer,
-            None => return Err(Error::Buffer),
-        });
-
-        // Status is required, parse to continue
-        let status = match Status::from_header(&bytes) {
-            Ok(status) => Ok(status),
-            Err(reason) => Err(match reason {
-                status::Error::Decode => Error::StatusDecode,
-                status::Error::Undefined => Error::StatusUndefined,
-            }),
-        }?;
-
-        // Done
-        Ok(Self {
-            status,
-            meta: match Meta::from_header(&bytes) {
-                Ok(meta) => Some(meta),
-                Err(_) => None,
+    pub fn from_socket_connection_async(
+        socket_connection: SocketConnection,
+        priority: Option<Priority>,
+        cancellable: Option<Cancellable>,
+        callback: impl FnOnce(Result<Self, (Error, Option<&str>)>) + 'static,
+    ) {
+        // Take header buffer from input stream
+        Self::read_from_socket_connection_async(
+            Vec::with_capacity(1024),
+            socket_connection,
+            match cancellable {
+                Some(value) => Some(value),
+                None => None::<Cancellable>,
             },
-            mime: match Mime::from_header(&bytes) {
-                Ok(mime) => Some(mime),
-                Err(_) => None,
+            match priority {
+                Some(value) => value,
+                None => Priority::DEFAULT,
             },
-        })
+            |result| {
+                callback(match result {
+                    Ok(buffer) => {
+                        // Status is required, parse to continue
+                        match Status::from_header(&buffer) {
+                            Ok(status) => Ok(Self {
+                                status,
+                                meta: match Meta::from_header(&buffer) {
+                                    Ok(meta) => Some(meta),
+                                    Err(_) => None,
+                                },
+                                mime: match Mime::from_header(&buffer) {
+                                    Ok(mime) => Some(mime),
+                                    Err(_) => None,
+                                },
+                            }),
+                            Err(reason) => Err((
+                                match reason {
+                                    status::Error::Decode => Error::StatusDecode,
+                                    status::Error::Undefined => Error::StatusUndefined,
+                                },
+                                None,
+                            )),
+                        }
+                    }
+                    Err(error) => Err(error),
+                })
+            },
+        );
     }
 
     // Getters
+
     pub fn status(&self) -> &Status {
         &self.status
     }
@@ -68,13 +91,58 @@ impl Header {
 
     // Tools
 
-    /// Get last header byte (until \r)
-    fn end(bytes: &Bytes) -> Result<usize, Error> {
-        for (offset, &byte) in bytes.iter().enumerate() {
-            if byte == b'\r' {
-                return Ok(offset);
-            }
-        }
-        Err(Error::Format)
+    pub fn read_from_socket_connection_async(
+        mut buffer: Vec<Bytes>,
+        connection: SocketConnection,
+        cancellable: Option<Cancellable>,
+        priority: Priority,
+        callback: impl FnOnce(Result<Vec<u8>, (Error, Option<&str>)>) + 'static,
+    ) {
+        connection.input_stream().read_bytes_async(
+            1, // do not change!
+            priority,
+            cancellable.clone().as_ref(),
+            move |result| match result {
+                Ok(bytes) => {
+                    // Expect valid header length
+                    if bytes.len() == 0 || buffer.len() + 1 > 1024 {
+                        return callback(Err((Error::Protocol, None)));
+                    }
+
+                    // Read next byte without buffer record
+                    if bytes.contains(&b'\r') {
+                        return Self::read_from_socket_connection_async(
+                            buffer,
+                            connection,
+                            cancellable,
+                            priority,
+                            callback,
+                        );
+                    }
+
+                    // Complete without buffer record
+                    if bytes.contains(&b'\n') {
+                        return callback(Ok(buffer
+                            .iter()
+                            .flat_map(|byte| byte.iter())
+                            .cloned()
+                            .collect())); // convert to UTF-8
+                    }
+
+                    // Record
+                    buffer.push(bytes);
+
+                    // Continue
+                    Self::read_from_socket_connection_async(
+                        buffer,
+                        connection,
+                        cancellable,
+                        priority,
+                        callback,
+                    );
+                }
+                Err(reason) => callback(Err((Error::InputStream, Some(reason.message())))),
+            },
+        );
     }
 }