488949865

Build Type

github

Committed by GitHub

Commit Message

Merge 791f6dd49 into 2763b9db4

Pull Request Pull Request #220: refactor(error): remove unused variants, move Error to its own mod, fixes #151

Run Details

4092 of 20746 branches covered (19.72%)

Branch coverage included in aggregate %.

47 of 129 new or added lines in 11 files covered. (36.43%)

3 existing lines in 3 files now uncovered.

1591 of 3375 relevant lines covered (47.14%)

474.68 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.15

/src/monitor.rs

// Copyright 2020 IOTA Stiftung
// SPDX-License-Identifier: Apache-2.0

use crate::{
    account::AccountHandle,
    address::{AddressOutput, IotaAddress},
    client::ClientOptions,
    message::{Message, MessageType},
};

use iota::{message::prelude::MessageId, MessageMetadata, OutputMetadata, Topic, TopicEvent};
use serde::Deserialize;

use std::convert::TryInto;

#[derive(Deserialize)]
struct AddressOutputPayload {
    #[serde(rename = "messageId")]
    message_id: String,
    #[serde(rename = "transactionId")]
    transaction_id: String,
    #[serde(rename = "outputIndex")]
    output_index: u16,
    #[serde(rename = "isSpent")]
    is_spent: bool,
    output: AddressOutputPayloadOutput,
}

#[derive(Deserialize)]
struct AddressOutputPayloadOutput {
    amount: u64,
}

/// Unsubscribe from all topics associated with the account.
pub async fn unsubscribe(account_handle: AccountHandle) -> crate::Result<()> {
    let account = account_handle.read().await;
    let client = crate::client::get_client(account.client_options());
    let mut client = client.write().await;
    client.subscriber().unsubscribe()?;
    Ok(())
}

async fn subscribe_to_topic<C: Fn(&TopicEvent) + Send + Sync + 'static>(
    client_options: &ClientOptions,
    topic: String,
    handler: C,
) -> crate::Result<()> {
    let client = crate::client::get_client(&client_options);
    let mut client = client.write().await;
    client.subscriber().with_topic(Topic::new(topic)?).subscribe(handler)?;
    Ok(())
}

/// Monitor account addresses for balance changes.
pub async fn monitor_account_addresses_balance(account_handle: AccountHandle) -> crate::Result<()> {
    let account = account_handle.read().await;
    for address in account.addresses() {
        monitor_address_balance(account_handle.clone(), address.address()).await?;
    }
    Ok(())
}

/// Monitor address for balance changes.
pub async fn monitor_address_balance(account_handle: AccountHandle, address: &IotaAddress) -> crate::Result<()> {
    let client_options = account_handle.client_options().await;
    let client_options_ = client_options.clone();
    let address = address.clone();

    subscribe_to_topic(
        &client_options_,
        format!("addresses/{}/outputs", address.to_bech32()),
        move |topic_event| {
            log::info!("[MQTT] got {:?}", topic_event);
            let topic_event = topic_event.clone();
            let address = address.clone();
            let client_options = client_options.clone();
            let account_handle = account_handle.clone();

            crate::block_on(async {
                if let Err(e) =
                    process_output(topic_event.payload.clone(), account_handle, address, client_options).await
                {
                    log::error!("[MQTT] error processing output: {:?}", e);
                }
            });
        },
    )
    .await
}

