anchor.diff:  }
   
   Future<String> _protectedHandle(String msg) async {
  -  var processor = new LogItemProcessor();
  -
     try {
       var inputData = JSON.decode(msg);
       String bucketName = inputData[0];
       String objectPath = inputData[1];
       var logItems = [];
  +    var errItems = [];
       int failureCount = 0;
       int lines = 0;
  +
  +    LogItemProcessor proc = new LogItemProcessor(completionExtraction);
   
       Stream dataStream = await getDataFromCloud(bucketName, objectPath);
       await for (String s in dataStream) {
         lines++;
         try {
  -        String result = processor.processLine(s);
  -        if (result != null) logItems.add(result);
  +        proc.addRawLine(s);
  +
  +        String nextMessage;
  +        while (proc.hasMoreMessages) {
  +          nextMessage = proc.readNextMessage();
  +          if (nextMessage != null) logItems.add(nextMessage);
  +        }
         } catch (e, st) {
  +        log.info("Message proc erred. $e \n $st \n");
  +        var rawErringLine = s.toString();
  +        if (rawErringLine.length > 300) rawErringLine = '${rawErringLine.substring(0, 300)} ...';
  +        errItems
  +            .add({"lastBlock": false, "rawDataLine": rawErringLine, "exception": "$e", "stackTrace": "$st"});
           failureCount++;
         }
       }
   
  +    if (client != null) client.close();
  +
       try {
  -      String result = processor.close();
  -      if (result != null) logItems.add(result);
  +      proc.close();
  +
  +      String nextMessage;
  +      while (proc.hasMoreMessages) {
  +        nextMessage = proc.readNextMessage();
  +        if (nextMessage != null) logItems.add(nextMessage);
  +      }
       } catch (e, st) {
  +      log.info("Message proc erred. $e \n $st \n");
  +      errItems
  +          .add({"lastBlock": true, "exception": "$e", "stackTrace": "$st"});
         failureCount++;
  +    }
  +
  +    // Finish the extraction process
  +    var finalResults = completionExtractionFinished();
  +    if (finalResults is String) {
  +      logItems.add(finalResults);
  +    } else if (finalResults is List<String>) {
  +      logItems.addAll(finalResults);
       }
   
       return JSON.encode({
         "result": logItems,
         "failureCount": failureCount,
  -      "linesProcessed": lines
  +      "errItems": errItems,
  +      "linesProcessed": lines,
  +      "input": "gs://$bucketName/$objectPath"
       });
     } catch (e, st) {
  -    log.info("Execution erred. $e \n $st \n");
  +    log.info("Message proc erred. $e \n $st \n");
       log.debug("Input data: $msg");
       return JSON.encode({"error": "${e}", "stackTrace": "${st}"});
     }
  
positive.code: import 'package:sintr_common/logging_utils.dart' as log;
  import 'package:sintr_worker_lib/instrumentation_lib.dart';
  
  const projectName = "liftoff-dev";
  
  Future main(List<String> args, SendPort sendPort) async {
    ReceivePort receivePort = new ReceivePort();
  
negative.code:       cryptoTokensLocation:
            "${config.userHomePath}/Communications/CryptoTokens");
  
    client = await auth.getAuthedClient();
    var sourceStorage = new storage.Storage(client, projectName);
    Stream<List<int>> rawStream =
        sourceStorage.bucket(bucketName).read(objectPath);
  