(self, since, forever)
Process changes on all the pillow processors.

            Processes changes serially on serial processors, and in batches on
            batch processors. If there are batch processors, checkpoint is updated
            at the end of the batch, otherwise is updated for every change.
Process changes on all the pillow processors.
Process changes on all the pillow processors .
def process_changes(self, since, forever):
        """
        Process changes on all the pillow processors.

            Processes changes serially on serial processors, and in batches on
            batch processors. If there are batch processors, checkpoint is updated
            at the end of the batch, otherwise is updated for every change.
        """
        context = PillowRuntimeContext(changes_seen=0)
        min_wait_seconds = 30

        def process_offset_chunk(chunk, context):
            if not chunk:
                return
            self._batch_process_with_error_handling(chunk)
            self._update_checkpoint(chunk[-1], context)

        # keep track of chunk for batch processors
        changes_chunk = []
        last_process_time = datetime.utcnow()

        try:
            for change in self.get_change_feed().iter_changes(since=since or None, forever=forever):
                context.changes_seen += 1
                if change:
                    if self.batch_processors:
                        # Queue and process in chunks for both batch
                        #   and serial processors
                        changes_chunk.append(change)
                        chunk_full = len(changes_chunk) == self.processor_chunk_size
                        time_elapsed = (datetime.utcnow() - last_process_time).seconds > min_wait_seconds
                        if chunk_full or time_elapsed:
                            last_process_time = datetime.utcnow()
                            self._batch_process_with_error_handling(changes_chunk)
                            # update checkpoint for just the latest change
                            self._update_checkpoint(changes_chunk[-1], context)
                            # reset for next chunk
                            changes_chunk = []
                    else:
                        # process all changes one by one
                        processing_time = self.process_with_error_handling(change)
                        self._record_change_in_datadog(change, processing_time)
                        self._update_checkpoint(change, context)
                else:
                    self._update_checkpoint(None, None)
            process_offset_chunk(changes_chunk, context)
        except PillowtopCheckpointReset:
            process_offset_chunk(changes_chunk, context)
            self.process_changes(since=self.get_last_checkpoint_sequence(), forever=forever)
        if forever:
            if context.changes_seen and change:
                self._update_checkpoint(change, context)
def process_changes ( self , since , forever ) : context = PillowRuntimeContext ( changes_seen = 0 ) min_wait_seconds = 30 def process_offset_chunk ( chunk , context ) : if not chunk : return self . _batch_process_with_error_handling ( chunk ) self . _update_checkpoint ( chunk [ - 1 ] , context ) # keep track of chunk for batch processors changes_chunk = [ ] last_process_time = datetime . utcnow ( ) try : for change in self . get_change_feed ( ) . iter_changes ( since = since or None , forever = forever ) : context . changes_seen += 1 if change : if self . batch_processors : # Queue and process in chunks for both batch #   and serial processors changes_chunk . append ( change ) chunk_full = len ( changes_chunk ) == self . processor_chunk_size time_elapsed = ( datetime . utcnow ( ) - last_process_time ) . seconds > min_wait_seconds if chunk_full or time_elapsed : last_process_time = datetime . utcnow ( ) self . _batch_process_with_error_handling ( changes_chunk ) # update checkpoint for just the latest change self . _update_checkpoint ( changes_chunk [ - 1 ] , context ) # reset for next chunk changes_chunk = [ ] else : # process all changes one by one processing_time = self . process_with_error_handling ( change ) self . _record_change_in_datadog ( change , processing_time ) self . _update_checkpoint ( change , context ) else : self . _update_checkpoint ( None , None ) process_offset_chunk ( changes_chunk , context ) except PillowtopCheckpointReset : process_offset_chunk ( changes_chunk , context ) self . process_changes ( since = self . get_last_checkpoint_sequence ( ) , forever = forever ) if forever : if context . changes_seen and change : self . _update_checkpoint ( change , context )