Implement Clone for With combinator (#2290)

Consider the following use-case of mpsc channels. There are two
categories of producers which produce items of two distinct types,
later unified in one using `With` combinator:

```
let (sender, receiver) = mspc::channel(100);

let download_status = sender.clone().with(|item| {
    futures::future::ok(Status::Download(item))
});

let unpack_status = sender.clone().with(|item| {
    futures::future::ok(Status::Unpack(item))
});
```

It would be convenient for `With` combinator to implement `Clone`,
since the underlying sink, `futures::channel::mpsc::Sender`,
implements it.

futures-util/src/sink/with.rs

@@ -53,6 +53,22 @@ where Si: Sink<Item>,
     }
 }
 
+impl<Si, Item, U, Fut, F> Clone for With<Si, Item, U, Fut, F>
+where
+    Si: Clone,
+    F: Clone,
+    Fut: Clone,
+{
+    fn clone(&self) -> Self {
+        Self {
+            state: self.state.clone(),
+            sink: self.sink.clone(),
+            f: self.f.clone(),
+            _phantom: PhantomData,
+        }
+    }
+}
+
 // Forwarding impl of Stream from the underlying sink
 impl<S, Item, U, Fut, F> Stream for With<S, Item, U, Fut, F>
     where S: Stream + Sink<Item>,

futures/tests/sink.rs

@@ -483,6 +483,41 @@ fn with_flush_propagate() {
     })
 }
 
+// test that `Clone` is implemented on `with` sinks
+#[test]
+fn with_implements_clone() {
+    use futures::channel::mpsc;
+    use futures::executor::block_on;
+    use futures::future;
+    use futures::{SinkExt, StreamExt};
+
+    let (mut tx, rx) = mpsc::channel(5);
+
+    {
+        let mut is_positive = tx
+            .clone()
+            .with(|item| future::ok::<bool, mpsc::SendError>(item > 0));
+
+        let mut is_long = tx
+            .clone()
+            .with(|item: &str| future::ok::<bool, mpsc::SendError>(item.len() > 5));
+
+        block_on(is_positive.clone().send(-1)).unwrap();
+        block_on(is_long.clone().send("123456")).unwrap();
+        block_on(is_long.send("123")).unwrap();
+        block_on(is_positive.send(1)).unwrap();
+    }
+
+    block_on(tx.send(false)).unwrap();
+
+    block_on(tx.close()).unwrap();
+
+    assert_eq!(
+        block_on(rx.collect::<Vec<_>>()),
+        vec![false, true, false, true, false]
+    );
+}
+
 // test that a buffer is a no-nop around a sink that always accepts sends
 #[test]
 fn buffer_noop() {