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(splunk_hec source): Flush messages #5069

Merged
merged 3 commits into from Nov 17, 2020
Merged

fix(splunk_hec source): Flush messages #5069

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
fd11b20
fix(architecture): Ensure that Pipelines are flushed
jszwedko Nov 16, 2020
6ffb14e
Flush explicitely
jszwedko Nov 16, 2020
634af80
Merge remote-tracking branch 'origin/master' into flush-sinks
jszwedko Nov 17, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
5 changes: 5 additions & 0 deletions src/pipeline.rs
View file
Open in desktop
Expand Up @@ -60,6 +60,11 @@ impl Sink for Pipeline {
debug_assert!(self.enqueued.is_empty());
self.inner.poll_complete()
}

fn close(&mut self) -> Poll<(), Self::SinkError> {
self.poll_complete()?;
Copy link
Contributor

Choose a reason for hiding this comment

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

Does this change alone fix it by chance?

Copy link
Member Author

Choose a reason for hiding this comment

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

I had the same thought before; it did not.

self.inner.close()
}
}

impl Pipeline {
Expand Down
50 changes: 33 additions & 17 deletions src/sources/splunk_hec.rs
View file
Open in desktop
Expand Up @@ -171,23 +171,7 @@ impl SplunkSource {
host: Option<String>,
gzip: bool,
body: Bytes| {
let out = out.clone();
async move {
// Construct event parser
if gzip {
EventStream::new(GzDecoder::new(body.reader()), channel, host)
.forward(out.clone().sink_map_err(|_| ApiError::ServerShutdown))
.map(|_| ())
.compat()
.await
} else {
EventStream::new(body.reader(), channel, host)
.forward(out.clone().sink_map_err(|_| ApiError::ServerShutdown))
.map(|_| ())
.compat()
.await
}
}
process_service_request(out.clone(), channel, host, gzip, body)
},
)
.map(finish_ok)
Expand Down Expand Up @@ -319,6 +303,38 @@ impl SplunkSource {
}
}

async fn process_service_request(
out: Pipeline,
channel: Option<String>,
host: Option<String>,
gzip: bool,
body: Bytes,
) -> Result<(), Rejection> {
use futures::compat::Sink01CompatExt;
use futures::compat::Stream01CompatExt;
use futures::{SinkExt, StreamExt};

let mut out = out
.sink_map_err(|_| Rejection::from(ApiError::ServerShutdown))
.sink_compat();

let reader: Box<dyn Read + Send> = if gzip {
Box::new(GzDecoder::new(body.reader()))
} else {
Box::new(body.reader())
};

let stream = EventStream::new(reader, channel, host).compat();

let res = stream.forward(&mut out).await;

out.flush()
.map_err(|_| Rejection::from(ApiError::ServerShutdown))
.await?;

res.map(|_| ())
}

/// Constructs one ore more events from json-s coming from reader.
/// If errors, it's done with input.
struct EventStream<R: Read> {
Expand Down