Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

fix: Switch to futures-0.3 channels #6283

Merged
merged 7 commits into from
Feb 7, 2021
Merged

fix: Switch to futures-0.3 channels #6283

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
4fa5900
Use future03 channel
ktff Jan 29, 2021
56a3007
Fuse channel
ktff Jan 31, 2021
df6d88e
Merge branch 'master' into ktff/future03_channel
ktff Feb 4, 2021
592cbf9
Merge remote-tracking branch 'origin/master' into ktff/future03_channel
jszwedko Feb 4, 2021
0a2140a
Bump
ktff Feb 5, 2021
0fe1771
Add bench futures channel
ktff Feb 7, 2021
9630730
Fmt
ktff Feb 7, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 5 additions & 4 deletions src/buffers/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
use crate::{config::Resource, sink::BoundedSink, Event};
use crate::{config::Resource, Event};
#[cfg(feature = "leveldb")]
use futures::compat::{Sink01CompatExt, Stream01CompatExt};
use futures::{Sink, Stream};
use futures::{channel::mpsc, Sink, SinkExt, Stream};
use futures01::task::AtomicTask;
use pin_project::pin_project;
use serde::{Deserialize, Serialize};
Expand All @@ -16,7 +16,6 @@ use std::{
};
#[cfg(feature = "leveldb")]
use tokio::stream::StreamExt;
use tokio::sync::mpsc;

#[cfg(feature = "leveldb")]
pub mod disk;
Expand Down Expand Up @@ -72,7 +71,9 @@ impl BufferInputCloner {
pub fn get(&self) -> Box<dyn Sink<Event, Error = ()> + Send> {
match self {
BufferInputCloner::Memory(tx, when_full) => {
let inner = BoundedSink::new(tx.clone());
let inner = tx
.clone()
.sink_map_err(|error| error!(message = "Sender error.", %error));
if when_full == &WhenFull::DropNewest {
Box::new(DropWhenFull::new(inner))
} else {
Expand Down
9 changes: 3 additions & 6 deletions src/topology/builder.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,10 +19,7 @@ use std::{
sync::{Arc, Mutex},
};
use stream_cancel::{StreamExt as StreamCancelExt, Trigger, Tripwire};
use tokio::{
sync::mpsc,
time::{timeout, Duration},
};
use tokio::time::{timeout, Duration};

pub struct Pieces {
pub inputs: HashMap<String, (buffers::BufferInputCloner, Vec<String>)>,
Expand Down Expand Up @@ -56,7 +53,7 @@ pub async fn build_pieces(
.iter()
.filter(|(name, _)| diff.sources.contains_new(&name))
{
let (tx, rx) = mpsc::channel(1000);
let (tx, rx) = tokio::sync::mpsc::channel(1000);
ktff marked this conversation as resolved.
Show resolved Hide resolved
let pipeline = Pipeline::from_sender(tx, vec![]);

let typetag = source.source_type();
Expand Down Expand Up @@ -115,7 +112,7 @@ pub async fn build_pieces(
Ok(transform) => transform,
};

let (input_tx, input_rx) = mpsc::channel(100);
let (input_tx, input_rx) = futures::channel::mpsc::channel(100);
let input_tx = buffers::BufferInputCloner::Memory(input_tx, buffers::WhenFull::Block);

let (output, control) = Fanout::new();
Expand Down
1 change: 1 addition & 0 deletions src/transforms/util/runtime_transform.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,6 +81,7 @@ where
Box::pin(
input_rx
.map(Message::Process)
.fuse()
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixes a bug where if no events are sent before shutting down this, then into_future will poll this and get None which it will pass with the rest of the stream which will be polled again later therefor breaking the stream contract of not calling it after first None.

.into_future()
.map(move |(first, rest)| {
// The first message is always `Message::Init`.
Expand Down