Add LogDog log stream fetcher code.

web/inc/rpc/client.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="../luci-sleep-promise/promise.ts" />
 
 namespace luci {
   /**
@@ skipping 55 matching lines @@
      * @param service the RPC service name to call.
      * @param method the RPC method name to call.
      * @param request optional request body content.
      *
      * @returns A Promise that resolves to the RPC response, or errors with an
      *     error, including HttpError or GrpcError on HTTP/gRPC failure
      *     respectively.
      */
     call<T, R>(service: string, method: string, request?: T): Promise<R> {
       let transientRetry = new RetryIterator(this.transientRetry);
-      let doCall = (): Promise<R> => {
-        // Configure the client for this request.
-        this.pc.service = service;
-        this.pc.method = method;
-        this.pc.request = request;
+      return transientRetry
+          .do(
+              () => {
+                // Configure the client for this request.
+                this.pc.service = service;
+                this.pc.method = method;
+                this.pc.request = request;
 
-        // Execute the configured request.
-        let callPromise: Promise<R> = this.pc.call().completes;
-        return callPromise.then((resp: any) => resp.response)
-            .catch((err: Error) => {
-              // Is this a transient error?
-              if (isTransientError(err)) {
-                let delay = transientRetry.next();
-                if (delay >= 0) {
-                  console.warn(
-                      `Transient error calling ` +
-                          `${service}.${method} with params:`,
-                      request, `:`, err, `; retrying after ${delay}ms.`);
-                  return luci.sleepPromise(delay).then(doCall);
+                // Execute the configured request.
+                return this.pc.call().completes;
+              },
+              (err: Error, delay: number) => {
+                // Is this a transient error?
+                if (!isTransientError(err)) {
+                  throw err;
                 }
-              }
-
-              // Non-transient, throw the error.
-              throw err;
-            });
-      };
-      return doCall();
+                console.warn(
+                    `Transient error calling ` +
+                        `${service}.${method} with params:`,
+                    request, `:`, err, `; retrying after ${delay}ms.`);
+              })
+          .then((resp: any) => resp.response);
     }
   }
 
   /**
    * gRPC Codes
    *
    * Copied from "rpc-code.html" and, more directly,
    * https://github.com/grpc/grpc-go/blob/972dbd2/codes/codes.go#L43
    */
   export enum Code {
@@ skipping 96 matching lines @@
 
   /**
    * RetryIterator configuration class.
    *
    * A user will define the retry parameters using a Retry instance, then create
    * a RetryIterator with them.
    */
   export type Retry = {
     // The number of retries to perform before failing. If undefined, will retry
     // indefinitely.
-    retries: number | undefined;
+    retries?: number;
 
     // The amount of time to delay in between retry attempts, in milliseconds.
     // If undefined or < 0, no delay will be imposed.
     delay: number;
     // The maximum delay to apply, in milliseconds. If > 0 and delay scales past
     // "maxDelay", it will be capped at "maxDelay".
-    maxDelay: number | undefined;
+    maxDelay?: number;
 
     // delayScaling is the multiplier applied to "delay" in between retries. If
     // undefined or <= 1, DEFAULT_DELAY_SCALING will be used.
     delayScaling?: number;
   };
 
+  /** RetryCallback is an optional callback type used in "do". */
+  export type RetryCallback = (err: Error, delay: number) => void;
+
+  /**
+   * Stopped is a sentinel error thrown by RetryIterator when it runs out of
+   * retries.
+   */
+  export const STOPPED = new Error('retry stopped');
+
   /**
    * Generic exponential backoff retry delay generator.
    *
    * A RetryIterator is a specific configured instance of a Retry. Each call to
    * "next()" returns the next delay in the retry sequence.
    */
   export class RetryIterator {
     // Default scaling if no delay is specified.
     static readonly DEAFULT_DELAY_SCALING = 2;
 
@@ skipping 10 matching lines @@
       this.delayScaling = (config.delayScaling || 0);
       if (this.delayScaling < 1) {
         this.delayScaling = RetryIterator.DEAFULT_DELAY_SCALING;
       }
     }
 
     /**
      * @returns the next delay, in milliseconds. If there are no more retries,
      *      returns undefined.
      */
-    next(): number|undefined {
+    next(): number {
       // Apply retries, if they have been enabled.
       if (this.retries !== undefined) {
         if (this.retries <= 0) {
           // No more retries remaining.
-          return undefined;
+          throw STOPPED;
         }
         this.retries--;
       }
 
       let delay = this.delay;
       this.delay *= this.delayScaling;
       if (this.maxDelay > 0 && delay > this.maxDelay) {
         this.delay = delay = this.maxDelay;
       }
       return delay;
     }
+
+    /**
+     * Executes a Promise, retrying if the Promise raises an error.
+     *
+     * "do" iteratively tries to execute a Promise, generated by "gen". If that
+     * Promise raises an error, "do" will retry until it either runs out of
+     * retries, or the Promise does not return an error. Each retry, "do" will
+     * invoke "gen" again to generate a new Promise.
+     *
+     * An optional "onError" callback can be supplied. If it is, it will be
+     * invoked in between each retry. The callback may, itself, throw, in which
+     * case the retry loop will stop. This can be used for reporting and/or
+     * selective retries.
+     *
+     * @param gen Promise generator function for retries.
+     * @param onError optional callback to be invoked in between retries.
+     *
+     * @throws any the error generated by "gen"'s Promise, if out of retries,
+     *     or the error raised by onError if it chooses to throw.
+     */
+    do
+      <T>(gen: () => Promise<T>, onError?: RetryCallback): Promise<T> {
+        let onErr = (err: Error): Promise<T> => {
+          let delay: number;
+          try {
+            delay = this.next();
+          } catch (e) {
+            if (e !== STOPPED) {
+              console.warn('Unexpected error generating next delay:', e);
+            }
+
+            // If we could not generate another retry delay, raise the initial
+            // Promise's error.
+            throw err;
+          }
+
+          if (onError) {
+            // Note: this may throw.
+            onError(err, delay);
+          }
+
+          return luci.sleepPromise(delay).then(() => {
+            return gen().catch(onErr);
+          });
+        };
+
+        return gen().catch(onErr);
+      }
   }
 }