services/redis: Implement Write::append with native support

Signed-off-by: Wenyu Huang <huangwenyuu@outlook.com>

src/raw/adapters/kv/api.rs

@@ -104,6 +104,31 @@ pub trait Adapter: Send + Sync + Debug + Unpin + 'static {
  )
  .with_operation("kv::Adapter::blocking_scan"))
  }
+
+ /// Append a key into service
+ async fn append(&self, path: &str, value: &[u8]) -> Result<()> {
+ let _ = path;
+ let _ = value;
+
+ Err(Error::new(
+ ErrorKind::Unsupported,
+ "kv adapter doesn't support this operation",
+ )
+ .with_operation("kv::Adapter::append"))
+ }
+
+ /// Append a key into service
+ /// in blocking way.
+ fn blocking_append(&self, path: &str, value: &[u8]) -> Result<()> {
+ let _ = path;
+ let _ = value;
+
+ Err(Error::new(
+ ErrorKind::Unsupported,
+ "kv adapter doesn't support this operation",
+ )
+ .with_operation("kv::Adapter::blocking_append"))
+ }
 }
 
 /// Metadata for this key value accessor.

src/raw/adapters/kv/backend.rs

@@ -285,13 +285,20 @@ impl<S: Adapter> oio::Write for KvWriter<S> {
  }
 
  async fn append(&mut self, bs: Bytes) -> Result<()> {
- self.buf.extend(bs);
-
+ if let Err(e) = self.kv.append(&self.path, bs.to_vec().as_slice()).await {
+ if e.kind() == ErrorKind::Unsupported {
+ self.buf.extend(bs);
+ } else {
+ return Err(e);
+ }
+ }
  Ok(())
  }
 
  async fn close(&mut self) -> Result<()> {
- self.kv.set(&self.path, &self.buf).await?;
+ if !self.buf.is_empty() {
+ self.kv.set(&self.path, &self.buf).await?;
+ }
 
  Ok(())
  }

src/services/redis/backend.rs

@@ -340,6 +340,12 @@ impl kv::Adapter for Adapter {
  let _: () = conn.del(key).await?;
  Ok(())
  }
+
+ async fn append(&self, key: &str, value: &[u8]) -> Result<()> {
+ let mut conn = self.conn().await?;
+ conn.append(key, value).await?;
+ Ok(())
+ }
 }
 
 impl From<RedisError> for Error {