Refactoring FileTransport to use the AsynchronousFileOutputStream.

Very thin wrapper, nothing except swallow+log all errors.

TESTED=presubmit
RELNOTES: None
PiperOrigin-RevId: 188177872

5 files changed, 34 insertions, 133 deletions

diff --git a/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/BUILD b/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/BUILD
index 2e883ff10e..7d8fe5709a 100644
--- a/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/BUILD
+++ b/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/BUILD
@@ -10,6 +10,7 @@ java_library(
     name = "transports",
     srcs = glob(["*.java"]),
     deps = [
+        "//src/main/java/com/google/devtools/build/lib:io",
         "//src/main/java/com/google/devtools/build/lib/buildeventstream",
         "//src/main/java/com/google/devtools/build/lib/buildeventstream/proto:build_event_stream_java_proto",
         "//src/main/java/com/google/devtools/build/lib/vfs",
diff --git a/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/BinaryFormatFileTransport.java b/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/BinaryFormatFileTransport.java
index 70d500dfeb..5af8cd3306 100644
--- a/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/BinaryFormatFileTransport.java
+++ b/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/BinaryFormatFileTransport.java
@@ -19,14 +19,8 @@ import static com.google.common.base.Preconditions.checkNotNull;
 import com.google.devtools.build.lib.buildeventstream.ArtifactGroupNamer;
 import com.google.devtools.build.lib.buildeventstream.BuildEvent;
 import com.google.devtools.build.lib.buildeventstream.BuildEventConverters;
-import com.google.devtools.build.lib.buildeventstream.BuildEventStreamProtos;
 import com.google.devtools.build.lib.buildeventstream.BuildEventTransport;
 import com.google.devtools.build.lib.buildeventstream.PathConverter;
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
-import java.util.concurrent.Future;
-import java.util.logging.Level;
-import java.util.logging.Logger;
 
 /**
  * A simple {@link BuildEventTransport} that writes a varint delimited binary representation of
@@ -34,9 +28,6 @@ import java.util.logging.Logger;
  */
 public final class BinaryFormatFileTransport extends FileTransport {
 
-  private static final Logger logger = Logger.getLogger(BinaryFormatFileTransport.class.getName());
-
-  private static final int MAX_VARINT_BYTES = 9;
   private final PathConverter pathConverter;
 
   BinaryFormatFileTransport(String path, PathConverter pathConverter) {
@@ -48,9 +39,10 @@ public final class BinaryFormatFileTransport extends FileTransport {
   public String name() {
     return this.getClass().getSimpleName();
   }
-  
+
   @Override
   public synchronized void sendBuildEvent(BuildEvent event, final ArtifactGroupNamer namer) {
+    checkNotNull(event);
     BuildEventConverters converters =
         new BuildEventConverters() {
           @Override
@@ -62,19 +54,6 @@ public final class BinaryFormatFileTransport extends FileTransport {
             return namer;
           }
         };
-    checkNotNull(event);
-    BuildEventStreamProtos.BuildEvent protoEvent = event.asStreamProto(converters);
-
-    int maxSerializedSize = MAX_VARINT_BYTES + protoEvent.getSerializedSize();
-    ByteArrayOutputStream out = new ByteArrayOutputStream(maxSerializedSize);
-
-    try {
-      protoEvent.writeDelimitedTo(out);
-      writeData(out.toByteArray());
-    } catch (IOException e) {
-      logger.log(Level.SEVERE, e.getMessage(), e);
-      @SuppressWarnings({"unused", "nullness"})
-      Future<?> possiblyIgnoredError = close();
-    }
+    write(event.asStreamProto(converters));
   }
 }
diff --git a/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/FileTransport.java b/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/FileTransport.java
index 528e6d6d25..6a59e96c12 100644
--- a/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/FileTransport.java
+++ b/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/FileTransport.java
@@ -14,20 +14,14 @@
 
 package com.google.devtools.build.lib.buildeventstream.transports;
 
-import static com.google.common.base.Preconditions.checkNotNull;
-import static com.google.common.base.Preconditions.checkState;
-
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
-import com.google.common.util.concurrent.SettableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
 import com.google.devtools.build.lib.buildeventstream.BuildEventTransport;
+import com.google.devtools.build.lib.util.io.AsynchronousFileOutputStream;
+import com.google.protobuf.Message;
 import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.nio.channels.AsynchronousFileChannel;
-import java.nio.channels.CompletionHandler;
-import java.nio.file.Paths;
-import java.nio.file.StandardOpenOption;
-import java.util.concurrent.Future;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
@@ -39,121 +33,51 @@ import java.util.logging.Logger;
  */
 abstract class FileTransport implements BuildEventTransport {
 
-  /**
-   * We use an {@link AsynchronousFileChannel} to perform non-blocking writes to a file. It get's
-   * tricky when it comes to {@link #close()}, as we may only complete the returned future when all
-   * writes have completed (succeeded or failed). Thus, we use a field {@link #outstandingWrites} to
-   * keep track of the number of writes that have not completed yet. It's simply incremented before
-   * a new write and decremented after a write has completed. When it's {@code 0} it's safe to
-   * complete the close future.
-   */
   private static final Logger logger = Logger.getLogger(FileTransport.class.getName());
-
   @VisibleForTesting
-  final AsynchronousFileChannel ch;
-  private final WriteCompletionHandler completionHandler = new WriteCompletionHandler();
-  // The offset in the file to begin the next write at.
-  private long writeOffset;
-  // Number of writes that haven't completed yet.
-  private long outstandingWrites;
-  // The future returned by close()
-  private SettableFuture<Void> closeFuture;
+  final AsynchronousFileOutputStream out;
 
   FileTransport(String path) {
     try {
-      ch = AsynchronousFileChannel.open(Paths.get(path), StandardOpenOption.CREATE,
-          StandardOpenOption.TRUNCATE_EXISTING, StandardOpenOption.WRITE);
+      out = new AsynchronousFileOutputStream(path);
     } catch (IOException e) {
       throw new RuntimeException(e);
     }
   }
 
-  synchronized void writeData(byte[] data) {
-    checkNotNull(data);
-    if (!ch.isOpen()) {
-      @SuppressWarnings({"unused", "nullness"})
-      Future<?> possiblyIgnoredError = close();
-      return;
-    }
-    if (closing()) {
-      return;
-    }
-
-    outstandingWrites++;
+  // Silent wrappers to AsynchronousFileOutputStream methods.
 
-    ch.write(ByteBuffer.wrap(data), writeOffset, null, completionHandler);
-
-    writeOffset += data.length;
-  }
-
-  @Override
-  public synchronized ListenableFuture<Void> close() {
-    if (closing()) {
-      return closeFuture;
-    }
-    closeFuture = SettableFuture.create();
-
-    if (writesComplete()) {
-      doClose();
+  protected void write(Message m) {
+    try {
+      out.write(m);
+    } catch (Exception e) {
+      logger.log(Level.SEVERE, e.getMessage(), e);
     }
-
-    return closeFuture;
   }
 
-  private void doClose() {
+  protected void write(String s) {
     try {
-      ch.force(true);
-      ch.close();
-    } catch (IOException e) {
+      out.write(s);
+    } catch (Exception e) {
       logger.log(Level.SEVERE, e.getMessage(), e);
-    } finally {
-      closeFuture.set(null);
     }
   }
 
-  @Override
-  @SuppressWarnings("FutureReturnValueIgnored")
-  public void closeNow() {
-    close();
-  }
-
-  private boolean closing() {
-    return closeFuture != null;
-  }
 
-  private boolean writesComplete() {
-    return outstandingWrites == 0;
+  @Override
+  public synchronized ListenableFuture<Void> close() {
+    return Futures.catching(
+        out.closeAsync(),
+        Throwable.class,
+        (t) -> {
+          logger.log(Level.SEVERE, t.getMessage(), t);
+          return null;
+        },
+        MoreExecutors.directExecutor());
   }
 
-  /**
-   * Handler that's notified when a write completes.
-   */
-  private final class WriteCompletionHandler implements CompletionHandler<Integer, Void> {
-
-    @Override
-    public void completed(Integer result, Void attachment) {
-      countWriteAndTryClose();
-    }
-
-    @Override
-    public void failed(Throwable exc, Void attachment) {
-      logger.log(Level.SEVERE, exc.getMessage(), exc);
-      countWriteAndTryClose();
-      // There is no point in trying to continue. Close the transport.
-      @SuppressWarnings({"unused", "nullness"})
-      Future<?> possiblyIgnoredError = close();
-    }
-
-    private void countWriteAndTryClose() {
-      synchronized (FileTransport.this) {
-        checkState(outstandingWrites > 0);
-
-        outstandingWrites--;
-
-        if (closing() && writesComplete()) {
-          doClose();
-        }
-      }
-    }
+  @Override
+  public void closeNow() {
+    out.closeNow();
   }
 }
diff --git a/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/JsonFormatFileTransport.java b/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/JsonFormatFileTransport.java
index eca7d37c7b..3909f2830f 100644
--- a/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/JsonFormatFileTransport.java
+++ b/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/JsonFormatFileTransport.java
@@ -22,7 +22,6 @@ import com.google.devtools.build.lib.buildeventstream.PathConverter;
 import com.google.protobuf.InvalidProtocolBufferException;
 import com.google.protobuf.util.JsonFormat;
 import java.io.IOException;
-import java.nio.charset.StandardCharsets;
 
 /**
  * A simple {@link BuildEventTransport} that writes the JSON representation of the protocol-buffer
@@ -69,6 +68,6 @@ public final class JsonFormatFileTransport extends FileTransport {
       protoJsonRepresentation =
           "{\"id\" : \"unknown\", \"exception\" : \"InvalidProtocolBufferException\"}\n";
     }
-    writeData(protoJsonRepresentation.getBytes(StandardCharsets.UTF_8));
+    write(protoJsonRepresentation);
   }
 }
diff --git a/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/TextFormatFileTransport.java b/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/TextFormatFileTransport.java
index 974dc523d5..1640fd9038 100644
--- a/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/TextFormatFileTransport.java
+++ b/src/main/java/com/google/devtools/build/lib/buildeventstream/transports/TextFormatFileTransport.java
@@ -21,7 +21,6 @@ import com.google.devtools.build.lib.buildeventstream.BuildEventTransport;
 import com.google.devtools.build.lib.buildeventstream.PathConverter;
 import com.google.protobuf.TextFormat;
 import java.io.IOException;
-import java.nio.charset.StandardCharsets;
 
 /**
  * A simple {@link BuildEventTransport} that writes the text representation of the protocol-buffer
@@ -58,7 +57,6 @@ public final class TextFormatFileTransport extends FileTransport {
           }
         };
     String protoTextRepresentation = TextFormat.printToString(event.asStreamProto(converters));
-    String line = "event {\n" + protoTextRepresentation + "}\n\n";
-    writeData(line.getBytes(StandardCharsets.UTF_8));
+    write("event {\n" + protoTextRepresentation + "}\n\n");
   }
 }