remote/http: support refresh of oauth2 tokens in the remote cache.

Closes #4622.

PiperOrigin-RevId: 188595430

src/main/java/com/google/devtools/build/lib/remote/SimpleBlobStoreActionCache.java

@@ -165,7 +165,9 @@ public Digest uploadBlob(byte[] blob) throws IOException, InterruptedException {
   }
 
   private Digest uploadBlob(byte[] blob, Digest digest) throws IOException, InterruptedException {
-    return uploadStream(digest, new ByteArrayInputStream(blob));
+    try (InputStream in = new ByteArrayInputStream(blob)) {
+      return uploadStream(digest, in);
+    }
   }
 
   public Digest uploadStream(Digest digest, InputStream in)

src/main/java/com/google/devtools/build/lib/remote/blobstore/SimpleBlobStore.java

@@ -32,21 +32,27 @@ public interface SimpleBlobStore {
   /**
    * Fetches the BLOB associated with the {@code key} from the CAS and writes it to {@code out}.
    *
+   * <p>The caller is responsible to close {@code out}.
+   *
    * @return {@code true} if the {@code key} was found. {@code false} otherwise.
    */
   boolean get(String key, OutputStream out) throws IOException, InterruptedException;
 
   /**
-   * Fetches the BLOB associated with the {@code key} from the Action Cache and writes it to
-   * {@code out}.
+   * Fetches the BLOB associated with the {@code key} from the Action Cache and writes it to {@code
+   * out}.
+   *
+   * <p>The caller is responsible to close {@code out}.
    *
    * @return {@code true} if the {@code key} was found. {@code false} otherwise.
    */
-  boolean getActionResult(String actionKey, OutputStream out) throws IOException,
-      InterruptedException;
+  boolean getActionResult(String actionKey, OutputStream out)
+      throws IOException, InterruptedException;
 
   /**
    * Uploads a BLOB (as {@code in}) with length {@code length} indexed by {@code key} to the CAS.
+   *
+   * <p>The caller is responsible to close {@code in}.
    */
   void put(String key, long length, InputStream in) throws IOException, InterruptedException;
 

src/main/java/com/google/devtools/build/lib/remote/blobstore/http/AbstractHttpHandler.java

@@ -26,6 +26,7 @@
 import java.io.IOException;
 import java.net.SocketAddress;
 import java.net.URI;
+import java.nio.channels.ClosedChannelException;
 import java.util.List;
 import java.util.Map;
 
@@ -93,56 +94,61 @@ protected String constructHost(URI uri) {
   }
 
   @Override
-  public void exceptionCaught(ChannelHandlerContext channelHandlerContext, Throwable throwable)
-      throws Exception {
+  public void exceptionCaught(ChannelHandlerContext channelHandlerContext, Throwable throwable) {
     failAndResetUserPromise(throwable);
   }
 
   @SuppressWarnings("FutureReturnValueIgnored")
   @Override
-  public void bind(ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise)
-      throws Exception {
+  public void bind(ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise) {
     ctx.bind(localAddress, promise);
   }
 
-  @SuppressWarnings("FutureReturnValueIgnored") 
+  @SuppressWarnings("FutureReturnValueIgnored")
   @Override
   public void connect(
       ChannelHandlerContext ctx,
       SocketAddress remoteAddress,
       SocketAddress localAddress,
-      ChannelPromise promise)
-      throws Exception {
+      ChannelPromise promise) {
     ctx.connect(remoteAddress, localAddress, promise);
   }
 
   @SuppressWarnings("FutureReturnValueIgnored")
   @Override
-  public void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {   
+  public void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) {
+    failAndResetUserPromise(new ClosedChannelException());
     ctx.disconnect(promise);
   }
 
   @SuppressWarnings("FutureReturnValueIgnored")
   @Override
-  public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
+  public void close(ChannelHandlerContext ctx, ChannelPromise promise) {
+    failAndResetUserPromise(new ClosedChannelException());
     ctx.close(promise);
   }
 
   @SuppressWarnings("FutureReturnValueIgnored")
   @Override
-  public void deregister(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
+  public void deregister(ChannelHandlerContext ctx, ChannelPromise promise) {
     ctx.deregister(promise);
   }
 
   @SuppressWarnings("FutureReturnValueIgnored")
   @Override
-  public void read(ChannelHandlerContext ctx) throws Exception { 
+  public void read(ChannelHandlerContext ctx) {
     ctx.read();
   }
 
-  @SuppressWarnings("FutureReturnValueIgnored") 
+  @SuppressWarnings("FutureReturnValueIgnored")
   @Override
-  public void flush(ChannelHandlerContext ctx) throws Exception {  
+  public void flush(ChannelHandlerContext ctx) {
     ctx.flush();
   }
+
+  @Override
+  public void channelInactive(ChannelHandlerContext channelHandlerContext) throws Exception {
+    failAndResetUserPromise(new ClosedChannelException());
+    super.channelInactive(channelHandlerContext);
+  }
 }