490410401

Build Type

github

Committed by GitHub

Commit Message

Merge 530c5b66a into 040b4a723

Pull Request Pull Request #222: fix(account): persist to storage after every mutable operation

Run Details

4071 of 21058 branches covered (19.33%)

Branch coverage included in aggregate %.

95 of 135 new or added lines in 5 files covered. (70.37%)

11 existing lines in 3 files now uncovered.

1579 of 3363 relevant lines covered (46.95%)

388.26 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)?,
        transaction_id: hex::decode(output.transaction_id)?,
        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));
                    Ok(())
                })
                .await?;
        }
        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(())
                })
                .await?;
        }
    }
    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));
                    Ok(())
                })
                .await?;

            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 {	×
99	message_id: hex::decode(output.message_id)?,	×
100	transaction_id: hex::decode(output.transaction_id)?,	×
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) => {	×
NEW 129	account	×
NEW 130	.do_mut(\|account\| {	×
NEW 131	let message = &mut account.messages_mut()[message_index];	×
NEW 132	message.set_confirmed(Some(true));	×
NEW 133	Ok(())	×
NEW 134	})	×
NEW 135	.await?;	×
136	}
137	None => {
138	let message = Message::from_iota_message(message_id_, account.addresses(), &message, Some(true)).unwrap();	×
139	crate::event::emit_transaction_event(	×
140	crate::event::TransactionEventType::NewTransaction,	×
141	account.id(),	×
UNCOV 142	&message,	×
143	);
NEW 144	account	×
NEW 145	.do_mut(\|account\| {	×
NEW 146	account.messages_mut().push(message);	×
NEW 147	Ok(())	×
NEW 148	})	×
NEW 149	.await?;	×
UNCOV 150	}	×
151	}
152	Ok(())	×
153	}	×
154
155	/// Monitor the account's unconfirmed messages for confirmation state change.
156	pub async fn monitor_unconfirmed_messages(account_handle: AccountHandle) -> crate::Result<()> {	×
157	let account = account_handle.read().await;	×
158	for message in account.list_messages(0, 0, Some(MessageType::Unconfirmed)) {	×
159	monitor_confirmation_state_change(account_handle.clone(), message.id()).await?;	×
160	}	×
161	Ok(())	×
162	}	×
163
164	/// Monitor message for confirmation state.
165	pub async fn monitor_confirmation_state_change(	×
166	account_handle: AccountHandle,	×
167	message_id: &MessageId,	×
168	) -> crate::Result<()> {	×
169	let (message, client_options) = {	×
170	let account = account_handle.read().await;	×
171	let message = account	×
172	.messages()
173	.iter()
174	.find(\|message\| message.id() == message_id)	×
175	.unwrap()
176	.clone();
177	(message, account.client_options().clone())	×
178	};	×
179	let message_id = *message_id;	×
180
181	subscribe_to_topic(	×
182	&client_options,	×
183	format!("messages/{}/metadata", message_id.to_string()),	×
184	move \|topic_event\| {	×
185	let account_handle = account_handle.clone();	×
186	crate::block_on(async {	×
187	if let Err(e) =	×
188	process_metadata(topic_event.payload.clone(), account_handle, message_id, &message).await	×
189	{
190	log::error!("[MQTT] error processing metadata: {:?}", e);	×
191	}	×
192	});	×
193	},	×
194	)	×
195	.await
196	}	×
197
198	async fn process_metadata(	×
199	payload: String,	×
200	account_handle: AccountHandle,	×
201	message_id: MessageId,	×
202	message: &Message,	×
203	) -> crate::Result<()> {	×
204	let metadata: MessageMetadata = serde_json::from_str(&payload)?;	×
205
206	if let Some(inclusion_state) = metadata.ledger_inclusion_state {	×
207	let confirmed = inclusion_state == "included";	×
208	if message.confirmed().is_none() \|\| confirmed != message.confirmed().unwrap() {	×
209	let mut account = account_handle.write().await;	×
210
NEW 211	account	×
NEW 212	.do_mut(\|account\| {	×
NEW 213	let messages = account.messages_mut();	×
NEW 214	let account_message = messages.iter_mut().find(\|m\| m.id() == &message_id).unwrap();	×
NEW 215	account_message.set_confirmed(Some(confirmed));	×
NEW 216	Ok(())	×
NEW 217	})	×
NEW 218	.await?;	×
219
220	crate::event::emit_confirmation_state_change(account.id(), &message, confirmed);	×
221	}	×
222	}	×
223	Ok(())	×
224	}	×

iotaledger / wallet.rs / 490410401

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