pub fn new(store: Arc<dyn KVStore>) -> Result<Transactor> {
        let (send, recv) = mpsc::channel();

        match store.get_metadata() {
            Ok(metadata) => {
                let mut next_id = metadata.next_id;
                let last_id = metadata.last_indexed_tx;
                let mut latest_tx = last_id;
                let mut db = Db::new(metadata, store.clone());
                let novelty = store.get_txs(last_id)?;
                for tx in novelty {
                    for record in tx.records {
                        let Entity(e) = record.entity;
                        if e > next_id {
                            next_id = e + 1;
                        }
                        db = db.add_record(record)?;
                    }

                    latest_tx = tx.id;
                }

                Ok(Transactor {
                    next_id,
                    store: store.clone(),
                    latest_tx: latest_tx,
                    last_indexed_tx: last_id,
                    current_db: db,
                    send,
                    recv,
                    catchup_txs: None,
                    throttled: false,
                })
            }
            // FIXME: this should happen if metadata is None, not on error
            Err(_) => {
                let (current_db, next_id) = create_db(store.clone())?;
                let mut tx = Transactor {
                    next_id,
                    store: store,
                    latest_tx: 0,
                    last_indexed_tx: -1,
                    current_db,
                    send,
                    recv,
                    catchup_txs: None,
                    throttled: false,
                };

                save_metadata(&tx.current_db, tx.next_id, tx.last_indexed_tx)?;

                // We need to persist the bootstrapping data because
                // it's not in the transaction log.
                // FIXME: unwind this from the channel communication code
                // which isn't needed here
                tx.rebuild_indices();
                match tx.recv.recv().unwrap() {
                    Event::RebuiltIndex(new_db) => {
                        tx.switch_to_rebuilt_indexes(new_db)?;
                    },
                    // no one can send messages on this channel before
                    // we return the transactor, so the only message
                    // that can arrive is the one notifying that the
                    // rebuild is complete
                    _ => unreachable!()
                }
                Ok(tx)
            }
        }
    }
/// Creates a transactor by retrieving the database metadata from
/// the store (if it exists already) or creating the metadata for
/// a new database (if no metadata is present in the store).
Creates a transactor by retrieving the database metadata from
the store (if it exists already) or creating the metadata for
a new database (if no metadata is present in the store).
Creates a transactor by retrieving the database metadata from the store ( if it exists already ) or creating the metadata for a new database ( if no metadata is present in the store ) .
pub fn new(store: Arc<dyn KVStore>) -> Result<Transactor> {
        let (send, recv) = mpsc::channel();
        match store.get_metadata() {
            Ok(metadata) => {
                let mut next_id = metadata.next_id;
                let last_id = metadata.last_indexed_tx;
                let mut latest_tx = last_id;
                let mut db = Db::new(metadata, store.clone());
                let novelty = store.get_txs(last_id)?;
                for tx in novelty {
                    for record in tx.records {
                        let Entity(e) = record.entity;
                        if e > next_id {
                            next_id = e + 1;
                        }
                        db = db.add_record(record)?;
                    }
                    latest_tx = tx.id;
                }
                Ok(Transactor {
                    next_id,
                    store: store.clone(),
                    latest_tx: latest_tx,
                    last_indexed_tx: last_id,
                    current_db: db,
                    send,
                    recv,
                    catchup_txs: None,
                    throttled: false,
                })
            }
            // FIXME: this should happen if metadata is None, not on error
            Err(_) => {
                let (current_db, next_id) = create_db(store.clone())?;
                let mut tx = Transactor {
                    next_id,
                    store: store,
                    latest_tx: 0,
                    last_indexed_tx: -1,
                    current_db,
                    send,
                    recv,
                    catchup_txs: None,
                    throttled: false,
                };
                save_metadata(&tx.current_db, tx.next_id, tx.last_indexed_tx)?;
                // We need to persist the bootstrapping data because
                // it's not in the transaction log.
                // FIXME: unwind this from the channel communication code
                // which isn't needed here
                tx.rebuild_indices();
                match tx.recv.recv().unwrap() {
                    Event::RebuiltIndex(new_db) => {
                        tx.switch_to_rebuilt_indexes(new_db)?;
                    },
                    // no one can send messages on this channel before
                    // we return the transactor, so the only message
                    // that can arrive is the one notifying that the
                    // rebuild is complete
                    _ => unreachable!()
                }
                Ok(tx)
            }
        }
    }
pub fn new ( store : Arc < dyn KVStore > ) -> Result < Transactor > { let ( send , recv ) = mpsc :: channel ( ) ; match store . get_metadata ( ) { Ok ( metadata ) => { let mut next_id = metadata . next_id ; let last_id = metadata . last_indexed_tx ; let mut latest_tx = last_id ; let mut db = Db :: new ( metadata , store . clone ( ) ) ; let novelty = store . get_txs ( last_id ) ? ; for tx in novelty { for record in tx . records { let Entity ( e ) = record . entity ; if e > next_id { next_id = e + 1 ; } db = db . add_record ( record ) ? ; } latest_tx = tx . id ; } Ok ( Transactor { next_id , store : store . clone ( ) , latest_tx : latest_tx , last_indexed_tx : last_id , current_db : db , send , recv , catchup_txs : None , throttled : false , } ) } // FIXME: this should happen if metadata is None, not on error Err ( _ ) => { let ( current_db , next_id ) = create_db ( store . clone ( ) ) ? ; let mut tx = Transactor { next_id , store : store , latest_tx : 0 , last_indexed_tx : - 1 , current_db , send , recv , catchup_txs : None , throttled : false , } ; save_metadata ( & tx . current_db , tx . next_id , tx . last_indexed_tx ) ? ; // We need to persist the bootstrapping data because // it's not in the transaction log. // FIXME: unwind this from the channel communication code // which isn't needed here tx . rebuild_indices ( ) ; match tx . recv . recv ( ) . unwrap ( ) { Event :: RebuiltIndex ( new_db ) => { tx . switch_to_rebuilt_indexes ( new_db ) ? ; } , // no one can send messages on this channel before // we return the transactor, so the only message // that can arrive is the one notifying that the // rebuild is complete _ => unreachable ! ( ) } Ok ( tx ) } } }
Creates a transactor by retrieving the database metadata from
the store (if it exists already) or creating the metadata for
a new database (if no metadata is present in the store).
Creates a transactor by retrieving the database metadata from the store ( if it exists already ) or creating the metadata for a new database ( if no metadata is present in the store ) .
store
Arc<dyn KVStore>
store
Arc<dyn KVStore>