Trait StreamsInterface 

pub trait StreamsInterface: ClientLike + Sized {
    // Provided methods
    fn xinfo_consumers<R, K, S>(
        &self,
        key: K,
        groupname: S,
    ) -> impl Future<Output = FredResult<R>> + Send
       where R: FromValue,
             K: Into<Key> + Send,
             S: Into<Str> + Send { ... }
    fn xinfo_groups<R, K>(
        &self,
        key: K,
    ) -> impl Future<Output = FredResult<R>> + Send
       where R: FromValue,
             K: Into<Key> + Send { ... }
    fn xinfo_stream<R, K>(
        &self,
        key: K,
        full: bool,
        count: Option<u64>,
    ) -> impl Future<Output = FredResult<R>> + Send
       where R: FromValue,
             K: Into<Key> + Send { ... }
    fn xadd<R, K, C, I, F>(
        &self,
        key: K,
        nomkstream: bool,
        cap: C,
        id: I,
        fields: F,
    ) -> impl Future<Output = FredResult<R>> + Send
       where R: FromValue,
             K: Into<Key> + Send,
             I: Into<XID> + Send,
             F: TryInto<MultipleOrderedPairs> + Send,
             F::Error: Into<Error> + Send,
             C: TryInto<XCap> + Send,
             C::Error: Into<Error> + Send { ... }
    fn xtrim<R, K, C>(
        &self,
        key: K,
        cap: C,
    ) -> impl Future<Output = FredResult<R>> + Send
       where R: FromValue,
             K: Into<Key> + Send,
             C: TryInto<XCap> + Send,
             C::Error: Into<Error> + Send { ... }
    fn xdel<R, K, S>(
        &self,
        key: K,
        ids: S,
    ) -> impl Future<Output = FredResult<R>> + Send
       where R: FromValue,
             K: Into<Key> + Send,
             S: Into<MultipleStrings> + Send { ... }
    fn xrange_values<Ri, Rk, Rv, K, S, E>(
        &self,
        key: K,
        start: S,
        end: E,
        count: Option<u64>,
    ) -> impl Future<Output = FredResult<Vec<XReadValue<Ri, Rk, Rv>>>> + Send
       where Ri: FromValue,
             Rk: FromKey + Hash + Eq,
             Rv: FromValue,
             K: Into<Key> + Send,
             S: TryInto<Value> + Send,
             S::Error: Into<Error> + Send,
             E: TryInto<Value> + Send,
             E::Error: Into<Error> + Send { ... }
    fn xrange<R, K, S, E>(
        &self,
        key: K,
        start: S,
        end: E,
        count: Option<u64>,
    ) -> impl Future<Output = FredResult<R>> + Send
       where R: FromValue,
             K: Into<Key> + Send,
             S: TryInto<Value> + Send,
             S::Error: Into<Error> + Send,
             E: TryInto<Value> + Send,
             E::Error: Into<Error> + Send { ... }
    fn xrevrange_values<Ri, Rk, Rv, K, E, S>(
        &self,
        key: K,
        end: E,
        start: S,
        count: Option<u64>,
    ) -> impl Future<Output = FredResult<Vec<XReadValue<Ri, Rk, Rv>>>> + Send
       where Ri: FromValue,
             Rk: FromKey + Hash + Eq,
             Rv: FromValue,
             K: Into<Key> + Send,
             S: TryInto<Value> + Send,
             S::Error: Into<Error> + Send,
             E: TryInto<Value> + Send,
             E::Error: Into<Error> + Send { ... }
    fn xrevrange<R, K, S, E>(
        &self,
        key: K,
        end: E,
        start: S,
        count: Option<u64>,
    ) -> impl Future<Output = FredResult<R>> + Send
       where R: FromValue,
             K: Into<Key> + Send,
             S: TryInto<Value> + Send,
             S::Error: Into<Error> + Send,
             E: TryInto<Value> + Send,
             E::Error: Into<Error> + Send { ... }
    fn xlen<R, K>(&self, key: K) -> impl Future<Output = FredResult<R>> + Send
       where R: FromValue,
             K: Into<Key> + Send { ... }
    fn xread_map<Rk1, Rk2, Rk3, Rv, K, I>(
        &self,
        count: Option<u64>,
        block: Option<u64>,
        keys: K,
        ids: I,
    ) -> impl Future<Output = FredResult<XReadResponse<Rk1, Rk2, Rk3, Rv>>> + Send
       where Rk1: FromKey + Hash + Eq,
             Rk2: FromValue,
             Rk3: FromKey + Hash + Eq,
             Rv: FromValue,
             K: Into<MultipleKeys> + Send,
             I: Into<MultipleIDs> + Send { ... }
    fn xread<R, K, I>(
        &self,
        count: Option<u64>,
        block: Option<u64>,
        keys: K,
        ids: I,
    ) -> impl Future<Output = FredResult<R>> + Send
       where R: FromValue,
             K: Into<MultipleKeys> + Send,
             I: Into<MultipleIDs> + Send { ... }
    fn xgroup_create<R, K, S, I>(
        &self,
        key: K,
        groupname: S,
        id: I,
        mkstream: bool,
    ) -> impl Future<Output = FredResult<R>> + Send
       where R: FromValue,
             K: Into<Key> + Send,
             S: Into<Str> + Send,
             I: Into<XID> + Send { ... }
    fn xgroup_createconsumer<R, K, G, C>(
        &self,
        key: K,
        groupname: G,
        consumername: C,
    ) -> impl Future<Output = FredResult<R>> + Send
       where R: FromValue,
             K: Into<Key> + Send,
             G: Into<Str> + Send,
             C: Into<Str> + Send { ... }
    fn xgroup_delconsumer<R, K, G, C>(
        &self,
        key: K,
        groupname: G,
        consumername: C,
    ) -> impl Future<Output = FredResult<R>> + Send
       where R: FromValue,
             K: Into<Key> + Send,
             G: Into<Str> + Send,
             C: Into<Str> + Send { ... }
    fn xgroup_destroy<R, K, S>(
        &self,
        key: K,
        groupname: S,
    ) -> impl Future<Output = FredResult<R>> + Send
       where R: FromValue,
             K: Into<Key> + Send,
             S: Into<Str> + Send { ... }
    fn xgroup_setid<R, K, S, I>(
        &self,
        key: K,
        groupname: S,
        id: I,
    ) -> impl Future<Output = FredResult<R>> + Send
       where R: FromValue,
             K: Into<Key> + Send,
             S: Into<Str> + Send,
             I: Into<XID> + Send { ... }
    fn xreadgroup_map<Rk1, Rk2, Rk3, Rv, G, C, K, I>(
        &self,
        group: G,
        consumer: C,
        count: Option<u64>,
        block: Option<u64>,
        noack: bool,
        keys: K,
        ids: I,
    ) -> impl Future<Output = FredResult<XReadResponse<Rk1, Rk2, Rk3, Rv>>> + Send
       where Rk1: FromKey + Hash + Eq,
             Rk2: FromValue,
             Rk3: FromKey + Hash + Eq,
             Rv: FromValue,
             G: Into<Str> + Send,
             C: Into<Str> + Send,
             K: Into<MultipleKeys> + Send,
             I: Into<MultipleIDs> + Send { ... }
    fn xreadgroup<R, G, C, K, I>(
        &self,
        group: G,
        consumer: C,
        count: Option<u64>,
        block: Option<u64>,
        noack: bool,
        keys: K,
        ids: I,
    ) -> impl Future<Output = FredResult<R>> + Send
       where R: FromValue,
             G: Into<Str> + Send,
             C: Into<Str> + Send,
             K: Into<MultipleKeys> + Send,
             I: Into<MultipleIDs> + Send { ... }
    fn xack<R, K, G, I>(
        &self,
        key: K,
        group: G,
        ids: I,
    ) -> impl Future<Output = FredResult<R>> + Send
       where R: FromValue,
             K: Into<Key> + Send,
             G: Into<Str> + Send,
             I: Into<MultipleIDs> + Send { ... }
    fn xclaim_values<Ri, Rk, Rv, K, G, C, I>(
        &self,
        key: K,
        group: G,
        consumer: C,
        min_idle_time: u64,
        ids: I,
        idle: Option<u64>,
        time: Option<u64>,
        retry_count: Option<u64>,
        force: bool,
        justid: bool,
    ) -> impl Future<Output = FredResult<Vec<XReadValue<Ri, Rk, Rv>>>> + Send
       where Ri: FromValue,
             Rk: FromKey + Hash + Eq,
             Rv: FromValue,
             K: Into<Key> + Send,
             G: Into<Str> + Send,
             C: Into<Str> + Send,
             I: Into<MultipleIDs> + Send { ... }
    fn xclaim<R, K, G, C, I>(
        &self,
        key: K,
        group: G,
        consumer: C,
        min_idle_time: u64,
        ids: I,
        idle: Option<u64>,
        time: Option<u64>,
        retry_count: Option<u64>,
        force: bool,
        justid: bool,
    ) -> impl Future<Output = FredResult<R>> + Send
       where R: FromValue,
             K: Into<Key> + Send,
             G: Into<Str> + Send,
             C: Into<Str> + Send,
             I: Into<MultipleIDs> + Send { ... }
    fn xautoclaim_values<Ri, Rk, Rv, K, G, C, I>(
        &self,
        key: K,
        group: G,
        consumer: C,
        min_idle_time: u64,
        start: I,
        count: Option<u64>,
        justid: bool,
    ) -> impl Future<Output = FredResult<(String, Vec<XReadValue<Ri, Rk, Rv>>)>> + Send
       where Ri: FromValue,
             Rk: FromKey + Hash + Eq,
             Rv: FromValue,
             K: Into<Key> + Send,
             G: Into<Str> + Send,
             C: Into<Str> + Send,
             I: Into<XID> + Send { ... }
    fn xautoclaim<R, K, G, C, I>(
        &self,
        key: K,
        group: G,
        consumer: C,
        min_idle_time: u64,
        start: I,
        count: Option<u64>,
        justid: bool,
    ) -> impl Future<Output = FredResult<R>> + Send
       where R: FromValue,
             K: Into<Key> + Send,
             G: Into<Str> + Send,
             C: Into<Str> + Send,
             I: Into<XID> + Send { ... }
    fn xpending<R, K, G, A>(
        &self,
        key: K,
        group: G,
        args: A,
    ) -> impl Future<Output = FredResult<R>> + Send
       where R: FromValue,
             K: Into<Key> + Send,
             G: Into<Str> + Send,
             A: Into<XPendingArgs> + Send { ... }
}

