fn write_to_db(
        db_config: RocksDbConfig,
        db: Arc<rocks_db>,
        receiver: mpsc::Receiver<KeyVal>,
        shutdown: Arc<AtomicBool>,
    ) {
        loop {
            let data: Vec<KeyVal> = receiver.try_iter().collect();

            //timeout, no data received. let's sleep
            if data.is_empty() {
                if shutdown.load(Ordering::SeqCst) {
                    info!("Shutdown received. Exiting while loop");
                    return;
                }
                thread::sleep(Duration::from_millis(
                    db_config.async_writer_threads_sleep_ms,
                ));

                continue;
            }

            //we got data, write to db as a single record
            if data.len() < db_config.min_count_for_batch_write {
                for kv in data.iter() {
                    if let Err(e) = db.put(&kv.key, &kv.val) {
                        error!("Failed to batch write to RocksDB. Error:{:?}", e);
                    }
                }
                continue;
            }
            // write data as batch
            let mut batch = WriteBatch::default();
            for kv in data.iter() {
                if let Err(e) = batch.put(&kv.key, &kv.val) {
                    error!(
                        "Failed to add into the batch for writing to RocksDB. Error:{:?}",
                        e
                    );
                }
            }

            if db_config.disable_wal {
                if let Err(e) = db.write_without_wal(batch) {
                    error!("Failed to batch write to RocksDB. Error:{:?}", e);
                }
            } else if let Err(e) = db.write(batch) {
                error!("Failed to batch write to RocksDB. Error:{:?}", e);
            }
        }
    }
/// Write to database async
/// It reads from the channel
Write to database async
It reads from the channel
Write to database async It reads from the channel
fn write_to_db(
        db_config: RocksDbConfig,
        db: Arc<rocks_db>,
        receiver: mpsc::Receiver<KeyVal>,
        shutdown: Arc<AtomicBool>,
    ) {
        loop {
            let data: Vec<KeyVal> = receiver.try_iter().collect();
            //timeout, no data received. let's sleep
            if data.is_empty() {
                if shutdown.load(Ordering::SeqCst) {
                    info!("Shutdown received. Exiting while loop");
                    return;
                }
                thread::sleep(Duration::from_millis(
                    db_config.async_writer_threads_sleep_ms,
                ));
                continue;
            }
            //we got data, write to db as a single record
            if data.len() < db_config.min_count_for_batch_write {
                for kv in data.iter() {
                    if let Err(e) = db.put(&kv.key, &kv.val) {
                        error!("Failed to batch write to RocksDB. Error:{:?}", e);
                    }
                }
                continue;
            }
            // write data as batch
            let mut batch = WriteBatch::default();
            for kv in data.iter() {
                if let Err(e) = batch.put(&kv.key, &kv.val) {
                    error!(
                        "Failed to add into the batch for writing to RocksDB. Error:{:?}",
                        e
                    );
                }
            }
            if db_config.disable_wal {
                if let Err(e) = db.write_without_wal(batch) {
                    error!("Failed to batch write to RocksDB. Error:{:?}", e);
                }
            } else if let Err(e) = db.write(batch) {
                error!("Failed to batch write to RocksDB. Error:{:?}", e);
            }
        }
    }
fn write_to_db ( db_config : RocksDbConfig , db : Arc < rocks_db > , receiver : mpsc :: Receiver < KeyVal > , shutdown : Arc < AtomicBool > , ) { loop { let data : Vec < KeyVal > = receiver . try_iter ( ) . collect ( ) ; //timeout, no data received. let's sleep if data . is_empty ( ) { if shutdown . load ( Ordering :: SeqCst ) { info ! ( " " ) ; return ; } thread :: sleep ( Duration :: from_millis ( db_config . async_writer_threads_sleep_ms , ) ) ; continue ; } //we got data, write to db as a single record if data . len ( ) < db_config . min_count_for_batch_write { for kv in data . iter ( ) { if let Err ( e ) = db . put ( & kv . key , & kv . val ) { error ! ( " " e ) ; } } continue ; } // write data as batch let mut batch = WriteBatch :: default ( ) ; for kv in data . iter ( ) { if let Err ( e ) = batch . put ( & kv . key , & kv . val ) { error ! ( " " e ) ; } } if db_config . disable_wal { if let Err ( e ) = db . write_without_wal ( batch ) { error ! ( " " e ) ; } } else if let Err ( e ) = db . write ( batch ) { error ! ( " " e ) ; } } }
Write to database async
It reads from the channel
Write to database async It reads from the channel
db_config
RocksDbConfig
db
Arc<rocks_db>
receiver
mpsc::Receiver<KeyVal>
shutdown
Arc<AtomicBool>