Merge pull request iotaledger#580 from iotaledger/bindings/get-messag…

…e-id

Bindings/get message id

.changes/getMessageId.md

@@ -0,0 +1,5 @@
+---
+"nodejs-binding": minor
+---
+
+Added getMessageId function.

bindings/nodejs/README.md

@@ -323,6 +323,16 @@ Checks if a given address is valid.
 
 **Returns** A boolean.
 
+#### getMessageId(message: string): boolean
+
+Returns the message id from a message.
+
+| Param   | Type                | Description    |
+| ------- | ------------------- | -------------- |
+| message | <code>string</code> | The message id |
+
+**Returns** the message id.
+
 #### retry(messageId: string): Promise<Message>
 
 Retries (promotes or reattaches) the message associated with the given id.

bindings/nodejs/examples/10_mqtt.js

@@ -10,6 +10,7 @@ async function run() {
 
     client.subscriber().topics(['milestones/confirmed', 'messages']).subscribe((err, data) => {
         console.log(data);
+        // To get the message id `client.getMessageId(data.payload)` can be used
     })
 }
 

bindings/nodejs/lib/index.d.ts

@@ -116,6 +116,7 @@ export declare class Client {
   bech32ToHex(address: string): string
   hexToBech32(address: string, bech32_hrp?: string): Promise<string>
   isAddressValid(address: string): boolean
+  getMessageId(message: string): string
   getMilestone(index: number): Promise<MilestoneMetadata>
   getMilestoneUtxoChanges(index: number): Promise<MilestoneUTXOChanges>
   getReceipts(): Promise<Receipts[]>

bindings/nodejs/native/src/classes/client/mod.rs

@@ -3,13 +3,14 @@
 
 #![allow(clippy::unnecessary_wraps)]
 use iota_client::{
-    bee_message::prelude::{Address, MessageId, TransactionId, UtxoInput},
+    bee_message::prelude::{Address, Message, MessageId, TransactionId, UtxoInput},
+    bee_rest_api::types::dtos::MessageDto as BeeMessageDto,
     AddressOutputsOptions, Client, OutputType, Seed,
 };
 use neon::prelude::*;
 use serde::Deserialize;
 
-use std::str::FromStr;
+use std::{convert::TryFrom, str::FromStr};
 
 mod builder;
 pub use builder::*;
@@ -665,5 +666,16 @@ declare_types! {
             let is_valid = Client::is_address_valid(address.as_str());
             Ok(cx.boolean(is_valid).upcast())
         }
+
+        method getMessageId(mut cx) {
+            let message_string = cx.argument::<JsString>(0)?.value();
+            // Try BeeMessageDto and if it fails Message
+            let message = match serde_json::from_str::<BeeMessageDto>(&message_string){
+                Ok(message_dto) => Message::try_from(&message_dto).expect("invalid message"),
+                Err(_) => serde_json::from_str::<Message>(&message_string).expect("invalid message"),
+            };
+            let message_id = message.id().0.to_string();
+            Ok(cx.string(message_id).upcast())
+        }
     }
 }

bindings/python/examples/10_mqtt.py

@@ -0,0 +1,62 @@
+import iota_client
+import json
+import os
+import queue
+import time
+
+# The node mqtt url
+node_url = 'https://chrysalis-nodes.iota.org/'
+
+# The queue to store received events
+q = queue.Queue()
+
+# The MQTT broker options
+broker_options = {
+    'automatic_disconnect': True,
+    'timeout': 30,
+    'use_ws': True,
+    'port': 443,
+    'max_reconnection_attempts': 5,
+}
+
+client = iota_client.Client(
+    nodes_name_password=[[node_url]], mqtt_broker_options=broker_options)
+
+# The queue to store received events
+q = queue.Queue()
+
+# The MQTT broker options
+broker_options = {
+    'automatic_disconnect': True,
+    'timeout': 5,
+    'use_ws': True,
+    'port': 443,
+    'max_reconnection_attempts': 5,
+}
+
+
+def worker(topics):
+    """The worker to process the queued events.
+    """
+    received_events = 0
+    while received_events < 10:
+        item = q.get(True)
+        event = json.loads(item)
+        print(f'Received Event: {event}')
+        message_id = client.get_message_id(str(event['payload']))
+        print(f'Received message_id: {message_id}')
+        received_events += 1
+        q.task_done()
+
+
+def on_mqtt_event(event):
+    """Put the received event to queue.
+    """
+    q.put(event)
+
+
+if __name__ == '__main__':
+    client.subscribe_topics(['messages'], on_mqtt_event)
+    worker(['messages'])
+    client.disconnect()
+    q.queue.clear()

bindings/python/native/src/client/high_level_api.rs

@@ -7,13 +7,15 @@ use crate::client::{
 };
 use iota_client::{
     bee_message::prelude::{
-        MessageId as RustMessageId, TransactionId as RustTransactionId, UtxoInput as RustUtxoInput,
+        Message as RustMessage, MessageId as RustMessageId, TransactionId as RustTransactionId,
+        UtxoInput as RustUtxoInput,
     },
+    bee_rest_api::types::dtos::MessageDto as BeeMessageDto,
     Client as RustClient, Seed as RustSeed,
 };
 use pyo3::{exceptions, prelude::*};
 use std::{
-    convert::{Into, TryInto},
+    convert::{Into, TryFrom, TryInto},
     str::FromStr,
 };
 
@@ -163,6 +165,22 @@ impl Client {
         })?;
         Ok(children.iter().map(|child| hex::encode(child.as_ref())).collect())
     }
+    /// Get the message id from the payload string.
+    ///
+    /// Args:
+    ///     payload_str (str): The identifier of message.
+    ///
+    /// Returns:
+    ///     message_id (str): The identifier of message.
+    fn get_message_id(&self, payload_str: &str) -> Result<String> {
+        // Try BeeMessageDto and if it fails Message
+        let message = match serde_json::from_str::<BeeMessageDto>(&payload_str) {
+            Ok(message_dto) => RustMessage::try_from(&message_dto).expect("invalid message"),
+            Err(_) => serde_json::from_str::<RustMessage>(&payload_str).expect("invalid message"),
+        };
+        let message_id = message.id().0.to_string();
+        Ok(message_id)
+    }
     /// Get the list of message indices from the message_id.
     ///
     /// Args: