chore: format logs

This commit is contained in:
nathan 2023-11-06 20:32:23 +08:00
parent 507d19628b
commit eb4814738b
1 changed files with 12 additions and 8 deletions

View File

@ -270,22 +270,16 @@ where
return None;
}
let mut merged_msg = vec![];
// If the message can merge other messages, try to merge the next message until the
// message is not mergeable.
if sending_msg.can_merge() {
let mut merged_msg = vec![];
while let Some(pending_msg) = pending_msg_queue.pop() {
// If the message is not mergeable, push the message back to the queue and break the loop.
match sending_msg.merge(&pending_msg, &self.config.maximum_payload_size) {
Ok(continue_merge) => {
merged_msg.push(pending_msg.msg_id());
if !continue_merge {
event!(
tracing::Level::TRACE,
"merge: {:?}, len: {}",
merged_msg,
sending_msg.get_msg().length()
);
break;
}
},
@ -300,9 +294,19 @@ where
sending_msg.set_ret(tx);
sending_msg.set_state(self.uid, MessageState::Processing);
let _ = self.state_notifier.send(SinkState::Syncing);
let collab_msg = sending_msg.get_msg().clone();
pending_msg_queue.push(sending_msg);
if !merged_msg.is_empty() {
event!(
tracing::Level::DEBUG,
"merge: {:?}, len: {}",
merged_msg,
collab_msg.length()
);
}
collab_msg
};
@ -456,7 +460,7 @@ impl Default for SinkConfig {
fn default() -> Self {
Self {
send_timeout: Duration::from_secs(DEFAULT_SYNC_TIMEOUT),
maximum_payload_size: 4096,
maximum_payload_size: 1024 * 64,
strategy: SinkStrategy::ASAP,
}
}