Adding async proto or text logging utility class.

WANT_LGTM=buchgr
TESTED=unit tests, 500 runs per test
RELNOTES: None
PiperOrigin-RevId: 188093043

2 files changed, 369 insertions, 0 deletions

diff --git a/src/test/java/com/google/devtools/build/lib/BUILD b/src/test/java/com/google/devtools/build/lib/BUILD
index c7d819a0ee..99d4335954 100644
--- a/src/test/java/com/google/devtools/build/lib/BUILD
+++ b/src/test/java/com/google/devtools/build/lib/BUILD
@@ -314,6 +314,8 @@ java_test(
         ":testutil",
         "//src/main/java/com/google/devtools/build/lib:io",
         "//src/main/java/com/google/devtools/build/lib:util",
+        "//src/main/protobuf:bazel_flags_java_proto",
+        "//third_party:mockito",
     ],
 )
 
diff --git a/src/test/java/com/google/devtools/build/lib/util/io/AsynchronousFileOutputStreamTest.java b/src/test/java/com/google/devtools/build/lib/util/io/AsynchronousFileOutputStreamTest.java
new file mode 100644
index 0000000000..39c4d6bd76
--- /dev/null
+++ b/src/test/java/com/google/devtools/build/lib/util/io/AsynchronousFileOutputStreamTest.java
@@ -0,0 +1,367 @@
+// Copyright 2018 The Bazel Authors. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package com.google.devtools.build.lib.util.io;
+
+import static com.google.common.truth.Truth.assertThat;
+import static org.junit.Assert.fail;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.when;
+
+import com.google.common.io.ByteStreams;
+import com.google.devtools.build.lib.runtime.commands.proto.BazelFlagsProto.FlagInfo;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.channels.AsynchronousFileChannel;
+import java.nio.channels.CompletionHandler;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
+import java.util.Random;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ThreadLocalRandom;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.MockitoAnnotations;
+
+/** Tests {@link AsynchronousFileOutputStream}. */
+@RunWith(JUnit4.class)
+public class AsynchronousFileOutputStreamTest {
+  @Rule public TemporaryFolder tmp = new TemporaryFolder();
+  @Mock AsynchronousFileChannel mockChannel;
+  Random random = ThreadLocalRandom.current();
+  static final char[] RAND_CHARS = "abcdefghijklmnopqrstuvwxzy0123456789-".toCharArray();
+  static final int RAND_STRING_LENGTH = 10;
+
+  @Before
+  public void initMocks() {
+    MockitoAnnotations.initMocks(this);
+  }
+
+  @After
+  public void validateMocks() {
+    Mockito.validateMockitoUsage();
+  }
+
+  private FlagInfo generateRandomMessage() {
+    FlagInfo.Builder b = FlagInfo.newBuilder();
+    b.setName(generateRandomString() + "a");  // Name is required, cannot be empty.
+    b.setHasNegativeFlag(random.nextBoolean());
+    b.setDocumentation(generateRandomString());
+    int commandsSize = random.nextInt(5);
+    for (int i = 0; i < commandsSize; ++i) {
+      b.addCommands(generateRandomString());
+    }
+    return b.build();
+  }
+
+  private String generateRandomString() {
+    int len = random.nextInt(RAND_STRING_LENGTH + 1);
+    char[] data = new char[len];
+    for (int i = 0; i < len; ++i) {
+      data[i] = RAND_CHARS[random.nextInt(RAND_CHARS.length)];
+    }
+    return new String(data);
+  }
+
+  @Test
+  public void testConcurrentWrites() throws Exception {
+    Path logPath = tmp.newFile().toPath();
+    AsynchronousFileOutputStream out = new AsynchronousFileOutputStream(logPath.toString());
+    Thread[] writers = new Thread[10];
+    final CountDownLatch start = new CountDownLatch(writers.length);
+    for (int i = 0; i < writers.length; ++i) {
+      String name = "Thread # " + i;
+      Thread thread = new Thread() {
+        @Override
+        public void run() {
+          try {
+            start.countDown();
+            start.await();
+          } catch (InterruptedException e) {
+            return;
+          }
+          for (int j = 0; j < 10; ++j) {
+            out.write(name + " time # " + j + "\n");
+          }
+        }
+      };
+      writers[i] = thread;
+      thread.start();
+    }
+    for (int i = 0; i < writers.length; ++i) {
+      writers[i].join();
+    }
+    out.close();
+    String contents =
+        new String(ByteStreams.toByteArray(Files.newInputStream(logPath)), StandardCharsets.UTF_8);
+    for (int i = 0; i < writers.length; ++i) {
+      for (int j = 0; j < 10; ++j) {
+        assertThat(contents).contains("Thread # " + i + " time # " + j + "\n");
+      }
+    }
+  }
+
+  @Test
+  public void testConcurrentProtoWrites() throws Exception {
+    Path logPath = tmp.newFile().toPath();
+    AsynchronousFileOutputStream out = new AsynchronousFileOutputStream(logPath.toString());
+    ArrayList<FlagInfo> messages = new ArrayList<>();
+    for (int i = 0; i < 100; ++i) {
+      messages.add(generateRandomMessage());
+    }
+    Thread[] writers = new Thread[messages.size() / 10];
+    final CountDownLatch start = new CountDownLatch(writers.length);
+    for (int i = 0; i < writers.length; ++i) {
+      int startIndex = i * 10;
+      Thread thread = new Thread() {
+        @Override
+        public void run() {
+          try {
+            start.countDown();
+            start.await();
+          } catch (InterruptedException e) {
+            return;
+          }
+          for (int j = startIndex; j < startIndex + 10; ++j) {
+            out.write(messages.get(j));
+          }
+        }
+      };
+      writers[i] = thread;
+      thread.start();
+    }
+    for (int i = 0; i < writers.length; ++i) {
+      writers[i].join();
+    }
+    out.close();
+    ArrayList<FlagInfo> readMessages = new ArrayList<>();
+    try (InputStream in = Files.newInputStream(logPath)) {
+      for (int i = 0; i < messages.size(); ++i) {
+        readMessages.add(FlagInfo.parseDelimitedFrom(in));
+      }
+    }
+    assertThat(readMessages).containsExactlyElementsIn(messages);
+  }
+
+  @Test
+  public void testFailedClosePropagatesIOException() throws Exception {
+    AsynchronousFileOutputStream out = new AsynchronousFileOutputStream(mockChannel);
+    when(mockChannel.isOpen()).thenReturn(true);
+    IOException ex = new IOException("foo");
+    Mockito.doThrow(ex).when(mockChannel).close();
+    Mockito.doAnswer(
+            invocationOnMock -> {
+              @SuppressWarnings("unchecked")
+              CompletionHandler<Integer, Void> handler =
+                  (CompletionHandler<Integer, Void>) invocationOnMock.getArguments()[3];
+              handler.completed(0, null); // We ignore the arguments.
+              return null;
+            })
+        .when(mockChannel)
+        .write(
+            any(ByteBuffer.class),
+            any(Integer.class),
+            eq(null),
+            Mockito.<CompletionHandler<Integer, Void>>anyObject());
+    out.write("bla");
+
+    try {
+      out.close();
+      fail("Expected an IOException");
+    } catch (IOException expected) {
+      assertThat(expected).hasMessageThat().isEqualTo("foo");
+    }
+  }
+
+  @Test
+  public void testFailedClosePropagatesUncheckedException() throws Exception {
+    AsynchronousFileOutputStream out = new AsynchronousFileOutputStream(mockChannel);
+    when(mockChannel.isOpen()).thenReturn(true);
+    RuntimeException ex = new RuntimeException("foo");
+    Mockito.doThrow(ex).when(mockChannel).close();
+    Mockito.doAnswer(
+            invocationOnMock -> {
+              @SuppressWarnings("unchecked")
+              CompletionHandler<Integer, Void> handler =
+                  (CompletionHandler<Integer, Void>) invocationOnMock.getArguments()[3];
+              handler.completed(0, null); // We ignore the arguments.
+              return null;
+            })
+        .when(mockChannel)
+        .write(
+            any(ByteBuffer.class),
+            any(Integer.class),
+            eq(null),
+            Mockito.<CompletionHandler<Integer, Void>>anyObject());
+    out.write("bla");
+
+    try {
+      out.close();
+      fail("Expected a RuntimeException");
+    } catch (RuntimeException expected) {
+      assertThat(expected).hasMessageThat().isEqualTo("foo");
+    }
+  }
+
+  @Test
+  public void testFailedForcePropagatesIOException() throws Exception {
+    AsynchronousFileOutputStream out = new AsynchronousFileOutputStream(mockChannel);
+    when(mockChannel.isOpen()).thenReturn(true);
+    IOException ex = new IOException("foo");
+    Mockito.doThrow(ex).when(mockChannel).force(eq(true));
+    Mockito.doAnswer(
+            invocationOnMock -> {
+              @SuppressWarnings("unchecked")
+              CompletionHandler<Integer, Void> handler =
+                  (CompletionHandler<Integer, Void>) invocationOnMock.getArguments()[3];
+              handler.completed(0, null); // We ignore the arguments.
+              return null;
+            })
+        .when(mockChannel)
+        .write(
+            any(ByteBuffer.class),
+            any(Integer.class),
+            eq(null),
+            Mockito.<CompletionHandler<Integer, Void>>anyObject());
+    out.write("bla");
+
+    try {
+      out.close();
+      fail("Expected an IOException");
+    } catch (IOException expected) {
+      assertThat(expected).hasMessageThat().isEqualTo("foo");
+    }
+  }
+
+  @Test
+  public void testFailedForcePropagatesUncheckedException() throws Exception {
+    AsynchronousFileOutputStream out = new AsynchronousFileOutputStream(mockChannel);
+    when(mockChannel.isOpen()).thenReturn(true);
+    RuntimeException ex = new RuntimeException("foo");
+    Mockito.doThrow(ex).when(mockChannel).force(eq(true));
+    Mockito.doAnswer(
+            invocationOnMock -> {
+              @SuppressWarnings("unchecked")
+              CompletionHandler<Integer, Void> handler =
+                  (CompletionHandler<Integer, Void>) invocationOnMock.getArguments()[3];
+              handler.completed(0, null); // We ignore the arguments.
+              return null;
+            })
+        .when(mockChannel)
+        .write(
+            any(ByteBuffer.class),
+            any(Integer.class),
+            eq(null),
+            Mockito.<CompletionHandler<Integer, Void>>anyObject());
+    out.write("bla");
+
+    try {
+      out.close();
+      fail("Expected a RuntimeException");
+    } catch (RuntimeException expected) {
+      assertThat(expected).hasMessageThat().isEqualTo("foo");
+    }
+  }
+
+  @Test
+  public void testFailedWritePropagatesIOException() throws Exception {
+    AsynchronousFileOutputStream out = new AsynchronousFileOutputStream(mockChannel);
+    when(mockChannel.isOpen()).thenReturn(true);
+    IOException ex = new IOException("foo");
+    Mockito.doAnswer(
+            invocationOnMock -> {
+              @SuppressWarnings("unchecked")
+              CompletionHandler<Integer, Void> handler =
+                  (CompletionHandler<Integer, Void>) invocationOnMock.getArguments()[3];
+              handler.failed(ex, null);
+              return null;
+            })
+        .when(mockChannel)
+        .write(
+            any(ByteBuffer.class),
+            any(Integer.class),
+            eq(null),
+            Mockito.<CompletionHandler<Integer, Void>>anyObject());
+    out.write("bla");
+    out.write("blo");
+
+    try {
+      out.close();
+      fail("Expected an IOException");
+    } catch (IOException expected) {
+      assertThat(expected).hasMessageThat().isEqualTo("foo");
+    }
+  }
+
+  @Test
+  public void testFailedWritePropagatesUncheckedException() throws Exception {
+    AsynchronousFileOutputStream out = new AsynchronousFileOutputStream(mockChannel);
+    when(mockChannel.isOpen()).thenReturn(true);
+    RuntimeException ex = new RuntimeException("foo");
+    Mockito.doAnswer(
+            invocationOnMock -> {
+              @SuppressWarnings("unchecked")
+              CompletionHandler<Integer, Void> handler =
+                  (CompletionHandler<Integer, Void>) invocationOnMock.getArguments()[3];
+              handler.failed(ex, null);
+              return null;
+            })
+        .when(mockChannel)
+        .write(
+            any(ByteBuffer.class),
+            any(Integer.class),
+            eq(null),
+            Mockito.<CompletionHandler<Integer, Void>>anyObject());
+    out.write("bla");
+    out.write("blo");
+
+    try {
+      out.close();
+      fail("Expected a RuntimeException");
+    } catch (RuntimeException expected) {
+      assertThat(expected).hasMessageThat().isEqualTo("foo");
+    }
+  }
+
+  @Test
+  public void testWriteAfterCloseThrowsException() throws Exception {
+    Path logPath = tmp.newFile().toPath();
+    AsynchronousFileChannel ch = AsynchronousFileChannel.open(
+            logPath,
+            StandardOpenOption.WRITE,
+            StandardOpenOption.CREATE,
+            StandardOpenOption.TRUNCATE_EXISTING);
+    AsynchronousFileOutputStream out = new AsynchronousFileOutputStream(ch);
+    out.write("bla");
+    ch.close();
+
+    try {
+      out.write("blo");
+      fail("Expected an IllegalStateException");
+    } catch (IllegalStateException expected) {
+      // Expected.
+    }
+  }
+}