mz_timely_util/
capture.rs

Help
1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License in the LICENSE file at the
6// root of this repository, or online at
7//
8//     http://www.apache.org/licenses/LICENSE-2.0
9//
10// Unless required by applicable law or agreed to in writing, software
11// distributed under the License is distributed on an "AS IS" BASIS,
12// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13// See the License for the specific language governing permissions and
14// limitations under the License.
15
16use mz_ore::channel::{InstrumentedChannelMetric, InstrumentedUnboundedSender};
17use timely::communication::Push;
18use timely::dataflow::operators::capture::{Event, EventPusher};
19
20pub struct UnboundedTokioCapture<T, C, M>(pub InstrumentedUnboundedSender<Event<T, C>, M>);
21
22impl<T, C, M> EventPusher<T, C> for UnboundedTokioCapture<T, C, M>
23where
24    M: InstrumentedChannelMetric,
25{
26    fn push(&mut self, event: Event<T, C>) {
27        // NOTE: An Err(x) result just means "data not accepted" most likely
28        //       because the receiver is gone. No need to panic.
29        let _ = self.0.send(event);
30    }
31}
32
33/// A helper type to allow capturing timely streams into timely pushers
34pub struct PusherCapture<P>(pub P);
35
36impl<P: Push<Event<T, D>>, T, D> EventPusher<T, D> for PusherCapture<P> {
37    fn push(&mut self, event: Event<T, D>) {
38        self.0.send(event);
39        self.0.done();
40    }
41}
mz_timely_util/capture.rs

mz_timely_util/
capture.rs
