Bump RabbitMQ adaptor version (#40)

* Bump RabbitMQ adaptor version * Use adaptor 0.3.3 version * Fix serialization format
chronicled · Sep 27, 2019 · f8c5cf7 · f8c5cf7
1 parent a5924f3
commit f8c5cf7
Show file tree

Hide file tree

Showing 6 changed files with 167 additions and 93 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/parity/cli/mod.rs b/parity/cli/mod.rs
@@ -557,13 +557,9 @@ usage! {
 			"Specify custom API set available via JSON-RPC over IPC using a comma-delimited list of API names. Possible names are: all, safe, web3, net, eth, pubsub, personal, signer, parity, parity_pubsub, parity_accounts, parity_set, traces, rpc, secretstore, shh, shh_pubsub. You can also disable a specific API by putting '-' in the front, example: all,-personal. 'safe' enables the following APIs: web3, net, eth, pubsub, parity, parity_pubsub, traces, rpc, shh, shh_pubsub",
 
 		["API and Console Options – RabbitMQ"]
-			ARG arg_rabbitmq_hostname: (String) = "localhost", or |c: &Config| c.rabbitmq.as_ref()?.hostname.clone(),
-			"--rabbitmq-hostname=[IP]",
-			"Specify the RabbitMQ server hostname",
-
-			ARG arg_rabbitmq_port: (u16) = 5672u16, or |c: &Config| c.rabbitmq.as_ref()?.port.clone(),
-			"--rabbitmq-port=[PORT]",
-			"Specify the RabbitMQ server port",
+			ARG arg_rabbitmq_uri: (String) = "amqp://localhost:5672", or |c: &Config| c.rabbitmq.as_ref()?.uri.clone(),
+			"--rabbitmq-uri=[URI]",
+			"Specify the RabbitMQ server uri",
 
 		["API and Console Options – IPFS"]
 			FLAG flag_ipfs_api: (bool) = false, or |c: &Config| c.ipfs.as_ref()?.enable.clone(),
@@ -1281,8 +1277,7 @@ struct Ipc {
 #[derive(Default, Debug, PartialEq, Deserialize)]
 #[serde(deny_unknown_fields)]
 struct RabbitMQ {
-	hostname: Option<String>,
-	port: Option<u16>,
+	uri: Option<String>,
 }
 
 #[derive(Default, Debug, PartialEq, Deserialize)]
@@ -1750,8 +1745,7 @@ mod tests {
 			arg_ipc_apis: "web3,eth,net,parity,parity_accounts,personal,traces,rpc,secretstore".into(),
 
 			// RabbitMQ
-			arg_rabbitmq_hostname: "localhost".into(),
-			arg_rabbitmq_port: 5672u16,
+			arg_rabbitmq_uri: "amqp://localhost:5672".into(),
 
 			// DAPPS
 			arg_dapps_path: Some("$HOME/.parity/dapps".into()),
@@ -2025,8 +2019,7 @@ mod tests {
 				apis: Some(vec!["rpc".into(), "eth".into()]),
 			}),
 			rabbitmq: Some(RabbitMQ {
-				hostname: Some("localhost".into()),
-				port: Some(5672),
+				uri: Some("amqp://localhost:5672".into()),
 			}),
 			dapps: Some(Dapps {
 				_legacy_disable: None,

diff --git a/parity/cli/tests/config.toml b/parity/cli/tests/config.toml
@@ -36,8 +36,7 @@ port = 8180
 apis = ["rpc", "eth"]
 
 [rabbitmq]
-hostname = "localhost"
-port = 5672
+uri = "amqp://localhost:5672"
 
 [dapps]
 port = 8080

diff --git a/parity/configuration.rs b/parity/configuration.rs
@@ -756,8 +756,7 @@ impl Configuration {
 
 	fn rabbitmq_config(&self) -> RabbitMqConfig {
 		RabbitMqConfig {
-			hostname: self.args.arg_rabbitmq_hostname.clone(),
-			port: self.args.arg_rabbitmq_port
+			uri: self.args.arg_rabbitmq_uri.clone()
 		}
 	}
 
@@ -1423,8 +1422,7 @@ mod tests {
 			ipc_conf: Default::default(),
 			net_conf: default_network_config(),
 			rabbitmq_conf: RabbitMqConfig {
-				hostname: "localhost".into(),
-				port: 5672,
+				uri: "amqp://localhost:5672".into(),
 			},
 			network_id: None,
 			warp_sync: true,

diff --git a/rabbitmq/Cargo.toml b/rabbitmq/Cargo.toml
@@ -22,7 +22,7 @@ common-types = { path = "../ethcore/types" }
 ethereum-types = "0.4"
 futures-executor = "0.2.1"
 parity-runtime = { path = "../util/runtime" }
-rabbitmq_adaptor = {version = "0.3.1", registry = "chronicled-platform-v2"}
+rabbitmq_adaptor = {version = "0.3.3", registry = "chronicled-platform-v2"}
 rlp = { version = "0.3.0", features = ["ethereum"] }
 
 [dev-dependencies]

diff --git a/rabbitmq/src/client.rs b/rabbitmq/src/client.rs
@@ -9,7 +9,7 @@ use failure::{format_err, Error};
 use futures::future::{err, lazy};
 use handler::{Handler, Sender};
 use parity_runtime::Executor;
-use rabbitmq_adaptor::{ConsumerResult, DeliveryExt, RabbitConnection, RabbitExt};
+use rabbitmq_adaptor::{ConfigUri, ConsumerResult, DeliveryExt, RabbitConnection, RabbitExt};
 use serde::Deserialize;
 use serde_json;
 use std::sync::Arc;
@@ -29,8 +29,7 @@ use TX_ERROR_ROUTING_KEY;
 
 #[derive(Debug, Default, Clone, PartialEq)]
 pub struct RabbitMqConfig {
-	pub hostname: String,
-	pub port: u16,
+	pub uri: String,
 }
 
 /// Eth PubSub implementation.
@@ -67,9 +66,10 @@ impl<C: 'static + miner::BlockChainClient + BlockChainClient> PubSubClient<C> {
 	) -> Result<Self, Error> {
 		let (sender, receiver) = channel::<Vec<u8>>(DEFAULT_CHANNEL_SIZE);
 		let sender_handler = Box::new(Sender::new(client.clone(), miner.clone()));
+		let config_uri = ConfigUri::Uri(config.uri);
 
 		executor.spawn(lazy(move || {
-			let rabbit = RabbitConnection::new(&config.hostname, config.port, DEFAULT_REPLY_QUEUE);
+			let rabbit = RabbitConnection::new(config_uri, None, DEFAULT_REPLY_QUEUE);
 			// Consume to public transaction messages
 			try_spawn(
 				rabbit