pub async fn on_disk(
    rx: Receiver<OnDisk>,
    counter: &Mutex<u128>,
    map_counter_to_hash: &RwLock<BTreeMap<u128, Arc<Vec<u8>>>>,
    map_expiration_time_to_hashes: &RwLock<BTreeMap<i64, RwLock<HashSet<Arc<Vec<u8>>>>>>,
    map_hash_to_counter: &RwLock<HashMap<Arc<Vec<u8>>, u128>>,
    map_hash_to_expiration_time: &RwLock<HashMap<Arc<Vec<u8>>, i64>>,
    connection: &Connection,
    mutate_tx: &Sender<Mutation>,
) {
    // It is better to execute SQLite operations sequentially. SQLite locks the database
    // during an operation, so there is nothing gained from spawning dedicated tasks for
    // each operation.
    while let Ok(command) = rx.recv().await {
        match command {
            OnDisk::GetMessage(hash, tx) => {
                let mut statement = connection
                    .prepare(include_str!("../sql/B. RPC/Retrieve message.sql"))
                    .unwrap();
                let mut rows = statement.query(params![&hash as &Vec<u8>]).unwrap();
                match rows.next().unwrap() {
                    Some(row) => {
                        let payload: Vec<u8> = row.get(0).unwrap();
                        let nonce: i64 = row.get(1).unwrap();
                        let expiration_time: i64 = row.get(2).unwrap();
                        tx.send(Some(Message {
                            payload,
                            nonce,
                            expiration_time,
                        }))
                        .await;
                    }
                    None => tx.send(None).await,
                }
            }
            OnDisk::InsertMessage(
                Message {
                    payload,
                    nonce,
                    expiration_time,
                },
                event,
            ) => {
                *counter.lock().await += 1;
                let hash = Arc::new(message_hash(&payload, expiration_time).to_vec());
                add_hash(
                    hash.clone(),
                    *counter.lock().await,
                    expiration_time,
                    &map_counter_to_hash,
                    &map_expiration_time_to_hashes,
                    &map_hash_to_counter,
                    &map_hash_to_expiration_time,
                )
                .await;
                connection
                    .execute(
                        include_str!("../sql/B. RPC/Put message.sql"),
                        params![&hash as &Vec<u8>, payload, nonce, expiration_time],
                    )
                    .unwrap();
                mutate_tx.send(Mutation::Insert(hash)).await;
                event.set();
            }
        }
    }
}
/// This task executes blocking DB operations.
This task executes blocking DB operations.
This task executes blocking DB operations .
pub async fn on_disk(
    rx: Receiver<OnDisk>,
    counter: &Mutex<u128>,
    map_counter_to_hash: &RwLock<BTreeMap<u128, Arc<Vec<u8>>>>,
    map_expiration_time_to_hashes: &RwLock<BTreeMap<i64, RwLock<HashSet<Arc<Vec<u8>>>>>>,
    map_hash_to_counter: &RwLock<HashMap<Arc<Vec<u8>>, u128>>,
    map_hash_to_expiration_time: &RwLock<HashMap<Arc<Vec<u8>>, i64>>,
    connection: &Connection,
    mutate_tx: &Sender<Mutation>,
) {
    // It is better to execute SQLite operations sequentially. SQLite locks the database
    // during an operation, so there is nothing gained from spawning dedicated tasks for
    // each operation.
    while let Ok(command) = rx.recv().await {
        match command {
            OnDisk::GetMessage(hash, tx) => {
                let mut statement = connection
                    .prepare(include_str!("../sql/B. RPC/Retrieve message.sql"))
                    .unwrap();
                let mut rows = statement.query(params![&hash as &Vec<u8>]).unwrap();
                match rows.next().unwrap() {
                    Some(row) => {
                        let payload: Vec<u8> = row.get(0).unwrap();
                        let nonce: i64 = row.get(1).unwrap();
                        let expiration_time: i64 = row.get(2).unwrap();
                        tx.send(Some(Message {
                            payload,
                            nonce,
                            expiration_time,
                        }))
                        .await;
                    }
                    None => tx.send(None).await,
                }
            }
            OnDisk::InsertMessage(
                Message {
                    payload,
                    nonce,
                    expiration_time,
                },
                event,
            ) => {
                *counter.lock().await += 1;
                let hash = Arc::new(message_hash(&payload, expiration_time).to_vec());
                add_hash(
                    hash.clone(),
                    *counter.lock().await,
                    expiration_time,
                    &map_counter_to_hash,
                    &map_expiration_time_to_hashes,
                    &map_hash_to_counter,
                    &map_hash_to_expiration_time,
                )
                .await;
                connection
                    .execute(
                        include_str!("../sql/B. RPC/Put message.sql"),
                        params![&hash as &Vec<u8>, payload, nonce, expiration_time],
                    )
                    .unwrap();
                mutate_tx.send(Mutation::Insert(hash)).await;
                event.set();
            }
        }
    }
}
pub async fn on_disk ( rx : Receiver < OnDisk > , counter : & Mutex < u128 > , map_counter_to_hash : & RwLock < BTreeMap < u128 , Arc < Vec < u8 > > > > , map_expiration_time_to_hashes : & RwLock < BTreeMap < i64 , RwLock < HashSet < Arc < Vec < u8 > > > > > > , map_hash_to_counter : & RwLock < HashMap < Arc < Vec < u8 > > , u128 > > , map_hash_to_expiration_time : & RwLock < HashMap < Arc < Vec < u8 > > , i64 > > , connection : & Connection , mutate_tx : & Sender < Mutation > , ) { // It is better to execute SQLite operations sequentially. SQLite locks the database // during an operation, so there is nothing gained from spawning dedicated tasks for // each operation. while let Ok ( command ) = rx . recv ( ) . await { match command { OnDisk :: GetMessage ( hash , tx ) => { let mut statement = connection . prepare ( include_str ! ( " " ) ) . unwrap ( ) ; let mut rows = statement . query ( params ! [ hash as Vec u8 ] ) . unwrap ( ) ; match rows . next ( ) . unwrap ( ) { Some ( row ) => { let payload : Vec < u8 > = row . get ( 0 ) . unwrap ( ) ; let nonce : i64 = row . get ( 1 ) . unwrap ( ) ; let expiration_time : i64 = row . get ( 2 ) . unwrap ( ) ; tx . send ( Some ( Message { payload , nonce , expiration_time , } ) ) . await ; } None => tx . send ( None ) . await , } } OnDisk :: InsertMessage ( Message { payload , nonce , expiration_time , } , event , ) => { * counter . lock ( ) . await += 1 ; let hash = Arc :: new ( message_hash ( & payload , expiration_time ) . to_vec ( ) ) ; add_hash ( hash . clone ( ) , * counter . lock ( ) . await , expiration_time , & map_counter_to_hash , & map_expiration_time_to_hashes , & map_hash_to_counter , & map_hash_to_expiration_time , ) . await ; connection . execute ( include_str ! ( " " ) , params ! [ hash as Vec u8 payload nonce expiration_time ] , ) . unwrap ( ) ; mutate_tx . send ( Mutation :: Insert ( hash ) ) . await ; event . set ( ) ; } } } }
This task executes blocking DB operations.
This task executes blocking DB operations .
rx
Receiver<OnDisk>
counter
&Mutex<u128>
map_counter_to_hash
&RwLock<BTreeMap<u128, Arc<Vec<u8>>>>
map_expiration_time_to_hashes
&RwLock<BTreeMap<i64, RwLock<HashSet<Arc<Vec<u8>>>>>>
map_hash_to_counter
&RwLock<HashMap<Arc<Vec<u8>>, u128>>
map_hash_to_expiration_time
&RwLock<HashMap<Arc<Vec<u8>>, i64>>
connection
&Connection
mutate_tx
&Sender<Mutation>