messages.rs 5.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229
  1. use futures::prelude::*;
  2. use log::*;
  3. use std::io;
  4. use std::net::SocketAddr;
  5. use crate::error::{Error, Result};
  6. use crate::serial::{Decodable, Encodable, VarInt};
  7. const MAGIC_BYTES: [u8; 4] = [0xd9, 0xef, 0xb6, 0x7d];
  8. /// Generic message template.
  9. pub trait Message: 'static + Encodable + Decodable + Send + Sync {
  10. fn name() -> &'static str;
  11. }
  12. /// Outbound keep-alive message.
  13. pub struct PingMessage {
  14. pub nonce: u32,
  15. }
  16. /// Inbound keep-alive message.
  17. pub struct PongMessage {
  18. pub nonce: u32,
  19. }
  20. /// Requests address of outbound connection.
  21. pub struct GetAddrsMessage {}
  22. /// Sends address information to inbound connection. Response to GetAddrs
  23. /// message.
  24. pub struct AddrsMessage {
  25. pub addrs: Vec<SocketAddr>,
  26. }
  27. /// Requests version information of outbound connection.
  28. pub struct VersionMessage {}
  29. /// Sends version information to inbound connection. Response to VersionMessage.
  30. pub struct VerackMessage {}
  31. impl Message for PingMessage {
  32. fn name() -> &'static str {
  33. "ping"
  34. }
  35. }
  36. impl Message for PongMessage {
  37. fn name() -> &'static str {
  38. "pong"
  39. }
  40. }
  41. impl Message for GetAddrsMessage {
  42. fn name() -> &'static str {
  43. "getaddr"
  44. }
  45. }
  46. impl Message for AddrsMessage {
  47. fn name() -> &'static str {
  48. "addr"
  49. }
  50. }
  51. impl Message for VersionMessage {
  52. fn name() -> &'static str {
  53. "version"
  54. }
  55. }
  56. impl Message for VerackMessage {
  57. fn name() -> &'static str {
  58. "verack"
  59. }
  60. }
  61. impl Encodable for PingMessage {
  62. fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
  63. let mut len = 0;
  64. len += self.nonce.encode(&mut s)?;
  65. Ok(len)
  66. }
  67. }
  68. impl Decodable for PingMessage {
  69. fn decode<D: io::Read>(mut d: D) -> Result<Self> {
  70. Ok(Self {
  71. nonce: Decodable::decode(&mut d)?,
  72. })
  73. }
  74. }
  75. impl Encodable for PongMessage {
  76. fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
  77. let mut len = 0;
  78. len += self.nonce.encode(&mut s)?;
  79. Ok(len)
  80. }
  81. }
  82. impl Decodable for PongMessage {
  83. fn decode<D: io::Read>(mut d: D) -> Result<Self> {
  84. Ok(Self {
  85. nonce: Decodable::decode(&mut d)?,
  86. })
  87. }
  88. }
  89. impl Encodable for GetAddrsMessage {
  90. fn encode<S: io::Write>(&self, mut _s: S) -> Result<usize> {
  91. let len = 0;
  92. Ok(len)
  93. }
  94. }
  95. impl Decodable for GetAddrsMessage {
  96. fn decode<D: io::Read>(mut _d: D) -> Result<Self> {
  97. Ok(Self {})
  98. }
  99. }
  100. impl Encodable for AddrsMessage {
  101. fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
  102. let mut len = 0;
  103. len += self.addrs.encode(&mut s)?;
  104. Ok(len)
  105. }
  106. }
  107. impl Decodable for AddrsMessage {
  108. fn decode<D: io::Read>(mut d: D) -> Result<Self> {
  109. Ok(Self {
  110. addrs: Decodable::decode(&mut d)?,
  111. })
  112. }
  113. }
  114. impl Encodable for VersionMessage {
  115. fn encode<S: io::Write>(&self, _s: S) -> Result<usize> {
  116. Ok(0)
  117. }
  118. }
  119. impl Decodable for VersionMessage {
  120. fn decode<D: io::Read>(_d: D) -> Result<Self> {
  121. Ok(Self {})
  122. }
  123. }
  124. impl Encodable for VerackMessage {
  125. fn encode<S: io::Write>(&self, _s: S) -> Result<usize> {
  126. Ok(0)
  127. }
  128. }
  129. impl Decodable for VerackMessage {
  130. fn decode<D: io::Read>(_d: D) -> Result<Self> {
  131. Ok(Self {})
  132. }
  133. }
  134. /// Packets are the base type read from the network. Converted to messages and
  135. /// passed to event loop.
  136. pub struct Packet {
  137. pub command: String,
  138. pub payload: Vec<u8>,
  139. }
  140. /// Reads and decodes an inbound payload.
  141. pub async fn read_packet<R: AsyncRead + Unpin>(stream: &mut R) -> Result<Packet> {
  142. // Packets have a 4 byte header of magic digits
  143. // This is used for network debugging
  144. let mut magic = [0u8; 4];
  145. debug!(target: "net", "reading magic...");
  146. stream.read_exact(&mut magic).await?;
  147. debug!(target: "net", "read magic {:?}", magic);
  148. if magic != MAGIC_BYTES {
  149. return Err(Error::MalformedPacket);
  150. }
  151. // The type of the message
  152. let command_len = VarInt::decode_async(stream).await?.0 as usize;
  153. let mut cmd = vec![0u8; command_len];
  154. if command_len > 0 {
  155. stream.read_exact(&mut cmd).await?;
  156. }
  157. let cmd = String::from_utf8(cmd)?;
  158. debug!(target: "net", "read command: {}", cmd);
  159. let payload_len = VarInt::decode_async(stream).await?.0 as usize;
  160. // The message-dependent data (see message types)
  161. let mut payload = vec![0u8; payload_len];
  162. if payload_len > 0 {
  163. stream.read_exact(&mut payload).await?;
  164. }
  165. debug!(target: "net", "read payload {} bytes", payload_len);
  166. Ok(Packet {
  167. command: cmd,
  168. payload,
  169. })
  170. }
  171. /// Sends an outbound packet by writing data to TCP stream.
  172. pub async fn send_packet<W: AsyncWrite + Unpin>(stream: &mut W, packet: Packet) -> Result<()> {
  173. debug!(target: "net", "sending magic...");
  174. stream.write_all(&MAGIC_BYTES).await?;
  175. debug!(target: "net", "sent magic...");
  176. VarInt(packet.command.len() as u64)
  177. .encode_async(stream)
  178. .await?;
  179. assert!(!packet.command.is_empty());
  180. stream.write_all(&packet.command.as_bytes()).await?;
  181. debug!(target: "net", "sent command: {}", packet.command);
  182. assert_eq!(std::mem::size_of::<usize>(), std::mem::size_of::<u64>());
  183. VarInt(packet.payload.len() as u64)
  184. .encode_async(stream)
  185. .await?;
  186. if packet.payload.len() > 0 {
  187. stream.write_all(&packet.payload).await?;
  188. }
  189. debug!(target: "net", "sent payload {} bytes", packet.payload.len() as u64);
  190. Ok(())
  191. }