Reset connection state after EXEC is done (#21)

EXEC was not resetting the connection status back to Normal, instead the
ExecutingTx state was kept and the connection would be prevented to
start another transaction until RESET is called.

src/cmd/transaction.rs

@@ -206,6 +206,62 @@ mod test {
         );
     }
 
+    #[tokio::test]
+    async fn test_two_consecutive_transactions() {
+        let c = create_connection();
+
+        assert_eq!(Ok(Value::Ok), run_command(&c, &["multi"]).await);
+        assert_eq!(Ok(Value::Queued), run_command(&c, &["get", "foo"]).await);
+        assert_eq!(
+            Ok(Value::Queued),
+            run_command(&c, &["set", "foo", "foo"]).await
+        );
+        assert_eq!(Ok(Value::Queued), run_command(&c, &["get", "foo"]).await);
+        assert_eq!(
+            Ok(Value::Array(vec![
+                Value::Null,
+                Value::Ok,
+                Value::Blob("foo".into()),
+            ])),
+            run_command(&c, &["exec"]).await
+        );
+
+        assert_eq!(Ok(Value::Ok), run_command(&c, &["multi"]).await);
+        assert_eq!(Ok(Value::Queued), run_command(&c, &["get", "foo"]).await);
+        assert_eq!(
+            Ok(Value::Queued),
+            run_command(&c, &["set", "foo", "bar"]).await
+        );
+        assert_eq!(Ok(Value::Queued), run_command(&c, &["get", "foo"]).await);
+        assert_eq!(
+            Ok(Value::Array(vec![
+                Value::Blob("foo".into()),
+                Value::Ok,
+                Value::Blob("bar".into()),
+            ])),
+            run_command(&c, &["exec"]).await
+        );
+    }
+
+    #[tokio::test]
+    async fn test_reset_drops_transaction() {
+        let c = create_connection();
+
+        assert_eq!(Ok(Value::Ok), run_command(&c, &["multi"]).await);
+        assert_eq!(Ok(Value::Queued), run_command(&c, &["get", "foo"]).await);
+        assert_eq!(
+            Ok(Value::Queued),
+            run_command(&c, &["set", "foo", "foo"]).await
+        );
+        assert_eq!(Ok(Value::Queued), run_command(&c, &["get", "foo"]).await);
+        assert_eq!(Ok(Value::String("RESET".into())), run_command(&c, &["reset"]).await);
+        assert_eq!(
+            Err(Error::NotInTx),
+            run_command(&c, &["exec"]).await
+        );
+
+    }
+
     fn get_keys(args: &[&str]) -> Vec<Bytes> {
         let args: Vec<Bytes> = args.iter().map(|s| Bytes::from(s.to_string())).collect();
         let d = Dispatcher::new();

src/connection/mod.rs

@@ -105,11 +105,11 @@ impl Connection {
     /// If the connection was not in a MULTI stage an error is thrown.
     pub fn stop_transaction(&self) -> Result<Value, Error> {
         let mut info = self.info.write();
-        if info.status == ConnectionStatus::Multi {
+        if info.status == ConnectionStatus::Multi || info.status == ConnectionStatus::ExecutingTx {
             info.commands = None;
             info.watch_keys.clear();
             info.tx_keys.clear();
-            info.status = ConnectionStatus::ExecutingTx;
+            info.status = ConnectionStatus::Normal;
 
             Ok(Value::Ok)
         } else {