Implement basic upward & downward messages (#118)

* Start by replacing branch names and set `DownwardMessage`

* Add the upward-message crate

* Add Kusama & Polkadot

* More work on getting the upward messages working

* Fix build

* Begin to integrate it into the test Parachain

* Update

* Make everything compile again

* Switch to westend and print parachain account on startup

* Use MultiSignature etc

* Fix validate block

* Some downward messages work

* Update git reference

* More downward messages integration

* Update test runtime for downward messages

* Enable downward message handler and withdraw send tokens

* Add some docs

* Begin to implement simple XCMP

* More work

* Fixes and make parachain id configurable

* Make parachain ID be part of the genesis

* Finishing the XCMP message demo

* Update and fixes tests

* Update branch

cumulus/message-broker/src/lib.rs

@@ -16,52 +16,128 @@
 
 //! Cumulus message broker pallet.
 //!
-//! This pallet provides support for handling downward and upward messages.
+//! This pallet provides support for handling downward, upward messages and
+//! XMCP messages.
 
 #![cfg_attr(not(feature = "std"), no_std)]
 
+use codec::{Encode, Decode};
 use cumulus_primitives::{
 	inherents::{DownwardMessagesType, DOWNWARD_MESSAGES_IDENTIFIER},
-	well_known_keys, DownwardMessageHandler, UpwardMessageSender,
+	well_known_keys,
+	xcmp::{RawXCMPMessage, XCMPMessageHandler, XCMPMessageSender},
+	DownwardMessage, DownwardMessageHandler, GenericUpwardMessage, ParaId, UpwardMessageOrigin,
+	UpwardMessageSender,
 };
+use cumulus_upward_message::XCMPMessage;
 use frame_support::{
-	decl_module, storage,
+	decl_event, decl_module, storage,
+	traits::Get,
 	weights::{DispatchClass, Weight},
 };
 use frame_system::ensure_none;
 use sp_inherents::{InherentData, InherentIdentifier, MakeFatalError, ProvideInherent};
+use sp_runtime::traits::Hash;
+use sp_std::vec::Vec;
 
 /// Configuration trait of this pallet.
 pub trait Trait: frame_system::Trait {
+	/// Event type used by the runtime.
+	type Event: From<Event<Self>> + Into<<Self as frame_system::Trait>::Event>;
+
 	/// The downward message handlers that will be informed when a message is received.
 	type DownwardMessageHandlers: DownwardMessageHandler;
+
+	/// The upward message type used by the Parachain runtime.
+	type UpwardMessage: codec::Codec + XCMPMessage;
+
+	/// The XCMP message handlers that will be informed when a XCMP message is received.
+	type XCMPMessageHandlers: XCMPMessageHandler<Self::XCMPMessage>;
+
+	/// The XCMP message type used by the Parachain runtime.
+	type XCMPMessage: codec::Codec;
+
+	/// The Id of the parachain.
+	type ParachainId: Get<ParaId>;
+}
+
+decl_event! {
+	pub enum Event<T> where Hash = <T as frame_system::Trait>::Hash {
+		/// An upward message was sent to the relay chain.
+		///
+		/// The hash corresponds to the hash of the encoded upward message.
+		UpwardMessageSent(Hash),
+	}
 }
 
 decl_module! {
-	pub struct Module<T: Trait> for enum Call where origin: T::Origin {
+	pub struct Module<T: Trait> for enum Call where origin: T::Origin, system = frame_system {
 		/// Executes the given downward messages by calling the message handlers.
 		///
 		/// The origin of this call needs to be `None` as this is an inherent.
 		#[weight = (10, DispatchClass::Mandatory)]
-		fn execute_downward_messages(origin, messages: Vec<()>) {
+		fn execute_downward_messages(origin, messages: Vec<DownwardMessage>) {
 			ensure_none(origin)?;
-			messages.iter().for_each(T::DownwardMessageHandlers::handle_downward_message);
+
+			//TODO: max messages should not be hardcoded. It should be determined based on the
+			// weight used by the handlers.
+			let max_messages = 10;
+			messages.iter().take(max_messages).for_each(|msg| {
+				match msg {
+					DownwardMessage::XCMPMessage(msg) => {
+						if let Ok(msg) = RawXCMPMessage::decode(&mut &msg[..]) {
+							if let Ok(xcmp_msg) = T::XCMPMessage::decode(&mut &msg.data[..]) {
+								T::XCMPMessageHandlers::handle_xcmp_message(msg.from, &xcmp_msg);
+							}
+						}
+					},
+					msg => T::DownwardMessageHandlers::handle_downward_message(msg),
+				}
+			});
+
+			let processed = sp_std::cmp::min(messages.len(), max_messages) as u32;
+			storage::unhashed::put(well_known_keys::PROCESSED_DOWNWARD_MESSAGES, &processed);
 		}
 
 		fn on_initialize() -> Weight {
 			storage::unhashed::kill(well_known_keys::UPWARD_MESSAGES);
 
-			0
+			T::DbWeight::get().writes(1)
 		}
+
+		fn deposit_event() = default;
 	}
 }
 
-impl<T: Trait> UpwardMessageSender for Module<T> {
-	fn send_upward_message(_: &()) -> Result<(), ()> {
+impl<T: Trait> UpwardMessageSender<T::UpwardMessage> for Module<T> {
+	fn send_upward_message(msg: &T::UpwardMessage, origin: UpwardMessageOrigin) -> Result<(), ()> {
+		//TODO: check fee schedule
+		let data = msg.encode();
+		let data_hash = T::Hashing::hash(&data);
+
+		let msg = GenericUpwardMessage { origin, data };
+		sp_io::storage::append(well_known_keys::UPWARD_MESSAGES, msg.encode());
+
+		Self::deposit_event(RawEvent::UpwardMessageSent(data_hash));
+
 		Ok(())
 	}
 }
 
+impl<T: Trait> XCMPMessageSender<T::XCMPMessage> for Module<T> {
+	fn send_xcmp_message(dest: ParaId, msg: &T::XCMPMessage) -> Result<(), ()> {
+		let message = RawXCMPMessage {
+			from: T::ParachainId::get(),
+			data: msg.encode(),
+		};
+
+		Self::send_upward_message(
+			&T::UpwardMessage::send_message(dest, message.encode()),
+			UpwardMessageOrigin::Parachain,
+		)
+	}
+}
+
 impl<T: Trait> ProvideInherent for Module<T> {
 	type Call = Call<T>;
 	type Error = MakeFatalError<()>;