Available on crate feature i-streams only.

Functions that implement the streams interface.

Note: Several of the stream commands can return types with verbose type declarations. Additionally, certain commands can be parsed differently in RESP2 and RESP3 modes. Functions such as xread_map, xreadgroup_map, xrange_values, etc exist to make this easier for callers. These functions apply an additional layer of parsing logic that can make declaring response types easier, as well as automatically handling any differences between RESP2 and RESP3 return value types.

Provided Methods

fn xinfo_consumers<R, K, S>( &self, key: K, groupname: S, ) -> impl Future<Output = FredResult<R>> + Send

where R: FromValue, K: Into<Key> + Send, S: Into<Str> + Send,

This command returns the list of consumers that belong to the groupname consumer group of the stream stored at key.

https://redis.io/commands/xinfo-consumers

fn xinfo_groups<R, K>( &self, key: K, ) -> impl Future<Output = FredResult<R>> + Send

where R: FromValue, K: Into<Key> + Send,

This command returns the list of all consumers groups of the stream stored at key.

https://redis.io/commands/xinfo-groups

fn xinfo_stream<R, K>( &self, key: K, full: bool, count: Option<u64>, ) -> impl Future<Output = FredResult<R>> + Send

where R: FromValue, K: Into<Key> + Send,

This command returns information about the stream stored at key.

