fn render_persistence_operators<G>(
        &self,
        source_name: String,
        ts_bindings_stream: Stream<G, ((SourceTimestamp, AssignedTimestamp), Timestamp, Diff)>,
    ) -> (
        Stream<G, ((SourceTimestamp, AssignedTimestamp), Timestamp, Diff)>,
        Stream<G, SourceError>,
    )
    where
        G: Scope<Timestamp = Timestamp>,
    {
        let persist_operator_name = format!("{}-timestamp-bindings", self.source_name);

        let (ts_bindings_stream, ts_bindings_persist_err) =
            ts_bindings_stream.persist(&persist_operator_name, self.config.write_handle.clone());

        // We're throwing away the differential information that we theoretically get from
        // `persist()`. We have to do this because sources currently only emit a `Stream<_,
        // SourceError>`. In practice, persist errors are non-retractable, currently.
        let ts_bindings_persist_err = ts_bindings_persist_err.map(move |(error, _ts, _diff)| {
            SourceError::new(source_name.clone(), SourceErrorDetails::Persistence(error))
        });

        (ts_bindings_stream, ts_bindings_persist_err)
    }
/// Renders operators that persist the given `ts_bindings_stream` to the configured persistent
/// collection.
///
/// This does not seal the persistent collection, calling code must ensure that this happens
/// eventually.
Renders operators that persist the given `ts_bindings_stream` to the configured persistent
collection.

This does not seal the persistent collection, calling code must ensure that this happens
eventually.
Renders operators that persist the given ` ts_bindings_stream ` to the configured persistent collection . This does not seal the persistent collection calling code must ensure that this happens eventually .
fn render_persistence_operators<G>(
        &self,
        source_name: String,
        ts_bindings_stream: Stream<G, ((SourceTimestamp, AssignedTimestamp), Timestamp, Diff)>,
    ) -> (
        Stream<G, ((SourceTimestamp, AssignedTimestamp), Timestamp, Diff)>,
        Stream<G, SourceError>,
    )
    where
        G: Scope<Timestamp = Timestamp>,
    {
        let persist_operator_name = format!("{}-timestamp-bindings", self.source_name);
        let (ts_bindings_stream, ts_bindings_persist_err) =
            ts_bindings_stream.persist(&persist_operator_name, self.config.write_handle.clone());
        // We're throwing away the differential information that we theoretically get from
        // `persist()`. We have to do this because sources currently only emit a `Stream<_,
        // SourceError>`. In practice, persist errors are non-retractable, currently.
        let ts_bindings_persist_err = ts_bindings_persist_err.map(move |(error, _ts, _diff)| {
            SourceError::new(source_name.clone(), SourceErrorDetails::Persistence(error))
        });
        (ts_bindings_stream, ts_bindings_persist_err)
    }
fn render_persistence_operators < G > ( & self , source_name : String , ts_bindings_stream : Stream < G , ( ( SourceTimestamp , AssignedTimestamp ) , Timestamp , Diff ) > , ) -> ( Stream < G , ( ( SourceTimestamp , AssignedTimestamp ) , Timestamp , Diff ) > , Stream < G , SourceError > , ) where G : Scope < Timestamp = Timestamp > , { let persist_operator_name = format ! ( " " self source_name ) ; let ( ts_bindings_stream , ts_bindings_persist_err ) = ts_bindings_stream . persist ( & persist_operator_name , self . config . write_handle . clone ( ) ) ; // We're throwing away the differential information that we theoretically get from // `persist()`. We have to do this because sources currently only emit a `Stream<_, // SourceError>`. In practice, persist errors are non-retractable, currently. let ts_bindings_persist_err = ts_bindings_persist_err . map ( move | ( error , _ts , _diff ) | { SourceError :: new ( source_name . clone ( ) , SourceErrorDetails :: Persistence ( error ) ) } ) ; ( ts_bindings_stream , ts_bindings_persist_err ) }
Renders operators that persist the given `ts_bindings_stream` to the configured persistent
collection.
Renders operators that persist the given ` ts_bindings_stream ` to the configured persistent collection .
&self
source_name
String
ts_bindings_stream
Stream<G, ((SourceTimestamp, AssignedTimestamp), Timestamp, Diff)>
&self
source_name
String
ts_bindings_stream
Stream<G, ((SourceTimestamp, AssignedTimestamp), Timestamp, Diff)>