pub fn store_region(&mut self, region_id: RegionID, orders: TaggedOrderVec) -> UpdateResult {
        // Acquire write lock
        // Iterate over orders, serialize (protobuf, append seen_at field later) and hash (build StoreObjects) -> rayon
        //  Check if hash is present
        //      yes -> Is the hash different?
        //          yes -> Persist in Store (and merge history), add to updated
        //          no -> Discard Order, add to unaffected
        //      no -> Persist in store, update indices, add to new
        // prune region
        // Release write lock
        // Return UpdateResult updated (caller can calculate total region_types affected after pruning)
        let mut new: FnvHashSet<OrderID> = FnvHashSet::default();
        let mut updated: FnvHashSet<OrderID> = FnvHashSet::default();
        let mut unaffected: FnvHashSet<OrderID> = FnvHashSet::default();
        let mut region_types: FnvHashSet<RegionType> = FnvHashSet::default();
        let mut all_oids: FnvHashSet<OrderID> = FnvHashSet::default();

        let mut store = self.0.write();

        // Check if there are any orders for this region
        // if not, this is out first run and we avoid flooding new orders
        let is_first_run = !store.regions.contains_key(&region_id);

        for (mut order, last_modified) in orders {
            // Add to all list
            all_oids.insert(order.order_id);

            // Add region ID to order
            order.region_id = region_id;

            // Encode without seen_at
            // TODO: parallelize encode/hash, add seen_at step using rayon?
            let mut hash_blob = Vec::with_capacity(Message::encoded_len(&order));
            Message::encode(&order, &mut hash_blob).unwrap();
            // Hash the blob
            let mut hasher = FnvHasher::default();
            hash_blob.as_slice().hash(&mut hasher);
            let hash = hasher.finish();

            if store.orders.contains_key(&order.order_id) {
                let entry = store.orders.get_mut(&order.order_id).unwrap();

                // Check if hash is different
                if entry.hash != hash {
                    // Set seen_at and reencode
                    order.seen_at = Some(Timestamp::from(time::SystemTime::from(last_modified)));

                    let mut seen_blob = Vec::with_capacity(Message::encoded_len(&order));
                    Message::encode(&order, &mut seen_blob).unwrap();

                    // Update entry
                    updated.insert(order.order_id);
                    region_types.insert((region_id, order.type_id));
                    entry.hash = hash;
                    entry.history.push(seen_blob.clone());
                    entry.order = seen_blob;
                } else {
                    unaffected.insert(order.order_id);
                }
            } else {
                // Not there, new entry
                // Set seen_at and reencode
                order.seen_at = Some(Timestamp::from(time::SystemTime::from(last_modified)));

                let mut seen_blob = Vec::with_capacity(Message::encoded_len(&order));
                Message::encode(&order, &mut seen_blob).unwrap();

                new.insert(order.order_id);
                region_types.insert((region_id, order.type_id));

                store.orders.insert(order.order_id,
                                    StoreObject { hash,
                                                  history: vec![seen_blob.clone()],
                                                  order: seen_blob,
                                                  region_id,
                                                  type_id: order.type_id });

                add_to_indices(&mut store, order.order_id, order.type_id, region_id);
            }
        }

        let (rt_closed, closed) = prune_region(&mut store, region_id, &all_oids);
        region_types.extend(rt_closed.iter());

        // Push result to stream and return
        let update_result = UpdateResult { is_first_run,
                                           new,
                                           updated,
                                           unaffected,
                                           closed,
                                           region_types };

        store.update_results_streams.drain_filter(|stream| {
            match stream.try_send(convert_for_stream(&update_result)) {
                Ok(()) => false,
                Err(mpsc::error::TrySendError::Closed(_)) => {
                    debug!("Result stream client disconnected.");
                    true
                }
                Err(mpsc::error::TrySendError::Full(_)) => {
                    warn!("Closing result stream due to backpressured client.");
                    true
                }
            }
        });
        update_result
    }