https://redis.io/commands/xinfo-stream

fn xadd<R, K, C, I, F>( &self, key: K, nomkstream: bool, cap: C, id: I, fields: F, ) -> impl Future<Output = FredResult<R>> + Send

where R: FromValue, K: Into<Key> + Send, I: Into<XID> + Send, F: TryInto<MultipleOrderedPairs> + Send, F::Error: Into<Error> + Send, C: TryInto<XCap> + Send, C::Error: Into<Error> + Send,

Appends the specified stream entry to the stream at the specified key. If the key does not exist, as a side effect of running this command the key is created with a stream value. The creation of stream’s key can be disabled with the NOMKSTREAM option.

https://redis.io/commands/xadd

fn xtrim<R, K, C>( &self, key: K, cap: C, ) -> impl Future<Output = FredResult<R>> + Send

where R: FromValue, K: Into<Key> + Send, C: TryInto<XCap> + Send, C::Error: Into<Error> + Send,

Trims the stream by evicting older entries (entries with lower IDs) if needed.

https://redis.io/commands/xtrim

fn xdel<R, K, S>( &self, key: K, ids: S, ) -> impl Future<Output = FredResult<R>> + Send

where R: FromValue, K: Into<Key> + Send, S: Into<MultipleStrings> + Send,

Removes the specified entries from a stream, and returns the number of entries deleted.

