/// Transform the proof of indexing changes into entity updates that will be
/// inserted when as_modifications is called.
Transform the proof of indexing changes into entity updates that will be
inserted when as_modifications is called.
Transform the proof of indexing changes into entity updates that will be inserted when as_modifications is called .
async fn update_proof_of_indexing(
    proof_of_indexing: ProofOfIndexing,
    stopwatch: &StopwatchMetrics,
    deployment_id: &DeploymentHash,
    entity_cache: &mut EntityCache,
) -> Result<(), Error> {
    let _section_guard = stopwatch.start_section("update_proof_of_indexing");
    let mut proof_of_indexing = proof_of_indexing.take();
    for (causality_region, stream) in proof_of_indexing.drain() {
        // Create the special POI entity key specific to this causality_region
        let entity_key = EntityKey {
            subgraph_id: deployment_id.clone(),
            entity_type: POI_OBJECT.to_owned(),
            entity_id: causality_region,
        };
        // Grab the current digest attribute on this entity
        let prev_poi =
            entity_cache
                .get(&entity_key)
                .map_err(Error::from)?
                .map(|entity| match entity.get("digest") {
                    Some(Value::Bytes(b)) => b.clone(),
                    _ => panic!("Expected POI entity to have a digest and for it to be bytes"),
                });
        // Finish the POI stream, getting the new POI value.
        let updated_proof_of_indexing = stream.pause(prev_poi.as_deref());
        let updated_proof_of_indexing: Bytes = (&updated_proof_of_indexing[..]).into();
        // Put this onto an entity with the same digest attribute
        // that was expected before when reading.
        let new_poi_entity = entity! {
            id: entity_key.entity_id.clone(),
            digest: updated_proof_of_indexing,
        };
        entity_cache.set(entity_key, new_poi_entity)?;
    }
    Ok(())
}
async fn update_proof_of_indexing ( proof_of_indexing : ProofOfIndexing , stopwatch : & StopwatchMetrics , deployment_id : & DeploymentHash , entity_cache : & mut EntityCache , ) -> Result < ( ) , Error > { let _section_guard = stopwatch . start_section ( " " ) ; let mut proof_of_indexing = proof_of_indexing . take ( ) ; for ( causality_region , stream ) in proof_of_indexing . drain ( ) { // Create the special POI entity key specific to this causality_region let entity_key = EntityKey { subgraph_id : deployment_id . clone ( ) , entity_type : POI_OBJECT . to_owned ( ) , entity_id : causality_region , } ; // Grab the current digest attribute on this entity let prev_poi = entity_cache . get ( & entity_key ) . map_err ( Error :: from ) ? . map ( | entity | match entity . get ( " " ) { Some ( Value :: Bytes ( b ) ) => b . clone ( ) , _ => panic ! ( " " ) , } ) ; // Finish the POI stream, getting the new POI value. let updated_proof_of_indexing = stream . pause ( prev_poi . as_deref ( ) ) ; let updated_proof_of_indexing : Bytes = ( & updated_proof_of_indexing [ .. ] ) . into ( ) ; // Put this onto an entity with the same digest attribute // that was expected before when reading. let new_poi_entity = entity ! { id entity_key entity_id clone ( ) digest updated_proof_of_indexing } ; entity_cache . set ( entity_key , new_poi_entity ) ? ; } Ok ( ( ) ) }
Transform the proof of indexing changes into entity updates that will be
inserted when as_modifications is called.
Transform the proof of indexing changes into entity updates that will be inserted when as_modifications is called .
proof_of_indexing
ProofOfIndexing
stopwatch
&StopwatchMetrics
deployment_id
&DeploymentHash
entity_cache
&mut EntityCache