/// Store multiple orders in the store. Update if existing
Store multiple orders in the store. Update if existing
Store multiple orders in the store . Update if existing
pub fn store_region(&mut self, region_id: RegionID, orders: TaggedOrderVec) -> UpdateResult {
        // Acquire write lock
        // Iterate over orders, serialize (protobuf, append seen_at field later) and hash (build StoreObjects) -> rayon
        //  Check if hash is present
        //      yes -> Is the hash different?
        //          yes -> Persist in Store (and merge history), add to updated
        //          no -> Discard Order, add to unaffected
        //      no -> Persist in store, update indices, add to new
        // prune region
        // Release write lock
        // Return UpdateResult updated (caller can calculate total region_types affected after pruning)
        let mut new: FnvHashSet<OrderID> = FnvHashSet::default();
        let mut updated: FnvHashSet<OrderID> = FnvHashSet::default();
        let mut unaffected: FnvHashSet<OrderID> = FnvHashSet::default();
        let mut region_types: FnvHashSet<RegionType> = FnvHashSet::default();
        let mut all_oids: FnvHashSet<OrderID> = FnvHashSet::default();
        let mut store = self.0.write();
        // Check if there are any orders for this region
        // if not, this is out first run and we avoid flooding new orders
        let is_first_run = !store.regions.contains_key(&region_id);
        for (mut order, last_modified) in orders {
            // Add to all list
            all_oids.insert(order.order_id);
            // Add region ID to order
            order.region_id = region_id;
            // Encode without seen_at
            // TODO: parallelize encode/hash, add seen_at step using rayon?
            let mut hash_blob = Vec::with_capacity(Message::encoded_len(&order));
            Message::encode(&order, &mut hash_blob).unwrap();
            // Hash the blob
            let mut hasher = FnvHasher::default();
            hash_blob.as_slice().hash(&mut hasher);
            let hash = hasher.finish();
            if store.orders.contains_key(&order.order_id) {
                let entry = store.orders.get_mut(&order.order_id).unwrap();
                // Check if hash is different
                if entry.hash != hash {
                    // Set seen_at and reencode
                    order.seen_at = Some(Timestamp::from(time::SystemTime::from(last_modified)));
                    let mut seen_blob = Vec::with_capacity(Message::encoded_len(&order));
                    Message::encode(&order, &mut seen_blob).unwrap();
                    // Update entry
                    updated.insert(order.order_id);
                    region_types.insert((region_id, order.type_id));
                    entry.hash = hash;
                    entry.history.push(seen_blob.clone());
                    entry.order = seen_blob;
                } else {
                    unaffected.insert(order.order_id);
                }
            } else {
                // Not there, new entry
                // Set seen_at and reencode
                order.seen_at = Some(Timestamp::from(time::SystemTime::from(last_modified)));
                let mut seen_blob = Vec::with_capacity(Message::encoded_len(&order));
                Message::encode(&order, &mut seen_blob).unwrap();
                new.insert(order.order_id);
                region_types.insert((region_id, order.type_id));
                store.orders.insert(order.order_id,
                                    StoreObject { hash,
                                                  history: vec![seen_blob.clone()],
                                                  order: seen_blob,
                                                  region_id,
                                                  type_id: order.type_id });
                add_to_indices(&mut store, order.order_id, order.type_id, region_id);
            }
        }
        let (rt_closed, closed) = prune_region(&mut store, region_id, &all_oids);
        region_types.extend(rt_closed.iter());
        // Push result to stream and return
        let update_result = UpdateResult { is_first_run,
                                           new,
                                           updated,
                                           unaffected,
                                           closed,
                                           region_types };
        store.update_results_streams.drain_filter(|stream| {
            match stream.try_send(convert_for_stream(&update_result)) {
                Ok(()) => false,
                Err(mpsc::error::TrySendError::Closed(_)) => {
                    debug!("Result stream client disconnected.");
                    true
                }
                Err(mpsc::error::TrySendError::Full(_)) => {
                    warn!("Closing result stream due to backpressured client.");
                    true
                }
            }
        });
        update_result
    }