https://redis.io/commands/xdel

fn xrange_values<Ri, Rk, Rv, K, S, E>( &self, key: K, start: S, end: E, count: Option<u64>, ) -> impl Future<Output = FredResult<Vec<XReadValue<Ri, Rk, Rv>>>> + Send

where Ri: FromValue, Rk: FromKey + Hash + Eq, Rv: FromValue, K: Into<Key> + Send, S: TryInto<Value> + Send, S::Error: Into<Error> + Send, E: TryInto<Value> + Send, E::Error: Into<Error> + Send,

Return the stream entries matching the provided range of IDs, automatically converting to a less verbose type definition.

https://redis.io/commands/xrange

fn xrange<R, K, S, E>( &self, key: K, start: S, end: E, count: Option<u64>, ) -> impl Future<Output = FredResult<R>> + Send

where R: FromValue, K: Into<Key> + Send, S: TryInto<Value> + Send, S::Error: Into<Error> + Send, E: TryInto<Value> + Send, E::Error: Into<Error> + Send,

The command returns the stream entries matching a given range of IDs. The range is specified by a minimum and maximum ID. All the entries having an ID between the two specified or exactly one of the two IDs specified (closed interval) are returned.

https://redis.io/commands/xrange

See xrange_values for a variation of this function that may be more useful.

fn xrevrange_values<Ri, Rk, Rv, K, E, S>( &self, key: K, end: E, start: S, count: Option<u64>, ) -> impl Future<Output = FredResult<Vec<XReadValue<Ri, Rk, Rv>>>> + Send

where Ri: FromValue, Rk: FromKey + Hash + Eq, Rv: FromValue, K: Into<Key> + Send, S: TryInto<Value> + Send, S::Error: Into<Error> + Send, E: TryInto<Value> + Send, E::Error: Into<Error> + Send,

Similar to XRANGE, but with the results returned in reverse order. The results will be automatically converted to a less verbose type definition.

https://redis.io/commands/xrevrange

fn xrevrange<R, K, S, E>( &self, key: K, end: E, start: S, count: Option<u64>, ) -> impl Future<Output = FredResult<R>> + Send

where R: FromValue, K: Into<Key> + Send, S: TryInto<Value> + Send, S::Error: Into<Error> + Send, E: TryInto<Value> + Send, E::Error: Into<Error> + Send,

Similar to XRANGE, but with the results returned in reverse order.

https://redis.io/commands/xrevrange

See the xrevrange_values for a variation of this function that may be more useful.

fn xlen<R, K>(&self, key: K) -> impl Future<Output = FredResult<R>> + Send

where R: FromValue, K: Into<Key> + Send,

Returns the number of entries inside a stream.

https://redis.io/commands/xlen

fn xread_map<Rk1, Rk2, Rk3, Rv, K, I>( &self, count: Option<u64>, block: Option<u64>, keys: K, ids: I, ) -> impl Future<Output = FredResult<XReadResponse<Rk1, Rk2, Rk3, Rv>>> + Send

where Rk1: FromKey + Hash + Eq, Rk2: FromValue, Rk3: FromKey + Hash + Eq, Rv: FromValue, K: Into<MultipleKeys> + Send, I: Into<MultipleIDs> + Send,

Read data from one or multiple streams, only returning entries with an ID greater than the last received ID reported by the caller.

https://redis.io/commands/xread

The XREAD and XREADGROUP commands return values that can be interpreted differently in RESP2 and RESP3 mode. In many cases it is also easier to operate on the return values of these functions as a HashMap, but manually declaring this type can be very verbose. This function will automatically convert the response to the most common map representation while also handling the encoding differences between RESP2 and RESP3.

