fluencelabs/rust-libp2p
1
0
Fork 0
mirror of https://github.com/fluencelabs/rust-libp2p synced 2025-06-14 02:21:21 +00:00
Code Issues Projects Releases Wiki Activity

feat: migrate to quick-protobuf

Browse Source 
Instead of relying on `protoc` and buildscripts, we generate the bindings using `pb-rs` and version them within our codebase. This makes for a better IDE integration, a faster build and an easier use of `rust-libp2p` because we don't force the `protoc` dependency onto them.

Resolves #3024.

Pull-Request: #3312.
This commit is contained in:
Miguel Guarniz
2023-03-02 05:45:07 -05:00
committed by GitHub
parent 4910160bea
commit db82e0210e
141 changed files with 3662 additions and 1276 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
protocols/gossipsub/src/behaviour.rs
View File

@ -32,7 +32,6 @@ use std::{
use futures::StreamExt;
use log::{debug, error, trace, warn};
use prometheus_client::registry::Registry;
use prost::Message as _;
use rand::{seq::SliceRandom, thread_rng};
use libp2p_core::{
@ -64,8 +63,9 @@ use crate::types::{
Subscription, SubscriptionAction,
};
use crate::types::{PeerConnections, PeerKind, Rpc};
use crate::{rpc_proto, TopicScoreParams};
use crate::{rpc_proto::proto, TopicScoreParams};
use crate::{PublishError, SubscriptionError, ValidationError};
use quick_protobuf::{MessageWrite, Writer};
use std::{cmp::Ordering::Equal, fmt::Debug};
use wasm_timer::Interval;
@ -179,8 +179,8 @@ impl From<MessageAuthenticity> for PublishConfig {
let public_key = keypair.public();
let key_enc = public_key.to_protobuf_encoding();
let key = if key_enc.len() <= 42 {
// The public key can be inlined in [`rpc_proto::Message::from`], so we don't include it
// specifically in the [`rpc_proto::Message::key`] field.
// The public key can be inlined in [`rpc_proto::proto::::Message::from`], so we don't include it
// specifically in the [`rpc_proto::proto::Message::key`] field.
None
} else {
// Include the protobuf encoding of the public key in the message.
@ -610,7 +610,7 @@ where
.into_protobuf();
// check that the size doesn't exceed the max transmission size
if event.encoded_len() > self.config.max_transmit_size() {
if event.get_size() > self.config.max_transmit_size() {
return Err(PublishError::MessageTooLarge);
}
@ -721,7 +721,7 @@ where
}
// Send to peers we know are subscribed to the topic.
let msg_bytes = event.encoded_len();
let msg_bytes = event.get_size();
for peer_id in recipient_peers.iter() {
trace!("Sending message to peer: {:?}", peer_id);
self.send_message(*peer_id, event.clone())?;
@ -1338,7 +1338,7 @@ where
}
.into_protobuf();
let msg_bytes = message.encoded_len();
let msg_bytes = message.get_size();
if self.send_message(*peer_id, message).is_err() {
error!("Failed to send cached messages. Messages too large");
@ -2733,7 +2733,7 @@ where
}
.into_protobuf();
let msg_bytes = event.encoded_len();
let msg_bytes = event.get_size();
for peer in recipient_peers.iter() {
debug!("Sending message: {:?} to peer {:?}", msg_id, peer);
self.send_message(*peer, event.clone())?;
@ -2764,7 +2764,7 @@ where
let sequence_number: u64 = rand::random();
let signature = {
let message = rpc_proto::Message {
let message = proto::Message {
from: Some(author.clone().to_bytes()),
data: Some(data.clone()),
seqno: Some(sequence_number.to_be_bytes().to_vec()),
@ -2773,10 +2773,12 @@ where
key: None,
};
let mut buf = Vec::with_capacity(message.encoded_len());
let mut buf = Vec::with_capacity(message.get_size());
let mut writer = Writer::new(&mut buf);
message
.encode(&mut buf)
.expect("Buffer has sufficient capacity");
.write_message(&mut writer)
.expect("Encoding to succeed");
// the signature is over the bytes "libp2p-pubsub:<protobuf-message>"
let mut signature_bytes = SIGNING_PREFIX.to_vec();
@ -2875,11 +2877,7 @@ where
/// Send a [`Rpc`] message to a peer. This will wrap the message in an arc if it
/// is not already an arc.
fn send_message(
&mut self,
peer_id: PeerId,
message: rpc_proto::Rpc,
) -> Result<(), PublishError> {
fn send_message(&mut self, peer_id: PeerId, message: proto::RPC) -> Result<(), PublishError> {
// If the message is oversized, try and fragment it. If it cannot be fragmented, log an
// error and drop the message (all individual messages should be small enough to fit in the
// max_transmit_size)
@ -2899,12 +2897,12 @@ where
// If a message is too large to be sent as-is, this attempts to fragment it into smaller RPC
// messages to be sent.
fn fragment_message(&self, rpc: rpc_proto::Rpc) -> Result<Vec<rpc_proto::Rpc>, PublishError> {
if rpc.encoded_len() < self.config.max_transmit_size() {
fn fragment_message(&self, rpc: proto::RPC) -> Result<Vec<proto::RPC>, PublishError> {
if rpc.get_size() < self.config.max_transmit_size() {
return Ok(vec![rpc]);
}
let new_rpc = rpc_proto::Rpc {
let new_rpc = proto::RPC {
subscriptions: Vec::new(),
publish: Vec::new(),
control: None,
@ -2920,7 +2918,7 @@ where
// create a new RPC if the new object plus 5% of its size (for length prefix
// buffers) exceeds the max transmit size.
if rpc_list[list_index].encoded_len() + (($object_size as f64) * 1.05) as usize
if rpc_list[list_index].get_size() + (($object_size as f64) * 1.05) as usize
> self.config.max_transmit_size()
&& rpc_list[list_index] != new_rpc
{
@ -2932,7 +2930,7 @@ where
macro_rules! add_item {
($object: ident, $type: ident ) => {
let object_size = $object.encoded_len();
let object_size = $object.get_size();
if object_size + 2 > self.config.max_transmit_size() {
// This should not be possible. All received and published messages have already
@ -2960,12 +2958,12 @@ where
// handle the control messages. If all are within the max_transmit_size, send them without
// fragmenting, otherwise, fragment the control messages
let empty_control = rpc_proto::ControlMessage::default();
let empty_control = proto::ControlMessage::default();
if let Some(control) = rpc.control.as_ref() {
if control.encoded_len() + 2 > self.config.max_transmit_size() {
if control.get_size() + 2 > self.config.max_transmit_size() {
// fragment the RPC
for ihave in &control.ihave {
let len = ihave.encoded_len();
let len = ihave.get_size();
create_or_add_rpc!(len);
rpc_list
.last_mut()
@ -2976,7 +2974,7 @@ where
.push(ihave.clone());
}
for iwant in &control.iwant {
let len = iwant.encoded_len();
let len = iwant.get_size();
create_or_add_rpc!(len);
rpc_list
.last_mut()
@ -2987,7 +2985,7 @@ where
.push(iwant.clone());
}
for graft in &control.graft {
let len = graft.encoded_len();
let len = graft.get_size();
create_or_add_rpc!(len);
rpc_list
.last_mut()
@ -2998,7 +2996,7 @@ where
.push(graft.clone());
}
for prune in &control.prune {
let len = prune.encoded_len();
let len = prune.get_size();
create_or_add_rpc!(len);
rpc_list
.last_mut()
@ -3009,7 +3007,7 @@ where
.push(prune.clone());
}
} else {
let len = control.encoded_len();
let len = control.get_size();
create_or_add_rpc!(len);
rpc_list.last_mut().expect("Always an element").control = Some(control.clone());
}
@ -3769,7 +3767,7 @@ mod local_test {
// Messages over the limit should be split
while rpc_proto.encoded_len() < max_transmit_size {
while rpc_proto.get_size() < max_transmit_size {
rpc.messages.push(test_message());
rpc_proto = rpc.clone().into_protobuf();
}
@ -3786,7 +3784,7 @@ mod local_test {
// all fragmented messages should be under the limit
for message in fragmented_messages {
assert!(
message.encoded_len() < max_transmit_size,
message.get_size() < max_transmit_size,
"all messages should be less than the transmission size"
);
}
@ -3813,7 +3811,7 @@ mod local_test {
.fragment_message(rpc_proto.clone())
.expect("Messages must be valid");
if rpc_proto.encoded_len() < max_transmit_size {
if rpc_proto.get_size() < max_transmit_size {
assert_eq!(
fragmented_messages.len(),
1,
@ -3829,12 +3827,12 @@ mod local_test {
// all fragmented messages should be under the limit
for message in fragmented_messages {
assert!(
message.encoded_len() < max_transmit_size,
"all messages should be less than the transmission size: list size {} max size{}", message.encoded_len(), max_transmit_size
message.get_size() < max_transmit_size,
"all messages should be less than the transmission size: list size {} max size{}", message.get_size(), max_transmit_size
);
// ensure they can all be encoded
let mut buf = bytes::BytesMut::with_capacity(message.encoded_len());
let mut buf = bytes::BytesMut::with_capacity(message.get_size());
codec.encode(message, &mut buf).unwrap()
}
}