pub fn store_region ( & mut self , region_id : RegionID , orders : TaggedOrderVec ) -> UpdateResult { // Acquire write lock // Iterate over orders, serialize (protobuf, append seen_at field later) and hash (build StoreObjects) -> rayon //  Check if hash is present //      yes -> Is the hash different? //          yes -> Persist in Store (and merge history), add to updated //          no -> Discard Order, add to unaffected //      no -> Persist in store, update indices, add to new // prune region // Release write lock // Return UpdateResult updated (caller can calculate total region_types affected after pruning) let mut new : FnvHashSet < OrderID > = FnvHashSet :: default ( ) ; let mut updated : FnvHashSet < OrderID > = FnvHashSet :: default ( ) ; let mut unaffected : FnvHashSet < OrderID > = FnvHashSet :: default ( ) ; let mut region_types : FnvHashSet < RegionType > = FnvHashSet :: default ( ) ; let mut all_oids : FnvHashSet < OrderID > = FnvHashSet :: default ( ) ; let mut store = self . 0 . write ( ) ; // Check if there are any orders for this region // if not, this is out first run and we avoid flooding new orders let is_first_run = ! store . regions . contains_key ( & region_id ) ; for ( mut order , last_modified ) in orders { // Add to all list all_oids . insert ( order . order_id ) ; // Add region ID to order order . region_id = region_id ; // Encode without seen_at // TODO: parallelize encode/hash, add seen_at step using rayon? let mut hash_blob = Vec :: with_capacity ( Message :: encoded_len ( & order ) ) ; Message :: encode ( & order , & mut hash_blob ) . unwrap ( ) ; // Hash the blob let mut hasher = FnvHasher :: default ( ) ; hash_blob . as_slice ( ) . hash ( & mut hasher ) ; let hash = hasher . finish ( ) ; if store . orders . contains_key ( & order . order_id ) { let entry = store . orders . get_mut ( & order . order_id ) . unwrap ( ) ; // Check if hash is different if entry . hash != hash { // Set seen_at and reencode order . seen_at = Some ( Timestamp :: from ( time :: SystemTime :: from ( last_modified ) ) ) ; let mut seen_blob = Vec :: with_capacity ( Message :: encoded_len ( & order ) ) ; Message :: encode ( & order , & mut seen_blob ) . unwrap ( ) ; // Update entry updated . insert ( order . order_id ) ; region_types . insert ( ( region_id , order . type_id ) ) ; entry . hash = hash ; entry . history . push ( seen_blob . clone ( ) ) ; entry . order = seen_blob ; } else { unaffected . insert ( order . order_id ) ; } } else { // Not there, new entry // Set seen_at and reencode order . seen_at = Some ( Timestamp :: from ( time :: SystemTime :: from ( last_modified ) ) ) ; let mut seen_blob = Vec :: with_capacity ( Message :: encoded_len ( & order ) ) ; Message :: encode ( & order , & mut seen_blob ) . unwrap ( ) ; new . insert ( order . order_id ) ; region_types . insert ( ( region_id , order . type_id ) ) ; store . orders . insert ( order . order_id , StoreObject { hash , history : vec ! [ seen_blob clone ( ) ] , order : seen_blob , region_id , type_id : order . type_id } ) ; add_to_indices ( & mut store , order . order_id , order . type_id , region_id ) ; } } let ( rt_closed , closed ) = prune_region ( & mut store , region_id , & all_oids ) ; region_types . extend ( rt_closed . iter ( ) ) ; // Push result to stream and return let update_result = UpdateResult { is_first_run , new , updated , unaffected , closed , region_types } ; store . update_results_streams . drain_filter ( | stream | { match stream . try_send ( convert_for_stream ( & update_result ) ) { Ok ( ( ) ) => false , Err ( mpsc :: error :: TrySendError :: Closed ( _ ) ) => { debug ! ( " " ) ; true } Err ( mpsc :: error :: TrySendError :: Full ( _ ) ) => { warn ! ( " " ) ; true } } } ) ; update_result }
Store multiple orders in the store.
Store multiple orders in the store .
&mut self
region_id
RegionID
orders
TaggedOrderVec
&mut self
region_id
RegionID
orders
TaggedOrderVec