1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94
use futures::Stream; use std::{ pin::Pin, task::{Context, Poll} }; use sum::Sum2; use super::{ParallelPipe, ParallelStream, PipeTask, PipeTaskAsync, StreamTask, StreamTaskAsync}; use crate::sink::Sink; impl_par_dist! { impl<A: ParallelStream, B: ParallelStream<Item = A::Item>> ParallelStream for Sum2<A, B> { type Item = A::Item; type Task = Sum2<A::Task, B::Task>; fn size_hint(&self) -> (usize, Option<usize>) { match self { Self::A(i) => i.size_hint(), Self::B(i) => i.size_hint(), } } fn next_task(&mut self) -> Option<Self::Task> { match self { Self::A(i) => i.next_task().map(Sum2::A), Self::B(i) => i.next_task().map(Sum2::B), } } } impl<A: ParallelPipe<Source>, B: ParallelPipe<Source, Item = A::Item>, Source> ParallelPipe<Source> for Sum2<A, B> { type Item = A::Item; type Task = Sum2<A::Task, B::Task>; fn task(&self) -> Self::Task { match self { Self::A(i) => Sum2::A(i.task()), Self::B(i) => Sum2::B(i.task()), } } } } impl<A: StreamTask, B: StreamTask<Item = A::Item>> StreamTask for Sum2<A, B> { type Item = A::Item; type Async = Sum2<A::Async, B::Async>; fn into_async(self) -> Self::Async { match self { Sum2::A(a) => Sum2::A(a.into_async()), Sum2::B(b) => Sum2::B(b.into_async()), } } } impl<A: PipeTask<Source>, B: PipeTask<Source, Item = A::Item>, Source> PipeTask<Source> for Sum2<A, B> { type Item = A::Item; type Async = Sum2<A::Async, B::Async>; fn into_async(self) -> Self::Async { match self { Sum2::A(a) => Sum2::A(a.into_async()), Sum2::B(b) => Sum2::B(b.into_async()), } } } impl<A: StreamTaskAsync, B: StreamTaskAsync<Item = A::Item>> StreamTaskAsync for Sum2<A, B> { type Item = A::Item; fn poll_run( self: Pin<&mut Self>, cx: &mut Context, sink: Pin<&mut impl Sink<Item = Self::Item>>, ) -> Poll<()> { match self.as_pin_mut() { Sum2::A(task) => task.poll_run(cx, sink), Sum2::B(task) => task.poll_run(cx, sink), } } } impl<A: PipeTaskAsync<Source>, B: PipeTaskAsync<Source, Item = A::Item>, Source> PipeTaskAsync<Source> for Sum2<A, B> { type Item = A::Item; fn poll_run( self: Pin<&mut Self>, cx: &mut Context, stream: Pin<&mut impl Stream<Item = Source>>, sink: Pin<&mut impl Sink<Item = Self::Item>>, ) -> Poll<()> { match self.as_pin_mut() { Sum2::A(task) => task.poll_run(cx, stream, sink), Sum2::B(task) => task.poll_run(cx, stream, sink), } } }