Enable sequential prefix index logs to be loaded.

web/inc/logdog-stream-view/viewer.ts

 /*
   Copyright 2016 The LUCI Authors. All rights reserved.
   Use of this source code is governed under the Apache License, Version 2.0
   that can be found in the LICENSE file.
 */
 
 ///<reference path="../logdog-stream/logdog.ts" />
 ///<reference path="../luci-operation/operation.ts" />
 ///<reference path="../luci-sleep-promise/promise.ts" />
 ///<reference path="../rpc/client.ts" />
@@ skipping 1005 matching lines @@
           this.setActiveFetch(this.fetcher.get(op, this.nextBottomIndex, opts));
       let logs = await f.p;
       if (logs && logs.length) {
         this.nextBottomIndex = (logs[logs.length - 1].streamIndex + 1);
       }
       return logs;
     }
   }
 
   /**
+   * LogSorter is an interface that used by AggregateLogStream to extract sorted
+   * logs from a set of BufferedLogs.
+   *
+   * It is used to compare two log entries to determine their relative order.
+   */
+  type LogSorter = {
+    /** Returns true if "a" comes before "b".  */
+    before: (a: LogDog.LogEntry, b: LogDog.LogEntry) => number;
+
+    /**
+     * If implemented, returns an implicit next log in the buffer set.
+     *
+     * This is useful if the next log can be determined from the current
+     * buffered data, even if it is partial or incomplete.
+     */
+    implicitNext?: (prev: LogDog.LogEntry, buffers: BufferedLogs[]) =>
+                       LogDog.LogEntry | null;
+  };
+
+  const prefixIndexLogSorter: LogSorter = {
+    before: (a: LogDog.LogEntry, b: LogDog.LogEntry) => {
+      return (a.prefixIndex - b.prefixIndex);
+    },
+
+    implicitNext: (prev: LogDog.LogEntry, buffers: BufferedLogs[]) => {
+      let nextPrefixIndex = (prev.prefixIndex + 1);
+      for (let buf of buffers) {
+        let le = buf.peek();
+        if (le && le.prefixIndex === nextPrefixIndex) {
+          return buf.next();
+        }
+      }
+      return null;
+    },
+  };
+
+  const timestampLogSorter: LogSorter = {
+    before: (a: LogDog.LogEntry, b: LogDog.LogEntry) => {
+      if (a.timestamp) {
+        if (b.timestamp) {
+          return a.timestamp.getTime() - b.timestamp.getTime();
+        }
+        return 1;
+      }
+      if (b.timestamp) {
+        return -1;
+      }
+      return 0;
+    },
+
+    // No implicit "next" with timestamp-based logs, since the next log in
+    // an empty buffer may actually be the next contiguous log.
+    implicitNext: undefined,
+  };
+
+  /**
    * An aggregate log stream. It presents a single-stream view, but is really
    * composed of several log streams interleaved based on their prefix indices
    * (if they share a prefix) or timestamps (if they don't).
    *
    * At least one log entry from each stream must be buffered before any log
    * entries can be yielded, since we don't know what ordering to apply
    * otherwise.  To make this fast, we will make the first request for each
    * stream small so it finishes quickly and we can start rendering. Subsequent
    * entries will be larger for efficiency.
    *
    * @param {LogStream} streams the composite streams.
    */
   class AggregateLogStream implements LogProvider {
     private streams: AggregateLogStream.Entry[];
     private active: AggregateLogStream.Entry[];
     private currentNextPromise: Promise<BufferedLogs[]>|null;
-    private compareLogs: (a: LogDog.LogEntry, b: LogDog.LogEntry) => number;
+    private readonly logSorter: LogSorter;
 
     private streamStatusCallback: StreamStatusCallback;
 
     constructor(streams: LogStream[]) {
       // Input streams, ordered by input order.
       this.streams = streams.map<AggregateLogStream.Entry>((ls, i) => {
         ls.setStreamStatusCallback((st: LogStreamStatus[]) => {
           if (st) {
             this.streams[i].status = st[0];
             this.statusChanged();
@@ skipping 14 matching lines @@
       // timestamp.
       let template: LogDog.StreamPath;
       let sharedPrefix = this.streams.every((entry) => {
         if (!template) {
           template = entry.ls.stream;
           return true;
         }
         return template.samePrefixAs(entry.ls.stream);
       });
 
-      this.compareLogs = ((sharedPrefix) ? (a, b) => {
-        return (a.prefixIndex - b.prefixIndex);
-      } : (a, b) => {
-        if (a.timestamp) {
-          if (b.timestamp) {
-            return a.timestamp.getTime() - b.timestamp.getTime();
-          }
-          return 1;
-        } else if (b.timestamp) {
-          return -1;
-        } else {
-          return 0;
-        }
-      });
+      if (sharedPrefix) {
+        this.logSorter = prefixIndexLogSorter;
+      } else {
+        this.logSorter = timestampLogSorter;
+      }
     }
 
     split(): SplitLogProvider|null {
       return null;
     }
     fetchedEndOfStream(): boolean {
       return (!this.active.length);
     }
 
     setStreamStatusCallback(cb: StreamStatusCallback) {
@@ skipping 115 matching lines @@
     private _aggregateBuffers(buffers: BufferedLogs[]): BufferedLogs {
       switch (buffers.length) {
         case 0:
           // No buffers, so no logs.
           return new BufferedLogs(null);
         case 1:
           // As a special case, if we only have one buffer, and we assume that
           // its entries are sorted, then that buffer is a return value.
           return new BufferedLogs(buffers[0].getAll());
         default:
-          // Nothing to do.
           break;
       }
 
       // Preload our peek array.
       let peek = new Array<LogDog.LogEntry>(buffers.length);
       peek.length = 0;
       for (let buf of buffers) {
         let le = buf.peek();
         if (!le) {
           // One of our input buffers had no log entries.
           return new BufferedLogs(null);
         }
         peek.push(le);
       }
 
       // Assemble our aggregate buffer array.
+      //
+      // As we add log entries, latestAdded will be updated to point to the most
+      // recently added LogEntry.
       let entries: LogDog.LogEntry[] = [];
+      let latestAdded: LogDog.LogEntry|null = null;
       while (true) {
         // Choose the next stream.
         let earliest = 0;
         for (let i = 1; i < buffers.length; i++) {
-          if (this.compareLogs(peek[i], peek[earliest]) < 0) {
+          if (this.logSorter.before(peek[i], peek[earliest])) {
             earliest = i;
           }
         }
 
         // Get the next log from the earliest stream.
         let next = buffers[earliest].next();
         if (next) {
-          entries.push(next);
+          latestAdded = next;
+          entries.push(latestAdded);
         }
 
         // Repopulate that buffer's "peek" value. If the buffer has no more
-        // entries, then we're done.
+        // entries, then we're done this round.
         next = buffers[earliest].peek();
         if (!next) {
-          return new BufferedLogs(entries);
+          break;
         }
         peek[earliest] = next;
       }
+
+      // One or more of our buffers is exhausted. If we have the ability to load
+      // implicit next logs, try and extract more using that.
+      if (latestAdded && this.logSorter.implicitNext) {
+        while (true) {
+          latestAdded = this.logSorter.implicitNext(latestAdded, buffers);
+          if (!latestAdded) {
+            break;
+          }
+          entries.push(latestAdded);
+        }
+      }
+      return new BufferedLogs(entries);
     }
   }
 
   /** Internal namespace for AggregateLogStream types. */
   namespace AggregateLogStream {
     /** Entry is an entry for a single log stream and its buffered logs. */
     export class Entry {
       buffer = new BufferedLogs(null);
       status: LogStreamStatus;
       lastError: Error|null;
@@ skipping 41 matching lines @@
 
     /**
      * Peek returns the next log in the buffer without modifying the buffer. If
      * there are no logs in the buffer, peek will return null.
      */
     peek(): LogDog.LogEntry|null {
       return (this.logs) ? (this.logs[this.index]) : (null);
     }
 
     /**
+     * Returns a copy of the remaining logs in the buffer.
+     * If there are no logs, an empty array will be returned.
+     */
+    peekAll(): LogDog.LogEntry[] {
+      return (this.logs || []).slice(0);
+    }
+
+    /**
      * GetAll returns all logs in the buffer. Afterwards, the buffer will be
      * empty.
      */
     getAll(): LogDog.LogEntry[] {
       // Pop all logs.
       let logs = this.logs;
       this.logs = null;
       return (logs || []);
     }
 
     /**
      * Next fetches the next log in the buffer, removing it from the buffer. If
      * no more logs are available, it will return null.
      */
     next(): LogDog.LogEntry|null {
       if (!(this.logs && this.logs.length)) {
         return null;
       }
 
       // Get the next log and increment our index.
       let log = this.logs[this.index++];
       if (this.index >= this.logs.length) {
         this.logs = null;
       }
       return log;
     }
   }
 }