async fn example(client: Client) -> Result<(), Error> {
  // borrowed from the tests. XREAD and XREADGROUP are very similar.
  let result: XReadResponse<String, String, String, usize> = client  
    .xreadgroup_map("group1", "consumer1", None, None, false, "foo", ">")
    .await?;
  println!("Result: {:?}", result);    
  // Result: {"foo": [("1646240801081-0", {"count": 0}), ("1646240801082-0", {"count": 1}), ("1646240801082-1", {"count": 2})]}

  assert_eq!(result.len(), 1);
  for (idx, (id, record)) in result.get("foo").unwrap().into_iter().enumerate() {
    let value = record.get("count").expect("Failed to read count");
    assert_eq!(idx, *value);
  }

  Ok(())
}

fn xread<R, K, I>( &self, count: Option<u64>, block: Option<u64>, keys: K, ids: I, ) -> impl Future<Output = FredResult<R>> + Send

where R: FromValue, K: Into<MultipleKeys> + Send, I: Into<MultipleIDs> + Send,

Read data from one or multiple streams, only returning entries with an ID greater than the last received ID reported by the caller.

https://redis.io/commands/xread

See xread_map for more information on a variation of this function that might be more useful.

fn xgroup_create<R, K, S, I>( &self, key: K, groupname: S, id: I, mkstream: bool, ) -> impl Future<Output = FredResult<R>> + Send

where R: FromValue, K: Into<Key> + Send, S: Into<Str> + Send, I: Into<XID> + Send,

This command creates a new consumer group uniquely identified by groupname for the stream stored at key.

https://redis.io/commands/xgroup-create

fn xgroup_createconsumer<R, K, G, C>( &self, key: K, groupname: G, consumername: C, ) -> impl Future<Output = FredResult<R>> + Send

where R: FromValue, K: Into<Key> + Send, G: Into<Str> + Send, C: Into<Str> + Send,

Create a consumer named consumername in the consumer group groupname of the stream that’s stored at key.

https://redis.io/commands/xgroup-createconsumer

fn xgroup_delconsumer<R, K, G, C>( &self, key: K, groupname: G, consumername: C, ) -> impl Future<Output = FredResult<R>> + Send

where R: FromValue, K: Into<Key> + Send, G: Into<Str> + Send, C: Into<Str> + Send,

Delete a consumer named consumername in the consumer group groupname of the stream that’s stored at key.

https://redis.io/commands/xgroup-delconsumer

fn xgroup_destroy<R, K, S>( &self, key: K, groupname: S, ) -> impl Future<Output = FredResult<R>> + Send

where R: FromValue, K: Into<Key> + Send, S: Into<Str> + Send,

Completely destroy a consumer group.

https://redis.io/commands/xgroup-destroy

fn xgroup_setid<R, K, S, I>( &self, key: K, groupname: S, id: I, ) -> impl Future<Output = FredResult<R>> + Send

where R: FromValue, K: Into<Key> + Send, S: Into<Str> + Send, I: Into<XID> + Send,

Set the last delivered ID for a consumer group.

https://redis.io/commands/xgroup-setid

fn xreadgroup_map<Rk1, Rk2, Rk3, Rv, G, C, K, I>( &self, group: G, consumer: C, count: Option<u64>, block: Option<u64>, noack: bool, keys: K, ids: I, ) -> impl Future<Output = FredResult<XReadResponse<Rk1, Rk2, Rk3, Rv>>> + Send

where Rk1: FromKey + Hash + Eq, Rk2: FromValue, Rk3: FromKey + Hash + Eq, Rv: FromValue, G: Into<Str> + Send, C: Into<Str> + Send, K: Into<MultipleKeys> + Send, I: Into<MultipleIDs> + Send,

A special version of the XREAD command with support for consumer groups.

Declaring proper type declarations for this command can be complicated due to the complex nature of the response values and the differences between RESP2 and RESP3. See the xread documentation for more information.

https://redis.io/commands/xreadgroup

The XREAD and XREADGROUP commands return values that can be interpreted differently in RESP2 and RESP3 mode. In many cases it is also easier to operate on the return values of these functions as a HashMap, but manually declaring this type can be very verbose. This function will automatically convert the response to the most common map representation while also handling the encoding differences between RESP2 and RESP3.

See the xread_map documentation for more information.

fn xreadgroup<R, G, C, K, I>( &self, group: G, consumer: C, count: Option<u64>, block: Option<u64>, noack: bool, keys: K, ids: I, ) -> impl Future<Output = FredResult<R>> + Send

