|
@@ -17,8 +17,8 @@ use crate::{
|
|
|
error::{Error, Result},
|
|
error::{Error, Result},
|
|
|
net::{
|
|
net::{
|
|
|
message_subscriber::{MessageSubscription, MessageSubsystem},
|
|
message_subscriber::{MessageSubscription, MessageSubsystem},
|
|
|
- messages,
|
|
|
|
|
- protocols::{ProtocolBase, ProtocolBasePtr},
|
|
|
|
|
|
|
+ message,
|
|
|
|
|
+ protocol::{ProtocolBase, ProtocolBasePtr},
|
|
|
},
|
|
},
|
|
|
system::{StoppableTask, StoppableTaskPtr, Subscriber, SubscriberPtr, Subscription},
|
|
system::{StoppableTask, StoppableTaskPtr, Subscriber, SubscriberPtr, Subscription},
|
|
|
};
|
|
};
|
|
@@ -108,7 +108,7 @@ impl Channel {
|
|
|
/// Sends a message across a channel. Calls function 'send_message' that
|
|
/// Sends a message across a channel. Calls function 'send_message' that
|
|
|
/// creates a new payload and sends it over the TCP connection as a
|
|
/// creates a new payload and sends it over the TCP connection as a
|
|
|
/// packet. Returns an error if something goes wrong.
|
|
/// packet. Returns an error if something goes wrong.
|
|
|
- pub async fn send<M: messages::Message>(&self, message: M) -> Result<()> {
|
|
|
|
|
|
|
+ pub async fn send<M: message::Message>(&self, message: M) -> Result<()> {
|
|
|
debug!(target: "net",
|
|
debug!(target: "net",
|
|
|
"Channel::send() [START, command={:?}, address={}]",
|
|
"Channel::send() [START, command={:?}, address={}]",
|
|
|
M::name(),
|
|
M::name(),
|
|
@@ -139,17 +139,17 @@ impl Channel {
|
|
|
/// it. Then creates a message packet- the base type of the network- and
|
|
/// it. Then creates a message packet- the base type of the network- and
|
|
|
/// copies the payload into it. Then we send the packet over the TCP
|
|
/// copies the payload into it. Then we send the packet over the TCP
|
|
|
/// stream.
|
|
/// stream.
|
|
|
- async fn send_message<M: messages::Message>(&self, message: M) -> Result<()> {
|
|
|
|
|
|
|
+ async fn send_message<M: message::Message>(&self, message: M) -> Result<()> {
|
|
|
let mut payload = Vec::new();
|
|
let mut payload = Vec::new();
|
|
|
message.encode(&mut payload)?;
|
|
message.encode(&mut payload)?;
|
|
|
- let packet = messages::Packet { command: String::from(M::name()), payload };
|
|
|
|
|
|
|
+ let packet = message::Packet { command: String::from(M::name()), payload };
|
|
|
|
|
|
|
|
let stream = &mut *self.writer.lock().await;
|
|
let stream = &mut *self.writer.lock().await;
|
|
|
- messages::send_packet(stream, packet).await
|
|
|
|
|
|
|
+ message::send_packet(stream, packet).await
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
/// Subscribe to a messages on the message subsystem.
|
|
/// Subscribe to a messages on the message subsystem.
|
|
|
- pub async fn subscribe_msg<M: messages::Message>(&self) -> Result<MessageSubscription<M>> {
|
|
|
|
|
|
|
+ pub async fn subscribe_msg<M: message::Message>(&self) -> Result<MessageSubscription<M>> {
|
|
|
debug!(target: "net",
|
|
debug!(target: "net",
|
|
|
"Channel::subscribe_msg() [START, command={:?}, address={}]",
|
|
"Channel::subscribe_msg() [START, command={:?}, address={}]",
|
|
|
M::name(),
|
|
M::name(),
|
|
@@ -179,12 +179,12 @@ impl Channel {
|
|
|
|
|
|
|
|
/// Perform network handshake for message subsystem dispatchers.
|
|
/// Perform network handshake for message subsystem dispatchers.
|
|
|
async fn setup_dispatchers(message_subsystem: &MessageSubsystem) {
|
|
async fn setup_dispatchers(message_subsystem: &MessageSubsystem) {
|
|
|
- message_subsystem.add_dispatch::<messages::VersionMessage>().await;
|
|
|
|
|
- message_subsystem.add_dispatch::<messages::VerackMessage>().await;
|
|
|
|
|
- message_subsystem.add_dispatch::<messages::PingMessage>().await;
|
|
|
|
|
- message_subsystem.add_dispatch::<messages::PongMessage>().await;
|
|
|
|
|
- message_subsystem.add_dispatch::<messages::GetAddrsMessage>().await;
|
|
|
|
|
- message_subsystem.add_dispatch::<messages::AddrsMessage>().await;
|
|
|
|
|
|
|
+ message_subsystem.add_dispatch::<message::VersionMessage>().await;
|
|
|
|
|
+ message_subsystem.add_dispatch::<message::VerackMessage>().await;
|
|
|
|
|
+ message_subsystem.add_dispatch::<message::PingMessage>().await;
|
|
|
|
|
+ message_subsystem.add_dispatch::<message::PongMessage>().await;
|
|
|
|
|
+ message_subsystem.add_dispatch::<message::GetAddrsMessage>().await;
|
|
|
|
|
+ message_subsystem.add_dispatch::<message::AddrsMessage>().await;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
/// Convenience function that returns the Message Subsystem.
|
|
/// Convenience function that returns the Message Subsystem.
|
|
@@ -203,7 +203,7 @@ impl Channel {
|
|
|
let reader = &mut *self.reader.lock().await;
|
|
let reader = &mut *self.reader.lock().await;
|
|
|
|
|
|
|
|
loop {
|
|
loop {
|
|
|
- let packet = match messages::read_packet(reader).await {
|
|
|
|
|
|
|
+ let packet = match message::read_packet(reader).await {
|
|
|
Ok(packet) => packet,
|
|
Ok(packet) => packet,
|
|
|
Err(err) => {
|
|
Err(err) => {
|
|
|
if Self::is_eof_error(err.clone()) {
|
|
if Self::is_eof_error(err.clone()) {
|