async fn process_output(
    payload: String,
    account_handle: AccountHandle,
    address: IotaAddress,
    client_options: ClientOptions,
) -> crate::Result<()> {
    let output: AddressOutputPayload = serde_json::from_str(&payload)?;
    let metadata = OutputMetadata {
        message_id: hex::decode(output.message_id).map_err(|_| crate::Error::InvalidMessageId)?,
        transaction_id: hex::decode(output.transaction_id).map_err(|_| crate::Error::InvalidTransactionId)?,
        output_index: output.output_index,
        is_spent: output.is_spent,
        amount: output.output.amount,
        address: address.clone(),
    };
    let address_output: AddressOutput = metadata.try_into()?;

    let client_options_ = client_options.clone();
    let message_id = address_output.message_id();
    let message_id_ = *message_id;

    let message = {
        let client = crate::client::get_client(&client_options_);
        let client = client.read().await;
        client.get_message().data(&message_id_).await?
    };

    let mut account = account_handle.write().await;
    let account_id = account.id().clone();
    let message_id_ = *message_id;

    let addresses = account.addresses_mut();
    let address_to_update = addresses.iter_mut().find(|a| a.address() == &address).unwrap();
    address_to_update.handle_new_output(address_output);
    crate::event::emit_balance_change(&account_id, &address_to_update, *address_to_update.balance());

    match account.messages_mut().iter().position(|m| m.id() == &message_id_) {
        Some(message_index) => {
            account.do_mut(|account| {
                let message = &mut account.messages_mut()[message_index];
                message.set_confirmed(Some(true));
            });
        }
        None => {
            let message = Message::from_iota_message(message_id_, account.addresses(), &message, Some(true)).unwrap();
            crate::event::emit_transaction_event(
                crate::event::TransactionEventType::NewTransaction,
                account.id(),
                &message,
            );
            account.do_mut(|account| {
                account.messages_mut().push(message);
            });
        }
    }
    Ok(())
}

/// Monitor the account's unconfirmed messages for confirmation state change.
pub async fn monitor_unconfirmed_messages(account_handle: AccountHandle) -> crate::Result<()> {
    let account = account_handle.read().await;
    for message in account.list_messages(0, 0, Some(MessageType::Unconfirmed)) {
        monitor_confirmation_state_change(account_handle.clone(), message.id()).await?;
    }
    Ok(())
}

/// Monitor message for confirmation state.
pub async fn monitor_confirmation_state_change(
    account_handle: AccountHandle,
    message_id: &MessageId,
) -> crate::Result<()> {
    let (message, client_options) = {
        let account = account_handle.read().await;
        let message = account
            .messages()
            .iter()
            .find(|message| message.id() == message_id)
            .unwrap()
            .clone();
        (message, account.client_options().clone())
    };
    let message_id = *message_id;

    subscribe_to_topic(
        &client_options,
        format!("messages/{}/metadata", message_id.to_string()),
        move |topic_event| {
            let account_handle = account_handle.clone();
            crate::block_on(async {
                if let Err(e) =
                    process_metadata(topic_event.payload.clone(), account_handle, message_id, &message).await
                {
                    log::error!("[MQTT] error processing metadata: {:?}", e);
                }
            });
        },
    )
    .await
}

async fn process_metadata(
    payload: String,
    account_handle: AccountHandle,
    message_id: MessageId,
    message: &Message,
) -> crate::Result<()> {
    let metadata: MessageMetadata = serde_json::from_str(&payload)?;

    if let Some(inclusion_state) = metadata.ledger_inclusion_state {
        let confirmed = inclusion_state == "included";
        if message.confirmed().is_none() || confirmed != message.confirmed().unwrap() {
            let mut account = account_handle.write().await;

            account.do_mut(|account| {
                let messages = account.messages_mut();
                let account_message = messages.iter_mut().find(|m| m.id() == &message_id).unwrap();
                account_message.set_confirmed(Some(confirmed));
            });

            crate::event::emit_confirmation_state_change(account.id(), &message, confirmed);
        }
    }
    Ok(())
}

