fn subscribe_and_consume(
    store: Arc<DieselStore>,
    subgraph: &SubgraphDeploymentId,
    entity_type: &str,
) -> StoreEventStream<impl Stream<Item = StoreEvent, Error = ()> + Send> {
    static MARKER_ID: &str = "fake marker";

    let subscription = store.subscribe(vec![(subgraph.clone(), entity_type.to_owned())]);

    // Generate fake activity on the stream by removing a nonexistent entity and
    // skip anything on the stream before this faked event. That makes sure that
    // users of the stream do not see any events that were generated before the
    // subscription was started which can happen depending on how fast the event
    // processing thread is compared to the thread that runs this code.
    //
    // There is still a small possibility that earlier events get through since
    // the fake event is generated in its own transaction and Postgres is free to
    // reorder delivery of events across transactions, like those generated by
    // setup code in the tests.
    let key = EntityKey {
        subgraph_id: subgraph.clone(),
        entity_type: entity_type.to_owned(),
        entity_id: MARKER_ID.to_owned(),
    };
    let op = EntityOperation::Remove { key };

    let block_ptr = EthereumBlockPointer::from((
        H256::from("0xf1ead03f5811aa2eacbb14e90cc62bd23003086562be21fcea4292a7aa6d9d85"),
        42u64,
    ));
    let source = EventSource::EthereumBlock(block_ptr);

    store
        .apply_entity_operations(vec![op], source.clone())
        .expect("Failed to apply marker operation");

    let source = subscription
        .skip_while(move |event| {
            // Skip events until we see the fake event we generated above
            future::ok(
                event
                    .changes
                    .iter()
                    .all(|change| change.entity_id != MARKER_ID),
            )
        })
        .skip(1);
    StoreEventStream::new(source)
}
// Subscribe to store events from the store. This implementation works
// very hard to make sure that only events that were generated after the
// subscription started will actually be seen on the returned stream
Subscribe to store events from the store. This implementation works
very hard to make sure that only events that were generated after the
subscription started will actually be seen on the returned stream
Subscribe to store events from the store . This implementation works very hard to make sure that only events that were generated after the subscription started will actually be seen on the returned stream
fn subscribe_and_consume(
    store: Arc<DieselStore>,
    subgraph: &SubgraphDeploymentId,
    entity_type: &str,
) -> StoreEventStream<impl Stream<Item = StoreEvent, Error = ()> + Send> {
    static MARKER_ID: &str = "fake marker";
    let subscription = store.subscribe(vec![(subgraph.clone(), entity_type.to_owned())]);
    // Generate fake activity on the stream by removing a nonexistent entity and
    // skip anything on the stream before this faked event. That makes sure that
    // users of the stream do not see any events that were generated before the
    // subscription was started which can happen depending on how fast the event
    // processing thread is compared to the thread that runs this code.
    //
    // There is still a small possibility that earlier events get through since
    // the fake event is generated in its own transaction and Postgres is free to
    // reorder delivery of events across transactions, like those generated by
    // setup code in the tests.
    let key = EntityKey {
        subgraph_id: subgraph.clone(),
        entity_type: entity_type.to_owned(),
        entity_id: MARKER_ID.to_owned(),
    };
    let op = EntityOperation::Remove { key };
    let block_ptr = EthereumBlockPointer::from((
        H256::from("0xf1ead03f5811aa2eacbb14e90cc62bd23003086562be21fcea4292a7aa6d9d85"),
        42u64,
    ));
    let source = EventSource::EthereumBlock(block_ptr);
    store
        .apply_entity_operations(vec![op], source.clone())
        .expect("Failed to apply marker operation");
    let source = subscription
        .skip_while(move |event| {
            // Skip events until we see the fake event we generated above
            future::ok(
                event
                    .changes
                    .iter()
                    .all(|change| change.entity_id != MARKER_ID),
            )
        })
        .skip(1);
    StoreEventStream::new(source)
}
fn subscribe_and_consume ( store : Arc < DieselStore > , subgraph : & SubgraphDeploymentId , entity_type : & str , ) -> StoreEventStream < impl Stream < Item = StoreEvent , Error = ( ) > + Send > { static MARKER_ID : & str = " " ; let subscription = store . subscribe ( vec ! [ ( subgraph clone ( ) entity_type to_owned ( ) ) ] ) ; // Generate fake activity on the stream by removing a nonexistent entity and // skip anything on the stream before this faked event. That makes sure that // users of the stream do not see any events that were generated before the // subscription was started which can happen depending on how fast the event // processing thread is compared to the thread that runs this code. // // There is still a small possibility that earlier events get through since // the fake event is generated in its own transaction and Postgres is free to // reorder delivery of events across transactions, like those generated by // setup code in the tests. let key = EntityKey { subgraph_id : subgraph . clone ( ) , entity_type : entity_type . to_owned ( ) , entity_id : MARKER_ID . to_owned ( ) , } ; let op = EntityOperation :: Remove { key } ; let block_ptr = EthereumBlockPointer :: from ( ( H256 :: from ( " " ) , 42u64 , ) ) ; let source = EventSource :: EthereumBlock ( block_ptr ) ; store . apply_entity_operations ( vec ! [ op ] , source . clone ( ) ) . expect ( " " ) ; let source = subscription . skip_while ( move | event | { // Skip events until we see the fake event we generated above future :: ok ( event . changes . iter ( ) . all ( | change | change . entity_id != MARKER_ID ) , ) } ) . skip ( 1 ) ; StoreEventStream :: new ( source ) }
Subscribe to store events from the store.
Subscribe to store events from the store .
store
Arc<DieselStore>
subgraph
&SubgraphDeploymentId
entity_type
&str
store
Arc<DieselStore>
subgraph
&SubgraphDeploymentId
entity_type
&str