[Bug] Mocked *Scan responses are dropped

[phogue]

Fred version - 10.0.1
Redis version - any
Platform - any
Deployment type - mocked

Describe the bug

src/protocol/command.rs::respond_to_caller drops Ok values from process_command for client.sscan(..) and other scan's.

Hacking a local solution by modifying src/protocol/command.rs::respond_to_caller to call src/protocol/responders.rs::respond_value_scan with Ok values propagates results correctly

To Reproduce
Steps to reproduce the behavior:

Some quick hacky test and mocked mock for SCAN.

I don't know if this is where it should be solved or any knock on effects of implementing a proper solution like this for an MR. Just verified the test at least passes/values are passed back with a hack

// src/modules/mocks.rs

impl SimpleMap {
  // ...

  /// Perform a `SCAN` operation. A mocked.. mock.
  pub fn scan(&self, args: Vec<Value>) -> Result<Value, Error> {
    let mut guard = self.values.lock();
    let keys: Vec<Value> = guard
      .keys()
      .filter_map(|k| k.as_str().map(|k| Value::String(k.into())))
      .collect();
    Ok(Value::Array(vec![Value::from_static_str("0"), Value::Array(keys)]))
  }
  // ...
}

// ...

mod tests {

  // ...

  #[tokio::test]
  async fn should_mock_scans() {
    let (client, _) = create_mock_client(Arc::new(SimpleMap::new())).await;
    client
      .set::<(), _, _>("foo1", "bar1", None, None, false)
      .await
      .expect("Failed to call SET");
    client
      .set::<(), _, _>("foo2", "bar2", None, None, false)
      .await
      .expect("Failed to call SET");
    let mut all: Vec<String> = Vec::new();
    let mut scan_stream = client.scan("foo*", Some(10), None);
    while let Some(mut page) = scan_stream.try_next().await.expect("failed to call try_next") {
      if let Some(keys) = page.take_results() {
        all.append(
          &mut keys
            .into_iter()
            .filter_map(|v| v.as_str().map(|v| v.to_string()))
            .collect(),
        );
      }
      page.next();
    }
    all.sort();
    assert_eq!(all, vec!["foo1".to_string(), "foo2".to_string()]);
  }
  // ...
}

// src/protocol/command.rs 

// copy+paste of respond_to_caller fn to change/own params for a test
pub fn mocked_respond_to_caller(mut self, inner: &RefCount<ClientInner>, result: Result<Resp3Frame, Error>) {
  // ...
      ResponseKind::KeyScan(ref scanner) => {
        if let Err(error) = result {
          let _ = scanner.tx.try_send(Err(error));
        } else if let (Ok(frame), ResponseKind::KeyScan(scanner)) = (result, self.take_response()) {
          // responders::mocked_respond_key_scan just drops the server arg and makes the command a ref
          responders::mocked_respond_key_scan(inner, &self, scanner, frame);
        }
      },
  // ...
}

// src/router/command.rs 

mod mocking {
  // ...
  // Edited to pass inner: &RefCount<ClientInner>, to pass down to mocked_respond_to_caller
  pub fn process_command(...) {
    // ...
    RouterCommand::Command(mut command) => {
      let result = mocks
        .process_command(command.to_mocked())
        .map(protocol_utils::mocked_value_to_frame);
      command.mocked_respond_to_caller(inner, result);
      Ok(())
    },
    // ...
  }
  // ...
}