1	// Copyright 2020 IOTA Stiftung	×
2	// SPDX-License-Identifier: Apache-2.0
3
4	use crate::{
5	account::AccountHandle,
6	address::{AddressOutput, IotaAddress},
7	client::ClientOptions,
8	message::{Message, MessageType},
9	};
10
11	use iota::{message::prelude::MessageId, MessageMetadata, OutputMetadata, Topic, TopicEvent};
12	use serde::Deserialize;
13
14	use std::convert::TryInto;
15
16	#[derive(Deserialize)]	×
17	struct AddressOutputPayload {
18	#[serde(rename = "messageId")]
19	message_id: String,
20	#[serde(rename = "transactionId")]
21	transaction_id: String,
22	#[serde(rename = "outputIndex")]
23	output_index: u16,
24	#[serde(rename = "isSpent")]
25	is_spent: bool,
26	output: AddressOutputPayloadOutput,
27	}
28
29	#[derive(Deserialize)]	×
30	struct AddressOutputPayloadOutput {
31	amount: u64,
32	}
33
34	/// Unsubscribe from all topics associated with the account.
35	pub async fn unsubscribe(account_handle: AccountHandle) -> crate::Result<()> {	16!
36	let account = account_handle.read().await;	×
37	let client = crate::client::get_client(account.client_options());	×
38	let mut client = client.write().await;	×
39	client.subscriber().unsubscribe()?;	×
40	Ok(())	×
41	}	16!
42
43	async fn subscribe_to_topic<C: Fn(&TopicEvent) + Send + Sync + 'static>(	×
44	client_options: &ClientOptions,	×
45	topic: String,	×
46	handler: C,	×
47	) -> crate::Result<()> {	×
48	let client = crate::client::get_client(&client_options);	×
49	let mut client = client.write().await;	×
50	client.subscriber().with_topic(Topic::new(topic)?).subscribe(handler)?;	×
51	Ok(())	×
52	}	×
53
54	/// Monitor account addresses for balance changes.
55	pub async fn monitor_account_addresses_balance(account_handle: AccountHandle) -> crate::Result<()> {	×
56	let account = account_handle.read().await;	×
57	for address in account.addresses() {	×
58	monitor_address_balance(account_handle.clone(), address.address()).await?;	×
59	}
60	Ok(())	×
61	}	×
62
63	/// Monitor address for balance changes.
64	pub async fn monitor_address_balance(account_handle: AccountHandle, address: &IotaAddress) -> crate::Result<()> {	×
65	let client_options = account_handle.client_options().await;	×
66	let client_options_ = client_options.clone();	×
67	let address = address.clone();	×
68
69	subscribe_to_topic(	×
70	&client_options_,	×
71	format!("addresses/{}/outputs", address.to_bech32()),	×
72	move \|topic_event\| {	×
73	log::info!("[MQTT] got {:?}", topic_event);	×
74	let topic_event = topic_event.clone();	×
75	let address = address.clone();	×
76	let client_options = client_options.clone();	×
77	let account_handle = account_handle.clone();	×
78
79	crate::block_on(async {	×
80	if let Err(e) =	×
81	process_output(topic_event.payload.clone(), account_handle, address, client_options).await	×
82	{
83	log::error!("[MQTT] error processing output: {:?}", e);	×
84	}	×
85	});	×
86	},	×
87	)	×
88	.await
89	}	×
90
91	async fn process_output(	×
92	payload: String,	×
93	account_handle: AccountHandle,	×
94	address: IotaAddress,	×
95	client_options: ClientOptions,	×
96	) -> crate::Result<()> {	×
97	let output: AddressOutputPayload = serde_json::from_str(&payload)?;	×
98	let metadata = OutputMetadata {	×
NEW 99	message_id: hex::decode(output.message_id).map_err(\|_\| crate::Error::InvalidMessageId)?,	×
NEW 100	transaction_id: hex::decode(output.transaction_id).map_err(\|_\| crate::Error::InvalidTransactionId)?,	×
101	output_index: output.output_index,	×
102	is_spent: output.is_spent,	×
103	amount: output.output.amount,	×
104	address: address.clone(),	×
105	};	×
106	let address_output: AddressOutput = metadata.try_into()?;	×
107
108	let client_options_ = client_options.clone();	×
109	let message_id = address_output.message_id();	×
110	let message_id_ = *message_id;	×
111
112	let message = {
113	let client = crate::client::get_client(&client_options_);	×
114	let client = client.read().await;	×
115	client.get_message().data(&message_id_).await?	×
116	};	×
117
118	let mut account = account_handle.write().await;	×
119	let account_id = account.id().clone();	×
120	let message_id_ = *message_id;	×
121
122	let addresses = account.addresses_mut();	×
123	let address_to_update = addresses.iter_mut().find(\|a\| a.address() == &address).unwrap();	×
124	address_to_update.handle_new_output(address_output);	×
125	crate::event::emit_balance_change(&account_id, &address_to_update, *address_to_update.balance());	×
126
127	match account.messages_mut().iter().position(\|m\| m.id() == &message_id_) {	×
128	Some(message_index) => {	×
129	account.do_mut(\|account\| {	×
130	let message = &mut account.messages_mut()[message_index];	×
131	message.set_confirmed(Some(true));	×
132	});	×
133	}
134	None => {
135	let message = Message::from_iota_message(message_id_, account.addresses(), &message, Some(true)).unwrap();	×
136	crate::event::emit_transaction_event(	×
137	crate::event::TransactionEventType::NewTransaction,	×
138	account.id(),	×
139	&message,
140	);
141	account.do_mut(\|account\| {	×
142	account.messages_mut().push(message);	×
143	});	×
144	}	×
145	}
146	Ok(())	×
147	}	×
148
149	/// Monitor the account's unconfirmed messages for confirmation state change.
150	pub async fn monitor_unconfirmed_messages(account_handle: AccountHandle) -> crate::Result<()> {	×
151	let account = account_handle.read().await;	×
152	for message in account.list_messages(0, 0, Some(MessageType::Unconfirmed)) {	×
153	monitor_confirmation_state_change(account_handle.clone(), message.id()).await?;	×
154	}	×
155	Ok(())	×
156	}	×
157
158	/// Monitor message for confirmation state.
159	pub async fn monitor_confirmation_state_change(	×
160	account_handle: AccountHandle,	×
161	message_id: &MessageId,	×
162	) -> crate::Result<()> {	×
163	let (message, client_options) = {	×
164	let account = account_handle.read().await;	×
165	let message = account	×
166	.messages()
167	.iter()
168	.find(\|message\| message.id() == message_id)	×
169	.unwrap()
170	.clone();
171	(message, account.client_options().clone())	×
172	};	×
173	let message_id = *message_id;	×
174
175	subscribe_to_topic(	×
176	&client_options,	×
177	format!("messages/{}/metadata", message_id.to_string()),	×
178	move \|topic_event\| {	×
179	let account_handle = account_handle.clone();	×
180	crate::block_on(async {	×
181	if let Err(e) =	×
182	process_metadata(topic_event.payload.clone(), account_handle, message_id, &message).await	×
183	{
184	log::error!("[MQTT] error processing metadata: {:?}", e);	×
185	}	×
186	});	×
187	},	×
188	)	×
189	.await
190	}	×
191
192	async fn process_metadata(	×
193	payload: String,	×
194	account_handle: AccountHandle,	×
195	message_id: MessageId,	×
196	message: &Message,	×
197	) -> crate::Result<()> {	×
198	let metadata: MessageMetadata = serde_json::from_str(&payload)?;	×
199
200	if let Some(inclusion_state) = metadata.ledger_inclusion_state {	×
201	let confirmed = inclusion_state == "included";	×
202	if message.confirmed().is_none() \|\| confirmed != message.confirmed().unwrap() {	×
203	let mut account = account_handle.write().await;	×
204
205	account.do_mut(\|account\| {	×
206	let messages = account.messages_mut();	×
207	let account_message = messages.iter_mut().find(\|m\| m.id() == &message_id).unwrap();	×
208	account_message.set_confirmed(Some(confirmed));	×
209	});	×
210
211	crate::event::emit_confirmation_state_change(account.id(), &message, confirmed);	×
212	}	×
213	}	×
214	Ok(())	×
215	}	×

iotaledger / wallet.rs / 488949865

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous