Merge pull request #27 from skytable/protocol/new-decoder

Fix protocol decoder and response issues

Author: ohsayan

CHANGELOG.md

@@ -2,54 +2,65 @@
 
 All changes in this project will be noted in this file.
 
-### 0.8.7 (unreleased)
+## 0.8.8
+
+### Fixes
+
+- Fixed response decoder and handling issues
+
+### Additions
+
+- Added `FromResponse` for `Vec<Row>`
+- Added `SQParam` impl for `&Vec<u8>`
+
+## 0.8.7
 
 > - **Field change warnings**:
 >   - The `Config` struct now has one additional field. This is not a breaking change because the functionality of the library remains unchanged
 
-#### Additions
+### Additions
 
 - Added support for pipelines
 - Added `Response::parse` to convert a response into compatible types
 
-### 0.8.6
+## 0.8.6
 
 Reduced allocations in `Query`.
 
-### 0.8.5
+## 0.8.5
 
 Fixed bugs with the derive macros.
 
-### 0.8.4
+## 0.8.4
 
 > **Yanked version**
 
 Fixed an issue with single-item struct derives when using the `Response` macro.
 
-### 0.8.3
+## 0.8.3
 
 Added the following implementations:
 - `FromResponse` for `Row`
 - `FromValue` for `Value` (this was erroneously missing)
 - Added the `Value::parse` and `Value::parse_cloned` member methods
 - Added `Row::into_first` and `Row::into_first_as` member methods
 
-### 0.8.2
+## 0.8.2
 
 Support deriving queries and responses.
 
-### 0.8.1
+## 0.8.1
 
 Fixed issues with documentation
 
 ## 0.8.0
 
-#### New features
+### New features
 - Completely up to date for Skyhash 2.0
 - New query API interface for Skytable Octave (completely breaking!)
 - No longer depends on OpenSSL
 
-#### Breaking changes
+### Breaking changes
 The enter query interface as changed and is incompatible with previous driver versions. Please consider reading the Skytable
 Octave upgrade guide.
 

src/io/aio.rs

@@ -26,8 +26,7 @@ use {
         error::{ClientResult, ConnectionSetupError, Error},
         protocol::{
             handshake::{ClientHandshake, ServerHandshake},
-            state_init::{DecodeState, MRespState, PipelineResult, RState},
-            Decoder,
+            DecodeState, Decoder, MRespState, PipelineResult, RState,
         },
         query::Pipeline,
         response::{FromResponse, Response},
@@ -168,11 +167,12 @@ impl<C: AsyncWriteExt + AsyncReadExt + Unpin> TcpConnection<C> {
                 return Err(Error::IoError(std::io::ErrorKind::ConnectionReset.into()));
             }
             self.buf.extend_from_slice(&buf[..n]);
-            let mut decoder = Decoder::new(&self.buf, cursor);
-            match decoder.validate_pipe(pipeline.query_count(), state) {
+            let (_state, _position) =
+                Decoder::new(&self.buf, cursor).validate_pipe(pipeline.query_count(), state);
+            match _state {
                 PipelineResult::Completed(r) => return Ok(r),
                 PipelineResult::Pending(_state) => {
-                    cursor = decoder.position();
+                    cursor = _position;
                     state = _state;
                 }
                 PipelineResult::Error(e) => return Err(e.into()),
@@ -198,13 +198,13 @@ impl<C: AsyncWriteExt + AsyncReadExt + Unpin> TcpConnection<C> {
                 continue;
             }
             self.buf.extend_from_slice(&buf[..n]);
-            let mut decoder = Decoder::new(&self.buf, cursor);
-            match decoder.validate_response(state) {
+            let (_state, _position) = Decoder::new(&self.buf, cursor).validate_response(state);
+            match _state {
                 DecodeState::Completed(resp) => return Ok(resp),
                 DecodeState::ChangeState(_state) => {
                     expected = 1;
                     state = _state;
-                    cursor = decoder.position();
+                    cursor = _position;
                 }
                 DecodeState::Error(e) => return Err(Error::ProtocolError(e)),
             }

src/io/sync.rs

@@ -28,8 +28,7 @@ use {
         error::{ClientResult, ConnectionSetupError, Error},
         protocol::{
             handshake::{ClientHandshake, ServerHandshake},
-            state_init::{DecodeState, MRespState, PipelineResult, RState},
-            Decoder,
+            DecodeState, Decoder, MRespState, PipelineResult, RState,
         },
         query::Pipeline,
         response::{FromResponse, Response},
@@ -163,11 +162,12 @@ impl<C: Write + Read> TcpConnection<C> {
                 return Err(Error::IoError(std::io::ErrorKind::ConnectionReset.into()));
             }
             self.buf.extend_from_slice(&buf[..n]);
-            let mut decoder = Decoder::new(&self.buf, cursor);
-            match decoder.validate_pipe(pipeline.query_count(), state) {
+            let (_state, _position) =
+                Decoder::new(&self.buf, cursor).validate_pipe(pipeline.query_count(), state);
+            match _state {
                 PipelineResult::Completed(r) => return Ok(r),
                 PipelineResult::Pending(_state) => {
-                    cursor = decoder.position();
+                    cursor = _position;
                     state = _state;
                 }
                 PipelineResult::Error(e) => return Err(e.into()),
@@ -189,15 +189,14 @@ impl<C: Write + Read> TcpConnection<C> {
                 return Err(Error::IoError(std::io::ErrorKind::ConnectionReset.into()));
             }
             self.buf.extend_from_slice(&buf[..n]);
-            let mut decoder = Decoder::new(&self.buf, cursor);
-            match decoder.validate_response(state) {
-                DecodeState::ChangeState(new_state) => {
-                    state = new_state;
-                    cursor = decoder.position();
-                    continue;
-                }
+            let (_state, _position) = Decoder::new(&self.buf, cursor).validate_response(state);
+            match _state {
                 DecodeState::Completed(resp) => return Ok(resp),
-                DecodeState::Error(e) => return Err(e.into()),
+                DecodeState::ChangeState(_state) => {
+                    state = _state;
+                    cursor = _position;
+                }
+                DecodeState::Error(e) => return Err(Error::ProtocolError(e)),
             }
         }
     }