188 fix delayed realtime stream (#189)

* checkpoint

* improve validation

* implement clone, partialeq

src/market_data/realtime/decoders.rs

@@ -13,10 +13,6 @@ use super::{
 mod tests;
 
 pub(super) fn decode_realtime_bar(message: &mut ResponseMessage) -> Result<Bar, Error> {
-    if message.len() < 11 {
-        return Err(Error::Simple("Invalid message length".into()));
-    }
-
     message.skip(); // message type
     message.skip(); // message version
     message.skip(); // message request id

src/messages.rs

@@ -439,6 +439,10 @@ impl ResponseMessage {
     }
 
     pub fn peek_int(&self, i: usize) -> Result<i32, Error> {
+        if self.i >= self.fields.len() {
+            return Err(Error::Simple("expected int and found end of message".into()));
+        }
+
         let field = &self.fields[i];
         match field.parse() {
             Ok(val) => Ok(val),
@@ -451,6 +455,10 @@ impl ResponseMessage {
     }
 
     pub fn next_int(&mut self) -> Result<i32, Error> {
+        if self.i >= self.fields.len() {
+            return Err(Error::Simple("expected int and found end of message".into()));
+        }
+
         let field = &self.fields[self.i];
         self.i += 1;
 
@@ -461,6 +469,10 @@ impl ResponseMessage {
     }
 
     pub fn next_optional_int(&mut self) -> Result<Option<i32>, Error> {
+        if self.i >= self.fields.len() {
+            return Err(Error::Simple("expected optional int and found end of message".into()));
+        }
+
         let field = &self.fields[self.i];
         self.i += 1;
 
@@ -475,13 +487,21 @@ impl ResponseMessage {
     }
 
     pub fn next_bool(&mut self) -> Result<bool, Error> {
+        if self.i >= self.fields.len() {
+            return Err(Error::Simple("expected bool and found end of message".into()));
+        }
+
         let field = &self.fields[self.i];
         self.i += 1;
 
         Ok(field == "1")
     }
 
     pub fn next_long(&mut self) -> Result<i64, Error> {
+        if self.i >= self.fields.len() {
+            return Err(Error::Simple("expected long and found end of message".into()));
+        }
+
         let field = &self.fields[self.i];
         self.i += 1;
 
@@ -492,6 +512,10 @@ impl ResponseMessage {
     }
 
     pub fn next_optional_long(&mut self) -> Result<Option<i64>, Error> {
+        if self.i >= self.fields.len() {
+            return Err(Error::Simple("expected optional long and found end of message".into()));
+        }
+
         let field = &self.fields[self.i];
         self.i += 1;
 
@@ -506,6 +530,10 @@ impl ResponseMessage {
     }
 
     pub fn next_date_time(&mut self) -> Result<OffsetDateTime, Error> {
+        if self.i >= self.fields.len() {
+            return Err(Error::Simple("expected datetime and found end of message".into()));
+        }
+
         let field = &self.fields[self.i];
         self.i += 1;
 
@@ -522,12 +550,20 @@ impl ResponseMessage {
     }
 
     pub fn next_string(&mut self) -> Result<String, Error> {
+        if self.i >= self.fields.len() {
+            return Err(Error::Simple("expected string and found end of message".into()));
+        }
+
         let field = &self.fields[self.i];
         self.i += 1;
         Ok(String::from(field))
     }
 
     pub fn next_double(&mut self) -> Result<f64, Error> {
+        if self.i >= self.fields.len() {
+            return Err(Error::Simple("expected double and found end of message".into()));
+        }
+
         let field = &self.fields[self.i];
         self.i += 1;
 
@@ -542,6 +578,10 @@ impl ResponseMessage {
     }
 
     pub fn next_optional_double(&mut self) -> Result<Option<f64>, Error> {
+        if self.i >= self.fields.len() {
+            return Err(Error::Simple("expected optional double and found end of message".into()));
+        }
+
         let field = &self.fields[self.i];
         self.i += 1;
 

src/scanner.rs

@@ -21,6 +21,8 @@ pub(super) fn scanner_parameters(client: &Client) -> Result<String, Error> {
     }
 }
 
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+/// Scanner subscription parameters.
 pub struct ScannerSubscription {
     /// The number of rows to be returned for the query
     pub number_of_rows: i32,