Skip to content

fix(websocket sink): send encoded message as binary frame #18060

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

Merged
neuronull merged 9 commits into from
Jul 25, 2023
Merged

fix(websocket sink): send encoded message as binary frame #18060

Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
0ef5d5f
fix(websocket sink): send encoded message as binary frame
zhongchen Jul 21, 2023
fd5e83b
fix: add an extra flag to decide frame format
zhongchen Jul 24, 2023
6a46b47
fix: address comments
zhongchen Jul 24, 2023
155b86f
fix: binary frames for avro and native codecs
zhongchen Jul 25, 2023
b932f08
fix: list all codes instead of using default
zhongchen Jul 25, 2023
2628ea1
fix: use use statement
zhongchen Jul 25, 2023
4f77f78
fix: extract the logic into a function
zhongchen Jul 25, 2023
34b4420
fix: make the func const
zhongchen Jul 25, 2023
fcb8d26
fix: fix format
zhongchen Jul 25, 2023
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
19 changes: 18 additions & 1 deletion src/sinks/websocket/sink.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -236,6 +236,17 @@ impl WebSocketSink {
Ok(())
}

const fn should_encode_as_binary(&self) -> bool {
use codecs::encoding::Serializer::{
Avro, Csv, Gelf, Json, Logfmt, Native, NativeJson, RawMessage, Text,
};

match self.encoder.serializer() {
RawMessage(_) | Avro(_) | Native(_) => true,
Csv(_) | Logfmt(_) | Gelf(_) | Json(_) | Text(_) | NativeJson(_) => false,
}
}

async fn handle_events<I, WS, O>(
&mut self,
input: &mut I,
Expand All @@ -261,6 +272,7 @@ impl WebSocketSink {

let bytes_sent = register!(BytesSent::from(Protocol("websocket".into())));
let events_sent = register!(EventsSent::from(Output(None)));
let encode_as_binary = self.should_encode_as_binary();

loop {
let result = tokio::select! {
Expand Down Expand Up @@ -301,7 +313,12 @@ impl WebSocketSink {
Ok(()) => {
finalizers.update_status(EventStatus::Delivered);

let message = Message::text(String::from_utf8_lossy(&bytes));
let message = if encode_as_binary {
Message::binary(bytes)
}
else {
Message::text(String::from_utf8_lossy(&bytes))
};
let message_len = message.len();

ws_sink.send(message).await.map(|_| {
Expand Down