Add message compression and versioning

bindings/wasm/examples/src/private_tangle.js

@@ -29,12 +29,18 @@ async function createIdentityPrivateTangle(restURL, networkName) {
     const config = new Config();
     config.setNetwork(network);
 
+    // Uncomment the following line to disable compression.
+    // config.disableCompression();
+
     // This URL points to the REST API of the locally running hornet node.
     config.setNode(restURL || "http://127.0.0.1:14265/");
 
     // Create a client instance from the configuration to publish messages to the Tangle.
     const client = Client.fromConfig(config);
 
+    // For debugging, message compression can be disabled with the next line.
+    // client.disableCompression();
+
     // Publish the Identity to the IOTA Network, this may take a few seconds to complete Proof-of-Work.
     const receipt = await client.publishDocument(doc.toJSON());
 

bindings/wasm/src/tangle/config.rs

@@ -48,6 +48,11 @@ impl Config {
     self.try_with_mut(|builder| builder.node(url).map_err(wasm_error))
   }
 
+  #[wasm_bindgen(js_name = disableCompression)]
+  pub fn disable_compression(&mut self) -> Result<(), JsValue> {
+    self.with_mut(|builder| builder.disable_compression())
+  }
+
   #[wasm_bindgen(js_name = setPrimaryNode)]
   pub fn set_primary_node(
     &mut self,

identity-iota/Cargo.toml

@@ -12,6 +12,7 @@ description = "An IOTA Tangle intergration for the identity-rs library."
 
 [dependencies]
 async-trait = { version = "0.1", default-features = false }
+brotli = { version = "3.3", default-features = false, features = ["std"] }
 dashmap = { version = "4.0" }
 form_urlencoded = { version = "1.0" }
 futures = { version = "0.3" }

identity-iota/src/error.rs

@@ -53,4 +53,8 @@ pub enum Error {
   NoExplorerURLSet,
   #[error("Invalid Explorer Url")]
   InvalidExplorerURL,
+  #[error("compression error")]
+  CompressionError,
+  #[error("invalid message flags")]
+  InvalidMessageFlags,
 }

identity-iota/src/tangle/client.rs

@@ -17,6 +17,9 @@ use crate::did::IotaDID;
 use crate::did::IotaDocument;
 use crate::error::Error;
 use crate::error::Result;
+use crate::tangle::did_encoding;
+use crate::tangle::did_encoding::DIDMessageEncoding;
+use crate::tangle::did_message_versioning;
 use crate::tangle::ClientBuilder;
 use crate::tangle::Message;
 use crate::tangle::MessageId;
@@ -30,6 +33,7 @@ use crate::tangle::TangleResolve;
 pub struct Client {
   pub(crate) client: IotaClient,
   pub(crate) network: Network,
+  pub(crate) encoding: DIDMessageEncoding,
 }
 
 impl Client {
@@ -65,6 +69,7 @@ impl Client {
     Ok(Self {
       client: client.finish().await?,
       network: builder.network,
+      encoding: builder.encoding,
     })
   }
 
@@ -84,13 +89,25 @@ impl Client {
     self.publish_json(&IotaDocument::diff_index(message_id)?, diff).await
   }
 
-  /// Publishes arbitrary JSON data to the specified index on the Tangle.
+  /// Compresses and Publishes arbitrary JSON data to the specified index on the Tangle.
   pub async fn publish_json<T: ToJson>(&self, index: &str, data: &T) -> Result<Receipt> {
+    let encoded_message_data = match self.encoding {
+      DIDMessageEncoding::JsonBrotli => {
+        did_encoding::compress_message(&data.to_json()?, DIDMessageEncoding::JsonBrotli).map(|compressed_data| {
+          did_encoding::add_encoding_version_flag(compressed_data, DIDMessageEncoding::JsonBrotli)
+        })?
+      }
+      DIDMessageEncoding::Json => data.to_json_vec().map(|uncompressed_data| {
+        did_encoding::add_encoding_version_flag(uncompressed_data, DIDMessageEncoding::Json)
+      })?,
+    };
+    let message_data =
+      did_message_versioning::add_version_flag(encoded_message_data, did_message_versioning::CURRENT_MESSAGE_VERSION);
     self
       .client
       .message()
       .with_index(index)
-      .with_data(data.to_json_vec()?)
+      .with_data(message_data)
       .finish()
       .await
       .map_err(Into::into)

identity-iota/src/tangle/client_builder.rs

@@ -7,6 +7,7 @@ use core::fmt::Result as FmtResult;
 use std::time::Duration;
 
 use crate::error::Result;
+use crate::tangle::did_encoding::DIDMessageEncoding;
 use crate::tangle::Client;
 use crate::tangle::Network;
 
@@ -17,6 +18,7 @@ pub struct ClientBuilder {
   pub(super) nodeset: bool,
   pub(super) network: Network,
   pub(super) builder: iota_client::ClientBuilder,
+  pub(super) encoding: DIDMessageEncoding,
 }
 
 impl ClientBuilder {
@@ -26,6 +28,7 @@ impl ClientBuilder {
       nodeset: false,
       network: Default::default(),
       builder: iota_client::ClientBuilder::new().with_local_pow(DEFAULT_LOCAL_POW),
+      encoding: DIDMessageEncoding::JsonBrotli,
     }
   }
 
@@ -36,6 +39,12 @@ impl ClientBuilder {
     self
   }
 
+  /// Disables message compression.
+  pub fn disable_compression(mut self) -> Self {
+    self.encoding = DIDMessageEncoding::Json;
+    self
+  }
+
   /// Adds an IOTA node by its URL.
   pub fn node(mut self, url: &str) -> Result<Self> {
     self.builder = self.builder.with_node(url)?;

identity-iota/src/tangle/compression_brotli.rs

@@ -0,0 +1,48 @@
+// Copyright 2020-2021 IOTA Stiftung
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::error::Error;
+use crate::error::Error::CompressionError;
+use std::io::prelude::*;
+
+const BUFFER_SIZE: usize = 4096;
+const QUALITY: u32 = 5; // compression level
+const WINDOWS_SIZE: u32 = 22;
+
+pub(crate) fn compress_brotli<T: AsRef<[u8]>>(input: T) -> Result<Vec<u8>, Error> {
+  let mut buf = Vec::new();
+  let mut compressor = brotli::CompressorReader::new(input.as_ref(), BUFFER_SIZE, QUALITY, WINDOWS_SIZE);
+  compressor.read_to_end(&mut buf).map_err(|_| Error::CompressionError)?;
+  Ok(buf)
+}
+
+pub(crate) fn decompress_brotli<T: AsRef<[u8]> + ?Sized>(input: &T) -> Result<Vec<u8>, Error> {
+  let mut decompressor = brotli::Decompressor::new(input.as_ref(), BUFFER_SIZE);
+  let mut buf = Vec::new();
+  decompressor.read_to_end(&mut buf).map_err(|_| CompressionError)?;
+  Ok(buf)
+}
+
+#[cfg(test)]
+mod test {
+  use crate::did::IotaDocument;
+  use crate::tangle::compression_brotli::compress_brotli;
+  use crate::tangle::compression_brotli::decompress_brotli;
+  use identity_core::convert::ToJson;
+  use identity_core::crypto::KeyPair;
+
+  #[test]
+  fn test_brotli() {
+    let keypair: KeyPair = KeyPair::new_ed25519().unwrap();
+    let mut document: IotaDocument = IotaDocument::new(&keypair).unwrap();
+    document
+      .sign_self(keypair.private(), &document.default_signing_method().unwrap().id())
+      .unwrap();
+
+    let data = document.to_json().unwrap();
+    let compressed = compress_brotli(data.as_str()).unwrap();
+    let decompressed = decompress_brotli(&compressed).unwrap();
+
+    assert_eq!(decompressed, data.as_bytes());
+  }
+}

identity-iota/src/tangle/did_encoding.rs

@@ -0,0 +1,53 @@
+// Copyright 2020-2021 IOTA Stiftung
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::error::Error::CompressionError;
+use crate::tangle::compression_brotli::compress_brotli;
+use crate::tangle::compression_brotli::decompress_brotli;
+use crate::Error;
+
+#[derive(Debug, Copy, Clone)]
+pub(crate) enum DIDMessageEncoding {
+  Json = 0,
+  JsonBrotli = 1,
+}
+
+/// Adds the current encoding flag at the beginning of arbitrary data.
+pub(crate) fn add_encoding_version_flag(mut data: Vec<u8>, encoding: DIDMessageEncoding) -> Vec<u8> {
+  let encoding_version = encoding as u8;
+
+  data.splice(0..0, [encoding_version].iter().cloned());
+  data
+}
+
+/// Decompresses a message depending on the encoding flag.
+pub(crate) fn decompress_message(encoding_flag: &u8, data: &[u8]) -> Result<Vec<u8>, Error> {
+  if *encoding_flag == DIDMessageEncoding::JsonBrotli as u8 {
+    decompress_brotli(data)
+  } else if *encoding_flag == DIDMessageEncoding::Json as u8 {
+    Ok(data.to_vec()) //todo prevent copying the slice.
+  } else {
+    Err(Error::InvalidMessageFlags)
+  }
+}
+
+/// Compresses a message depending on current encoding version.
+pub(crate) fn compress_message<T: AsRef<[u8]>>(message: T, encoding: DIDMessageEncoding) -> Result<Vec<u8>, Error> {
+  match encoding {
+    DIDMessageEncoding::JsonBrotli => compress_brotli(message),
+    DIDMessageEncoding::Json => Err(CompressionError),
+  }
+}
+
+#[cfg(test)]
+mod test {
+  use crate::tangle::did_encoding::add_encoding_version_flag;
+  use crate::tangle::did_encoding::DIDMessageEncoding;
+
+  #[test]
+  fn test_add_version_flag() {
+    let message: Vec<u8> = vec![10, 4, 5, 5];
+    let message_with_flag = add_encoding_version_flag(message, DIDMessageEncoding::JsonBrotli);
+    assert_eq!(message_with_flag, [DIDMessageEncoding::JsonBrotli as u8, 10, 4, 5, 5])
+  }
+}

identity-iota/src/tangle/did_message_versioning.rs

@@ -0,0 +1,40 @@
+// Copyright 2020-2021 IOTA Stiftung
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::Error;
+
+#[derive(Copy, Clone)]
+pub(crate) enum DIDMessageVersion {
+  V1 = 1,
+}
+
+pub(crate) const CURRENT_MESSAGE_VERSION: DIDMessageVersion = DIDMessageVersion::V1;
+
+/// Adds the current message version flag at the beginning of arbitrary data.
+pub(crate) fn add_version_flag(mut data: Vec<u8>, message_version: DIDMessageVersion) -> Vec<u8> {
+  let version_flag = message_version as u8;
+  data.splice(0..0, [version_flag].iter().cloned());
+  data
+}
+
+/// Checks if flag matches message_version.
+pub(crate) fn check_version_flag(flag: &u8, message_version: DIDMessageVersion) -> Result<(), Error> {
+  if message_version as u8 == *flag {
+    Ok(())
+  } else {
+    Err(Error::InvalidMessageFlags)
+  }
+}
+
+#[cfg(test)]
+mod test {
+  use crate::tangle::did_message_versioning::add_version_flag;
+  use crate::tangle::did_message_versioning::CURRENT_MESSAGE_VERSION;
+
+  #[test]
+  fn test_add_version_flag() {
+    let message: Vec<u8> = vec![10, 4, 5, 5];
+    let message_with_flag = add_version_flag(message, CURRENT_MESSAGE_VERSION);
+    assert_eq!(message_with_flag, [CURRENT_MESSAGE_VERSION as u8, 10, 4, 5, 5])
+  }
+}

identity-iota/src/tangle/message_ext.rs

@@ -14,6 +14,9 @@ use crate::did::DocumentDiff;
 use crate::did::IotaDID;
 use crate::did::IotaDocument;
 use crate::error::Result;
+use crate::tangle::did_encoding;
+use crate::tangle::did_message_versioning;
+use crate::tangle::did_message_versioning::DIDMessageVersion;
 use crate::tangle::TangleRef;
 
 // TODO: Use MessageId when it has a const ctor
@@ -45,11 +48,20 @@ fn parse_payload<T: FromJson + TangleRef>(message_id: MessageId, payload: Option
 }
 
 fn parse_data<T: FromJson + TangleRef>(message_id: MessageId, data: &[u8]) -> Option<T> {
-  let mut resource: T = T::from_json_slice(data).ok()?;
-
-  resource.set_message_id(message_id);
+  let version_check = did_message_versioning::check_version_flag(&data[0], DIDMessageVersion::V1);
+  if version_check.is_err() {
+    return None;
+  }
 
-  Some(resource)
+  let compression_result = did_encoding::decompress_message(&data[1], &data[2..]);
+  match compression_result {
+    Ok(decompressed_message) => {
+      let mut resource: T = T::from_json_slice(decompressed_message.as_slice()).ok()?;
+      resource.set_message_id(message_id);
+      Some(resource)
+    }
+    _ => None,
+  }
 }
 
 pub trait MessageIdExt: Sized {

identity-iota/src/tangle/mod.rs

@@ -25,6 +25,9 @@ pub use self::traits::TangleResolve;
 mod client;
 mod client_builder;
 mod client_map;
+mod compression_brotli;
+mod did_encoding;
+mod did_message_versioning;
 mod message_ext;
 mod message_index;
 mod network;