where R: FromValue, G: Into<Str> + Send, C: Into<Str> + Send, K: Into<MultipleKeys> + Send, I: Into<MultipleIDs> + Send,

A special version of the XREAD command with support for consumer groups.

Declaring proper type declarations for this command can be complicated due to the complex nature of the response values and the differences between RESP2 and RESP3. See the xread documentation for more information.

https://redis.io/commands/xreadgroup

See xreadgroup_map for a variation of this function that might be more useful.

fn xack<R, K, G, I>( &self, key: K, group: G, ids: I, ) -> impl Future<Output = FredResult<R>> + Send

where R: FromValue, K: Into<Key> + Send, G: Into<Str> + Send, I: Into<MultipleIDs> + Send,

Remove one or more messages from the Pending Entries List (PEL) of a stream consumer group.

https://redis.io/commands/xack

fn xclaim_values<Ri, Rk, Rv, K, G, C, I>( &self, key: K, group: G, consumer: C, min_idle_time: u64, ids: I, idle: Option<u64>, time: Option<u64>, retry_count: Option<u64>, force: bool, justid: bool, ) -> impl Future<Output = FredResult<Vec<XReadValue<Ri, Rk, Rv>>>> + Send

where Ri: FromValue, Rk: FromKey + Hash + Eq, Rv: FromValue, K: Into<Key> + Send, G: Into<Str> + Send, C: Into<Str> + Send, I: Into<MultipleIDs> + Send,

A variation of xclaim with a less verbose return type.

fn xclaim<R, K, G, C, I>( &self, key: K, group: G, consumer: C, min_idle_time: u64, ids: I, idle: Option<u64>, time: Option<u64>, retry_count: Option<u64>, force: bool, justid: bool, ) -> impl Future<Output = FredResult<R>> + Send

where R: FromValue, K: Into<Key> + Send, G: Into<Str> + Send, C: Into<Str> + Send, I: Into<MultipleIDs> + Send,

In the context of a stream consumer group, this command changes the ownership of a pending message, so that the new owner is the consumer specified as the command argument.

https://redis.io/commands/xclaim

See xclaim_values for a variation of this function that might be more useful.

fn xautoclaim_values<Ri, Rk, Rv, K, G, C, I>( &self, key: K, group: G, consumer: C, min_idle_time: u64, start: I, count: Option<u64>, justid: bool, ) -> impl Future<Output = FredResult<(String, Vec<XReadValue<Ri, Rk, Rv>>)>> + Send

where Ri: FromValue, Rk: FromKey + Hash + Eq, Rv: FromValue, K: Into<Key> + Send, G: Into<Str> + Send, C: Into<Str> + Send, I: Into<XID> + Send,

This command transfers ownership of pending stream entries that match the specified criteria. It also converts the response type to a less verbose type declaration and handles potential differences between RESP2 and RESP3.

https://redis.io/commands/xautoclaim

fn xautoclaim<R, K, G, C, I>( &self, key: K, group: G, consumer: C, min_idle_time: u64, start: I, count: Option<u64>, justid: bool, ) -> impl Future<Output = FredResult<R>> + Send

where R: FromValue, K: Into<Key> + Send, G: Into<Str> + Send, C: Into<Str> + Send, I: Into<XID> + Send,

This command transfers ownership of pending stream entries that match the specified criteria.

https://redis.io/commands/xautoclaim

Note: See xautoclaim_values for a variation of this function that may be more useful.

fn xpending<R, K, G, A>( &self, key: K, group: G, args: A, ) -> impl Future<Output = FredResult<R>> + Send

where R: FromValue, K: Into<Key> + Send, G: Into<Str> + Send, A: Into<XPendingArgs> + Send,

Inspect the list of pending messages in a consumer group.

https://redis.io/commands/xpending

Dyn Compatibility

This trait is not dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety", so this trait is not object safe.

Implementors

impl StreamsInterface for Client

impl StreamsInterface for Pool

impl StreamsInterface for SubscriberClient

impl StreamsInterface for Transaction

impl<C: ClientLike> StreamsInterface for Replicas<C>

impl<C: StreamsInterface> StreamsInterface for Pipeline<C>

impl<C: StreamsInterface> StreamsInterface for WithOptions<C>