mirror of
https://github.com/fluencelabs/rust-libp2p
synced 2025-06-12 09:31:20 +00:00
Replace protobuf crate with prost! (#1390)
* Replace protobuf crate with prost! * Add copyright headers to build.rs files. * kad: Fix error when mapping connection types. Co-Authored-By: Pierre Krieger <pierre.krieger1708@gmail.com> * Fix more mapping mistakes. Co-authored-by: Pierre Krieger <pierre.krieger1708@gmail.com>
This commit is contained in:
Toralf Wittner
committed by
Pierre Krieger
Pierre Krieger
parent
9d2df148cd
commit
680c467f7e
@ -24,9 +24,12 @@
|
||||
pub mod protocol;
|
||||
|
||||
mod layer;
|
||||
mod rpc_proto;
|
||||
mod topic;
|
||||
|
||||
mod rpc_proto {
|
||||
include!(concat!(env!("OUT_DIR"), "/floodsub.pb.rs"));
|
||||
}
|
||||
|
||||
pub use self::layer::{Floodsub, FloodsubEvent};
|
||||
pub use self::protocol::{FloodsubMessage, FloodsubRpc};
|
||||
pub use self::topic::{Topic, TopicBuilder, TopicHash};
|
||||
|
||||
@ -22,7 +22,7 @@ use crate::rpc_proto;
|
||||
use crate::topic::TopicHash;
|
||||
use futures::prelude::*;
|
||||
use libp2p_core::{InboundUpgrade, OutboundUpgrade, UpgradeInfo, PeerId, upgrade};
|
||||
use protobuf::{ProtobufError, Message as ProtobufMessage};
|
||||
use prost::Message;
|
||||
use std::{error, fmt, io, iter, pin::Pin};
|
||||
|
||||
/// Implementation of `ConnectionUpgrade` for the floodsub protocol.
|
||||
@ -31,7 +31,6 @@ pub struct FloodsubConfig {}
|
||||
|
||||
impl FloodsubConfig {
|
||||
/// Builds a new `FloodsubConfig`.
|
||||
#[inline]
|
||||
pub fn new() -> FloodsubConfig {
|
||||
FloodsubConfig {}
|
||||
}
|
||||
@ -41,7 +40,6 @@ impl UpgradeInfo for FloodsubConfig {
|
||||
type Info = &'static [u8];
|
||||
type InfoIter = iter::Once<Self::Info>;
|
||||
|
||||
#[inline]
|
||||
fn protocol_info(&self) -> Self::InfoIter {
|
||||
iter::once(b"/floodsub/1.0.0")
|
||||
}
|
||||
@ -58,18 +56,17 @@ where
|
||||
fn upgrade_inbound(self, mut socket: TSocket, _: Self::Info) -> Self::Future {
|
||||
Box::pin(async move {
|
||||
let packet = upgrade::read_one(&mut socket, 2048).await?;
|
||||
let mut rpc: rpc_proto::RPC = protobuf::parse_from_bytes(&packet)?;
|
||||
let rpc = rpc_proto::Rpc::decode(&packet[..])?;
|
||||
|
||||
let mut messages = Vec::with_capacity(rpc.get_publish().len());
|
||||
for mut publish in rpc.take_publish().into_iter() {
|
||||
let mut messages = Vec::with_capacity(rpc.publish.len());
|
||||
for publish in rpc.publish.into_iter() {
|
||||
messages.push(FloodsubMessage {
|
||||
source: PeerId::from_bytes(publish.take_from()).map_err(|_| {
|
||||
source: PeerId::from_bytes(publish.from.unwrap_or_default()).map_err(|_| {
|
||||
FloodsubDecodeError::InvalidPeerId
|
||||
})?,
|
||||
data: publish.take_data(),
|
||||
sequence_number: publish.take_seqno(),
|
||||
topics: publish
|
||||
.take_topicIDs()
|
||||
data: publish.data.unwrap_or_default(),
|
||||
sequence_number: publish.seqno.unwrap_or_default(),
|
||||
topics: publish.topic_ids
|
||||
.into_iter()
|
||||
.map(TopicHash::from_raw)
|
||||
.collect(),
|
||||
@ -78,16 +75,15 @@ where
|
||||
|
||||
Ok(FloodsubRpc {
|
||||
messages,
|
||||
subscriptions: rpc
|
||||
.take_subscriptions()
|
||||
subscriptions: rpc.subscriptions
|
||||
.into_iter()
|
||||
.map(|mut sub| FloodsubSubscription {
|
||||
action: if sub.get_subscribe() {
|
||||
.map(|sub| FloodsubSubscription {
|
||||
action: if Some(true) == sub.subscribe {
|
||||
FloodsubSubscriptionAction::Subscribe
|
||||
} else {
|
||||
FloodsubSubscriptionAction::Unsubscribe
|
||||
},
|
||||
topic: TopicHash::from_raw(sub.take_topicid()),
|
||||
topic: TopicHash::from_raw(sub.topic_id.unwrap_or_default()),
|
||||
})
|
||||
.collect(),
|
||||
})
|
||||
@ -101,21 +97,19 @@ pub enum FloodsubDecodeError {
|
||||
/// Error when reading the packet from the socket.
|
||||
ReadError(upgrade::ReadOneError),
|
||||
/// Error when decoding the raw buffer into a protobuf.
|
||||
ProtobufError(ProtobufError),
|
||||
ProtobufError(prost::DecodeError),
|
||||
/// Error when parsing the `PeerId` in the message.
|
||||
InvalidPeerId,
|
||||
}
|
||||
|
||||
impl From<upgrade::ReadOneError> for FloodsubDecodeError {
|
||||
#[inline]
|
||||
fn from(err: upgrade::ReadOneError) -> Self {
|
||||
FloodsubDecodeError::ReadError(err)
|
||||
}
|
||||
}
|
||||
|
||||
impl From<ProtobufError> for FloodsubDecodeError {
|
||||
#[inline]
|
||||
fn from(err: ProtobufError) -> Self {
|
||||
impl From<prost::DecodeError> for FloodsubDecodeError {
|
||||
fn from(err: prost::DecodeError) -> Self {
|
||||
FloodsubDecodeError::ProtobufError(err)
|
||||
}
|
||||
}
|
||||
@ -156,7 +150,6 @@ impl UpgradeInfo for FloodsubRpc {
|
||||
type Info = &'static [u8];
|
||||
type InfoIter = iter::Once<Self::Info>;
|
||||
|
||||
#[inline]
|
||||
fn protocol_info(&self) -> Self::InfoIter {
|
||||
iter::once(b"/floodsub/1.0.0")
|
||||
}
|
||||
@ -170,7 +163,6 @@ where
|
||||
type Error = io::Error;
|
||||
type Future = Pin<Box<dyn Future<Output = Result<Self::Output, Self::Error>> + Send>>;
|
||||
|
||||
#[inline]
|
||||
fn upgrade_outbound(self, mut socket: TSocket, _: Self::Info) -> Self::Future {
|
||||
Box::pin(async move {
|
||||
let bytes = self.into_bytes();
|
||||
@ -183,33 +175,34 @@ where
|
||||
impl FloodsubRpc {
|
||||
/// Turns this `FloodsubRpc` into a message that can be sent to a substream.
|
||||
fn into_bytes(self) -> Vec<u8> {
|
||||
let mut proto = rpc_proto::RPC::new();
|
||||
let rpc = rpc_proto::Rpc {
|
||||
publish: self.messages.into_iter()
|
||||
.map(|msg| {
|
||||
rpc_proto::Message {
|
||||
from: Some(msg.source.into_bytes()),
|
||||
data: Some(msg.data),
|
||||
seqno: Some(msg.sequence_number),
|
||||
topic_ids: msg.topics
|
||||
.into_iter()
|
||||
.map(TopicHash::into_string)
|
||||
.collect()
|
||||
}
|
||||
})
|
||||
.collect(),
|
||||
|
||||
for message in self.messages {
|
||||
let mut msg = rpc_proto::Message::new();
|
||||
msg.set_from(message.source.into_bytes());
|
||||
msg.set_data(message.data);
|
||||
msg.set_seqno(message.sequence_number);
|
||||
msg.set_topicIDs(
|
||||
message
|
||||
.topics
|
||||
.into_iter()
|
||||
.map(TopicHash::into_string)
|
||||
.collect(),
|
||||
);
|
||||
proto.mut_publish().push(msg);
|
||||
}
|
||||
subscriptions: self.subscriptions.into_iter()
|
||||
.map(|topic| {
|
||||
rpc_proto::rpc::SubOpts {
|
||||
subscribe: Some(topic.action == FloodsubSubscriptionAction::Subscribe),
|
||||
topic_id: Some(topic.topic.into_string())
|
||||
}
|
||||
})
|
||||
.collect()
|
||||
};
|
||||
|
||||
for topic in self.subscriptions {
|
||||
let mut subscription = rpc_proto::RPC_SubOpts::new();
|
||||
subscription.set_subscribe(topic.action == FloodsubSubscriptionAction::Subscribe);
|
||||
subscription.set_topicid(topic.topic.into_string());
|
||||
proto.mut_subscriptions().push(subscription);
|
||||
}
|
||||
|
||||
proto
|
||||
.write_to_bytes()
|
||||
.expect("there is no situation in which the protobuf message can be invalid")
|
||||
let mut buf = Vec::with_capacity(rpc.encoded_len());
|
||||
rpc.encode(&mut buf).expect("Vec<u8> provides capacity as needed");
|
||||
buf
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -8,7 +8,7 @@ message RPC {
|
||||
|
||||
message SubOpts {
|
||||
optional bool subscribe = 1; // subscribe or unsubcribe
|
||||
optional string topicid = 2;
|
||||
optional string topic_id = 2;
|
||||
}
|
||||
}
|
||||
|
||||
@ -16,7 +16,7 @@ message Message {
|
||||
optional bytes from = 1;
|
||||
optional bytes data = 2;
|
||||
optional bytes seqno = 3;
|
||||
repeated string topicIDs = 4;
|
||||
repeated string topic_ids = 4;
|
||||
}
|
||||
|
||||
// topicID = hash(topicDescriptor); (not the topic.name)
|
||||
@ -38,7 +38,7 @@ message TopicDescriptor {
|
||||
|
||||
message EncOpts {
|
||||
optional EncMode mode = 1;
|
||||
repeated bytes keyHashes = 2; // the hashes of the shared keys used (salted)
|
||||
repeated bytes key_hashes = 2; // the hashes of the shared keys used (salted)
|
||||
|
||||
enum EncMode {
|
||||
NONE = 0; // no encryption, anyone can read
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@ -20,7 +20,7 @@
|
||||
|
||||
use bs58;
|
||||
use crate::rpc_proto;
|
||||
use protobuf::Message;
|
||||
use prost::Message;
|
||||
|
||||
/// Represents the hash of a topic.
|
||||
///
|
||||
@ -33,12 +33,10 @@ pub struct TopicHash {
|
||||
|
||||
impl TopicHash {
|
||||
/// Builds a new `TopicHash` from the given hash.
|
||||
#[inline]
|
||||
pub fn from_raw(hash: String) -> TopicHash {
|
||||
TopicHash { hash }
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub fn into_string(self) -> String {
|
||||
self.hash
|
||||
}
|
||||
@ -53,28 +51,24 @@ pub struct Topic {
|
||||
|
||||
impl Topic {
|
||||
/// Returns the hash of the topic.
|
||||
#[inline]
|
||||
pub fn hash(&self) -> &TopicHash {
|
||||
&self.hash
|
||||
}
|
||||
}
|
||||
|
||||
impl AsRef<TopicHash> for Topic {
|
||||
#[inline]
|
||||
fn as_ref(&self) -> &TopicHash {
|
||||
&self.hash
|
||||
}
|
||||
}
|
||||
|
||||
impl From<Topic> for TopicHash {
|
||||
#[inline]
|
||||
fn from(topic: Topic) -> TopicHash {
|
||||
topic.hash
|
||||
}
|
||||
}
|
||||
|
||||
impl<'a> From<&'a Topic> for TopicHash {
|
||||
#[inline]
|
||||
fn from(topic: &'a Topic) -> TopicHash {
|
||||
topic.hash.clone()
|
||||
}
|
||||
@ -91,21 +85,22 @@ impl TopicBuilder {
|
||||
where
|
||||
S: Into<String>,
|
||||
{
|
||||
let mut builder = rpc_proto::TopicDescriptor::new();
|
||||
builder.set_name(name.into());
|
||||
|
||||
TopicBuilder { builder }
|
||||
TopicBuilder {
|
||||
builder: rpc_proto::TopicDescriptor {
|
||||
name: Some(name.into()),
|
||||
auth: None,
|
||||
enc: None
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Turns the builder into an actual `Topic`.
|
||||
pub fn build(self) -> Topic {
|
||||
let bytes = self
|
||||
.builder
|
||||
.write_to_bytes()
|
||||
.expect("protobuf message is always valid");
|
||||
let mut buf = Vec::with_capacity(self.builder.encoded_len());
|
||||
self.builder.encode(&mut buf).expect("Vec<u8> provides capacity as needed");
|
||||
// TODO: https://github.com/libp2p/rust-libp2p/issues/473
|
||||
let hash = TopicHash {
|
||||
hash: bs58::encode(&bytes).into_string(),
|
||||
hash: bs58::encode(&buf).into_string(),
|
||||
};
|
||||
Topic {
|
||||
descriptor: self.builder,
|
||||
|
||||
Reference in New Issue
Block a user