pub(crate) fn process_response(&mut self, response: TailResponse) -> bool {
        let mut packer = Row::default();
        match response {
            TailResponse::Progress(upper) => {
                if self.emit_progress && !upper.is_empty() {
                    assert_eq!(
                        upper.len(),
                        1,
                        "TAIL only supports single-dimensional timestamps"
                    );
                    packer.push(Datum::from(numeric::Numeric::from(*&upper[0])));
                    packer.push(Datum::True);
                    // Fill in the diff column and all table columns with NULL.
                    for _ in 0..(self.object_columns + 1) {
                        packer.push(Datum::Null);
                    }
                    let row = packer.finish_and_reuse();

                    let result = self.channel.send(vec![row]);
                    if result.is_err() {
                        // TODO(benesch): we should actually drop the sink if the
                        // receiver has gone away. E.g. form a DROP SINK command?
                    }
                }
                upper.is_empty()
            }
            TailResponse::Rows(mut rows) => {
                // Sort results by time. We use stable sort here because it will produce deterministic
                // results since the cursor will always produce rows in the same order.
                // TODO: Is sorting necessary?
                rows.sort_by_key(|(_, time, _)| *time);

                let rows = rows
                    .into_iter()
                    .map(|(row, time, diff)| {
                        packer.push(Datum::from(numeric::Numeric::from(time)));
                        if self.emit_progress {
                            // When sinking with PROGRESS, the output
                            // includes an additional column that
                            // indicates whether a timestamp is
                            // complete. For regular "data" upates this
                            // is always `false`.
                            packer.push(Datum::False);
                        }

                        packer.push(Datum::Int64(i64::cast_from(diff)));

                        packer.extend_by_row(&row);

                        packer.finish_and_reuse()
                    })
                    .collect();
                // TODO(benesch): the lack of backpressure here can result in
                // unbounded memory usage.
                let result = self.channel.send(rows);
                if result.is_err() {
                    // TODO(benesch): we should actually drop the sink if the
                    // receiver has gone away. E.g. form a DROP SINK command?
                }
                false
            }
            TailResponse::Complete | TailResponse::Dropped => {
                // TODO: Could perhaps do this earlier, in response to DROP SINK.
                true
            }
        }
    }
/// Process a tail response
///
/// Returns `true` if the sink should be removed.
Process a tail response
Returns `true` if the sink should be removed.
Process a tail response Returns ` true ` if the sink should be removed .
pub(crate) fn process_response(&mut self, response: TailResponse) -> bool {
        let mut packer = Row::default();
        match response {
            TailResponse::Progress(upper) => {
                if self.emit_progress && !upper.is_empty() {
                    assert_eq!(
                        upper.len(),
                        1,
                        "TAIL only supports single-dimensional timestamps"
                    );
                    packer.push(Datum::from(numeric::Numeric::from(*&upper[0])));
                    packer.push(Datum::True);
                    // Fill in the diff column and all table columns with NULL.
                    for _ in 0..(self.object_columns + 1) {
                        packer.push(Datum::Null);
                    }
                    let row = packer.finish_and_reuse();
                    let result = self.channel.send(vec![row]);
                    if result.is_err() {
                        // TODO(benesch): we should actually drop the sink if the
                        // receiver has gone away. E.g. form a DROP SINK command?
                    }
                }
                upper.is_empty()
            }
            TailResponse::Rows(mut rows) => {
                // Sort results by time. We use stable sort here because it will produce deterministic
                // results since the cursor will always produce rows in the same order.
                // TODO: Is sorting necessary?
                rows.sort_by_key(|(_, time, _)| *time);
                let rows = rows
                    .into_iter()
                    .map(|(row, time, diff)| {
                        packer.push(Datum::from(numeric::Numeric::from(time)));
                        if self.emit_progress {
                            // When sinking with PROGRESS, the output
                            // includes an additional column that
                            // indicates whether a timestamp is
                            // complete. For regular "data" upates this
                            // is always `false`.
                            packer.push(Datum::False);
                        }
                        packer.push(Datum::Int64(i64::cast_from(diff)));
                        packer.extend_by_row(&row);
                        packer.finish_and_reuse()
                    })
                    .collect();
                // TODO(benesch): the lack of backpressure here can result in
                // unbounded memory usage.
                let result = self.channel.send(rows);
                if result.is_err() {
                    // TODO(benesch): we should actually drop the sink if the
                    // receiver has gone away. E.g. form a DROP SINK command?
                }
                false
            }
            TailResponse::Complete | TailResponse::Dropped => {
                // TODO: Could perhaps do this earlier, in response to DROP SINK.
                true
            }
        }
    }
pub ( crate ) fn process_response ( & mut self , response : TailResponse ) -> bool { let mut packer = Row :: default ( ) ; match response { TailResponse :: Progress ( upper ) => { if self . emit_progress && ! upper . is_empty ( ) { assert_eq ! ( upper len ( ) 1 " " ) ; packer . push ( Datum :: from ( numeric :: Numeric :: from ( * & upper [ 0 ] ) ) ) ; packer . push ( Datum :: True ) ; // Fill in the diff column and all table columns with NULL. for _ in 0 .. ( self . object_columns + 1 ) { packer . push ( Datum :: Null ) ; } let row = packer . finish_and_reuse ( ) ; let result = self . channel . send ( vec ! [ row ] ) ; if result . is_err ( ) { // TODO(benesch): we should actually drop the sink if the // receiver has gone away. E.g. form a DROP SINK command? } } upper . is_empty ( ) } TailResponse :: Rows ( mut rows ) => { // Sort results by time. We use stable sort here because it will produce deterministic // results since the cursor will always produce rows in the same order. // TODO: Is sorting necessary? rows . sort_by_key ( | ( _ , time , _ ) | * time ) ; let rows = rows . into_iter ( ) . map ( | ( row , time , diff ) | { packer . push ( Datum :: from ( numeric :: Numeric :: from ( time ) ) ) ; if self . emit_progress { // When sinking with PROGRESS, the output // includes an additional column that // indicates whether a timestamp is // complete. For regular "data" upates this // is always `false`. packer . push ( Datum :: False ) ; } packer . push ( Datum :: Int64 ( i64 :: cast_from ( diff ) ) ) ; packer . extend_by_row ( & row ) ; packer . finish_and_reuse ( ) } ) . collect ( ) ; // TODO(benesch): the lack of backpressure here can result in // unbounded memory usage. let result = self . channel . send ( rows ) ; if result . is_err ( ) { // TODO(benesch): we should actually drop the sink if the // receiver has gone away. E.g. form a DROP SINK command? } false } TailResponse :: Complete | TailResponse :: Dropped => { // TODO: Could perhaps do this earlier, in response to DROP SINK. true } } }
Process a tail response
Returns `true` if the sink should be removed.
Process a tail response Returns ` true ` if the sink should be removed .
&mut self
response
TailResponse
&mut self
response
TailResponse