feat(core): Add start-after support for list

Closes apache#2069
This PR makes following changes:
1. added `start_after` argument for `OpList`.
2. added `list_with` method for `Operator`.
3. added `list_with_start_after` option for `Capability`.

core/src/types/capability.rs

@@ -94,6 +94,8 @@ pub struct Capability {
     pub list: bool,
     /// If backend supports list with limit, it will be true.
     pub list_with_limit: bool,
+    /// If backend supports list with start after, it will be true.
+    pub list_with_start_after: bool,
 
     /// If operator supports scan natively, it will be true.
     pub scan: bool,

core/src/types/operator/operator.rs

@@ -1082,6 +1082,45 @@ impl Operator {
     /// # }
     /// ```
     pub async fn list(&self, path: &str) -> Result<Lister> {
+        self.list_with(path, OpList::new()).await
+    }
+
+    /// List given path with OpList.
+    ///
+    /// This function will create a new handle to list entries.
+    ///
+    /// An error will be returned if given path doesn't end with `/`.
+    ///
+    /// # Examples
+    ///
+    /// ```no_run
+    /// # use anyhow::Result;
+    /// # use futures::io;
+    /// use futures::TryStreamExt;
+    /// use opendal::EntryMode;
+    /// use opendal::Metakey;
+    /// use opendal::Operator;
+    /// use opendal::ops::OpList;
+    /// # #[tokio::main]
+    /// # async fn test(op: Operator) -> Result<()> {
+    /// let option = OpList::new().with_limit(10).with_start_after("start");
+    /// let mut ds = op.list_with("path/to/dir/", option).await?;
+    /// while let Some(mut de) = ds.try_next().await? {
+    ///     let meta = op.metadata(&de, Metakey::Mode).await?;
+    ///     match meta.mode() {
+    ///         EntryMode::FILE => {
+    ///             println!("Handling file")
+    ///         }
+    ///         EntryMode::DIR => {
+    ///             println!("Handling dir like start a new list via meta.path()")
+    ///         }
+    ///         EntryMode::Unknown => continue,
+    ///     }
+    /// }
+    /// # Ok(())
+    /// # }
+    /// ```
+    pub async fn list_with(&self, path: &str, op: OpList) -> Result<Lister> {
         let path = normalize_path(path);
 
         if !validate_path(&path, EntryMode::DIR) {
@@ -1094,7 +1133,7 @@ impl Operator {
             .with_context("path", &path));
         }
 
-        let (_, pager) = self.inner().list(&path, OpList::new()).await?;
+        let (_, pager) = self.inner().list(&path, op).await?;
 
         Ok(Lister::new(pager))
     }

core/src/types/ops.rs

@@ -55,6 +55,10 @@ pub struct OpList {
     /// The limit passed to underlying service to specify the max results
     /// that could return.
     limit: Option<usize>,
+
+    /// The start_after passes to underlying service to specify the specified key
+    /// to start listing from.
+    start_after: Option<String>,
 }
 
 impl OpList {
@@ -73,6 +77,17 @@ impl OpList {
     pub fn limit(&self) -> Option<usize> {
         self.limit
     }
+
+    /// Change the start_after of this list operation.
+    pub fn with_start_after(mut self, start_after: &str) -> Self {
+        self.start_after = Some(start_after.into());
+        self
+    }
+
+    /// Get the start_after of list operation.
+    pub fn start_after(&self) -> Option<&str> {
+        self.start_after.as_deref()
+    }
 }
 
 /// Args for `scan` operation.