Revamp the client/server communication protocol so that it is portable to Windows.

Progress towards #930.

--
MOS_MIGRATED_REVID=117799006

5 files changed, 92 insertions, 69 deletions

diff --git a/src/test/java/com/google/devtools/build/lib/server/RPCServerTest.java b/src/test/java/com/google/devtools/build/lib/server/RPCServerTest.java
index 5d57544062..f81a0384f3 100644
--- a/src/test/java/com/google/devtools/build/lib/server/RPCServerTest.java
+++ b/src/test/java/com/google/devtools/build/lib/server/RPCServerTest.java
@@ -102,27 +102,26 @@ public class RPCServerTest {
     FileSystemUtils.deleteTree(serverDir);
   }
 
-  private void runTestRequest(String request, int ret, String out, String err,
-                             String control) throws Exception {
-    assertEquals(new ServerResponse(control, ret), client.sendRequest(request));
+  private void runTestRequest(String request, int ret, String out, String err) throws Exception {
+    assertEquals(ret, client.sendRequest(request));
     assertEquals(out, outErr.outAsLatin1());
     assertThat(outErr.errAsLatin1()).contains(err);
   }
 
   @Test
   public void testUnknownCommand() throws Exception {
-    runTestRequest("unknown", 2, "", "SERVER ERROR: Unknown command: unknown\n", "");
+    runTestRequest("unknown", 2, "", "SERVER ERROR: Unknown command: unknown\n");
   }
 
   @Test
   public void testEmptyBlazeCommand() throws Exception {
-    runTestRequest("unknown", 2, "", "SERVER ERROR: Unknown command: unknown\n", "");
+    runTestRequest("unknown", 2, "", "SERVER ERROR: Unknown command: unknown\n");
   }
 
   @Test
   public void testWorkspaceDies() throws Exception {
     assertTrue(serverThread.isAlive());
-    runTestRequest("blaze", 42, COMMAND_STDOUT, COMMAND_STDERR, "");
+    runTestRequest("blaze", 42, COMMAND_STDOUT, COMMAND_STDERR);
     Thread.sleep(HEALTH_CHECK_MILLIS * 2);
     assertTrue(serverThread.isAlive());
 
diff --git a/src/test/java/com/google/devtools/build/lib/server/RPCTestingClient.java b/src/test/java/com/google/devtools/build/lib/server/RPCTestingClient.java
index e24e0bb3ac..4a659a14f8 100644
--- a/src/test/java/com/google/devtools/build/lib/server/RPCTestingClient.java
+++ b/src/test/java/com/google/devtools/build/lib/server/RPCTestingClient.java
@@ -43,7 +43,7 @@ public class RPCTestingClient {
     this.outErr = outErr;
   }
 
-  public ServerResponse sendRequest(String command, String... params)
+  public int sendRequest(String command, String... params)
       throws Exception {
     String request = command;
     for (String param : params) {
@@ -52,24 +52,35 @@ public class RPCTestingClient {
     return sendRequest(request);
   }
 
-  public ServerResponse sendRequest(String request) throws Exception {
+  public int sendRequest(String request) throws Exception {
     LocalClientSocket connection = new LocalClientSocket();
     connection.connect(new LocalSocketAddress(socketFile.getPathFile()));
     try {
       OutputStream out = connection.getOutputStream();
-      out.write(request.getBytes(UTF_8));
+      byte[] requestBytes = request.getBytes(UTF_8);
+      byte[] requestLength = new byte[4];
+      requestLength[0] = (byte) (requestBytes.length << 24);
+      requestLength[1] = (byte) ((requestBytes.length << 16) & 0xff);
+      requestLength[2] = (byte) ((requestBytes.length << 8) & 0xff);
+      requestLength[3] = (byte) (requestBytes.length & 0xff);
+      out.write(requestLength);
+      out.write(requestBytes);
       out.flush();
       connection.shutdownOutput();
 
       OutputStream stdout = outErr.getOutputStream();
       OutputStream stderr = outErr.getErrorStream();
       ByteArrayOutputStream control = new ByteArrayOutputStream();
-      StreamDemultiplexer demux = new StreamDemultiplexer((byte) '1',
+      StreamDemultiplexer demux = new StreamDemultiplexer((byte) 1,
           stdout, stderr, control);
       ByteStreams.copy(connection.getInputStream(), demux);
       demux.flush();
 
-      return ServerResponse.parseFrom(control);
+      byte[] controlBytes = control.toByteArray();
+      return (((int) controlBytes[0]) << 24)
+          + (((int) controlBytes[1]) << 16)
+          + (((int) controlBytes[2]) << 8)
+          + ((int) controlBytes[3]);
     } finally {
       connection.close();
     }
diff --git a/src/test/java/com/google/devtools/build/lib/util/io/StreamDemultiplexerTest.java b/src/test/java/com/google/devtools/build/lib/util/io/StreamDemultiplexerTest.java
index 47d693116a..786eb6c42b 100644
--- a/src/test/java/com/google/devtools/build/lib/util/io/StreamDemultiplexerTest.java
+++ b/src/test/java/com/google/devtools/build/lib/util/io/StreamDemultiplexerTest.java
@@ -16,8 +16,6 @@ package com.google.devtools.build.lib.util.io;
 import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
 
-import com.google.devtools.build.lib.util.StringUtilities;
-
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
@@ -25,6 +23,7 @@ import org.junit.runners.JUnit4;
 import java.io.ByteArrayOutputStream;
 import java.io.OutputStream;
 import java.io.UnsupportedEncodingException;
+import java.nio.charset.Charset;
 import java.util.Random;
 
 /**
@@ -37,14 +36,6 @@ public class StreamDemultiplexerTest {
   private ByteArrayOutputStream err = new ByteArrayOutputStream();
   private ByteArrayOutputStream ctl = new ByteArrayOutputStream();
 
-  private byte[] lines(String... lines) {
-    try {
-      return StringUtilities.joinLines(lines).getBytes("ISO-8859-1");
-    } catch (UnsupportedEncodingException e) {
-      throw new AssertionError(e);
-    }
-  }
-
   private String toAnsi(ByteArrayOutputStream stream) {
     try {
       return new String(stream.toByteArray(), "ISO-8859-1");
@@ -63,8 +54,8 @@ public class StreamDemultiplexerTest {
 
   @Test
   public void testHelloWorldOnStandardOut() throws Exception {
-    byte[] multiplexed = lines("@1@", "Hello, world.");
-    try (final StreamDemultiplexer demux = new StreamDemultiplexer((byte) '1', out)) {
+    byte[] multiplexed = chunk(1, "Hello, world.");
+    try (final StreamDemultiplexer demux = new StreamDemultiplexer((byte) 1, out)) {
       demux.write(multiplexed);
     }
     assertEquals("Hello, world.", out.toString("ISO-8859-1"));
@@ -72,8 +63,8 @@ public class StreamDemultiplexerTest {
 
   @Test
   public void testOutErrCtl() throws Exception {
-    byte[] multiplexed = lines("@1@", "out", "@2@", "err", "@3@", "ctl", "");
-    try (final StreamDemultiplexer demux = new StreamDemultiplexer((byte) '1', out, err, ctl)) {
+    byte[] multiplexed = concat(chunk(1, "out"), chunk(2, "err"), chunk(3, "ctl"));
+    try (final StreamDemultiplexer demux = new StreamDemultiplexer((byte) 1, out, err, ctl)) {
       demux.write(multiplexed);
     }
     assertEquals("out", toAnsi(out));
@@ -83,26 +74,16 @@ public class StreamDemultiplexerTest {
 
   @Test
   public void testWithoutLineBreaks() throws Exception {
-    byte[] multiplexed = lines("@1@", "just ", "@1@", "one ", "@1@", "line", "");
-    try (final StreamDemultiplexer demux = new StreamDemultiplexer((byte) '1', out)) {
+    byte[] multiplexed = concat(chunk(1, "just "), chunk(1, "one "), chunk(1, "line"));
+    try (final StreamDemultiplexer demux = new StreamDemultiplexer((byte) 1, out)) {
       demux.write(multiplexed);
     }
     assertEquals("just one line", out.toString("ISO-8859-1"));
   }
 
   @Test
-  public void testLineBreaks() throws Exception {
-    byte[] multiplexed = lines("@1", "two", "@1", "lines", "");
-    try (StreamDemultiplexer demux = new StreamDemultiplexer((byte) '1', out)) {
-      demux.write(multiplexed);
-      demux.flush();
-      assertEquals("two\nlines\n", out.toString("ISO-8859-1"));
-    }
-  }
-
-  @Test
   public void testMultiplexAndBackWithHelloWorld() throws Exception {
-    StreamDemultiplexer demux = new StreamDemultiplexer((byte) '1', out);
+    StreamDemultiplexer demux = new StreamDemultiplexer((byte) 1, out);
     StreamMultiplexer mux = new StreamMultiplexer(demux);
     OutputStream out = mux.createStdout();
     out.write(inAnsi("Hello, world."));
@@ -112,7 +93,7 @@ public class StreamDemultiplexerTest {
 
   @Test
   public void testMultiplexDemultiplexBinaryStress() throws Exception {
-    StreamDemultiplexer demux = new StreamDemultiplexer((byte) '1', out, err, ctl);
+    StreamDemultiplexer demux = new StreamDemultiplexer((byte) 1, out, err, ctl);
     StreamMultiplexer mux = new StreamMultiplexer(demux);
     OutputStream[] muxOuts = {mux.createStdout(), mux.createStderr(), mux.createControl()};
     ByteArrayOutputStream[] expectedOuts =
@@ -132,4 +113,32 @@ public class StreamDemultiplexerTest {
     assertArrayEquals(expectedOuts[1].toByteArray(), err.toByteArray());
     assertArrayEquals(expectedOuts[2].toByteArray(), ctl.toByteArray());
   }
+
+  private static byte[] chunk(int stream, String payload) {
+    byte[] payloadBytes = payload.getBytes(Charset.defaultCharset());
+    byte[] result = new byte[payloadBytes.length + 5];
+
+    System.arraycopy(payloadBytes, 0, result, 5, payloadBytes.length);
+    result[0] = (byte) stream;
+    result[1] = (byte) (payloadBytes.length >> 24);
+    result[2] = (byte) ((payloadBytes.length >> 16) & 0xff);
+    result[3] = (byte) ((payloadBytes.length >> 8) & 0xff);
+    result[4] = (byte) (payloadBytes.length & 0xff);
+    return result;
+  }
+
+  private static byte[] concat(byte[]... chunks) {
+    int length = 0;
+    for (byte[] chunk : chunks) {
+      length += chunk.length;
+    }
+
+    byte[] result = new byte[length];
+    int previousChunks = 0;
+    for (byte[] chunk : chunks) {
+      System.arraycopy(chunk, 0, result, previousChunks, chunk.length);
+      previousChunks += chunk.length;
+    }
+    return result;
+  }
 }
diff --git a/src/test/java/com/google/devtools/build/lib/util/io/StreamMultiplexerParallelStressTest.java b/src/test/java/com/google/devtools/build/lib/util/io/StreamMultiplexerParallelStressTest.java
index 932850fb0c..0c99a16d08 100644
--- a/src/test/java/com/google/devtools/build/lib/util/io/StreamMultiplexerParallelStressTest.java
+++ b/src/test/java/com/google/devtools/build/lib/util/io/StreamMultiplexerParallelStressTest.java
@@ -49,8 +49,7 @@ public class StreamMultiplexerParallelStressTest {
    */
   OutputStream devNull  = ByteStreams.nullOutputStream();
 
-  StreamDemultiplexer demux = new StreamDemultiplexer((byte)'1',
-      devNull, devNull, devNull);
+  StreamDemultiplexer demux = new StreamDemultiplexer((byte) 1, devNull, devNull, devNull);
 
   /**
    * The multiplexer under test.
diff --git a/src/test/java/com/google/devtools/build/lib/util/io/StreamMultiplexerTest.java b/src/test/java/com/google/devtools/build/lib/util/io/StreamMultiplexerTest.java
index 26664a155d..6d3f2c3292 100644
--- a/src/test/java/com/google/devtools/build/lib/util/io/StreamMultiplexerTest.java
+++ b/src/test/java/com/google/devtools/build/lib/util/io/StreamMultiplexerTest.java
@@ -12,7 +12,8 @@
 // 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.devtools.build.lib.util.StringUtilities.joinLines;
+
+import static com.google.common.truth.Truth.assertThat;
 import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
 
@@ -27,6 +28,7 @@ import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.OutputStream;
 import java.io.UnsupportedEncodingException;
+import java.util.Arrays;
 
 /**
  * Test for {@link StreamMultiplexer}.
@@ -61,42 +63,41 @@ public class StreamMultiplexerTest {
     return string.getBytes("ISO-8859-1");
   }
 
-  private static String getLatin(byte[] bytes)
-      throws UnsupportedEncodingException {
-    return new String(bytes, "ISO-8859-1");
-  }
-
   @Test
-  public void testHelloWorldOnStdOut() throws IOException {
+  public void testHelloWorldOnStdOut() throws Exception {
     out.write(getLatin("Hello, world."));
     out.flush();
-    assertEquals(joinLines("@1@", "Hello, world.", ""),
-                 getLatin(multiplexed.toByteArray()));
+    assertMessage(multiplexed.toByteArray(), 0, "Hello, world.");
   }
 
   @Test
   public void testInterleavedStdoutStderrControl() throws Exception {
+    int start = 0;
     out.write(getLatin("Hello, stdout."));
     out.flush();
+    assertMessage(multiplexed.toByteArray(), start, "Hello, stdout.");
+    start = multiplexed.toByteArray().length;
+
     err.write(getLatin("Hello, stderr."));
     err.flush();
+    assertMessage(multiplexed.toByteArray(), start, "Hello, stderr.");
+    start = multiplexed.toByteArray().length;
+
     ctl.write(getLatin("Hello, control."));
     ctl.flush();
+    assertMessage(multiplexed.toByteArray(), start, "Hello, control.");
+    start = multiplexed.toByteArray().length;
+
     out.write(getLatin("... and back!"));
     out.flush();
-    assertEquals(joinLines("@1@", "Hello, stdout.",
-                           "@2@", "Hello, stderr.",
-                           "@3@", "Hello, control.",
-                           "@1@", "... and back!",
-                           ""),
-                 getLatin(multiplexed.toByteArray()));
+    assertMessage(multiplexed.toByteArray(), start, "... and back!");
   }
 
   @Test
   public void testWillNotCommitToUnderlyingStreamUnlessFlushOrNewline()
       throws Exception {
     out.write(getLatin("There are no newline characters in here, so it won't" +
-                       " get written just yet."));
+        " get written just yet."));
     assertArrayEquals(multiplexed.toByteArray(), new byte[0]);
   }
 
@@ -105,16 +106,11 @@ public class StreamMultiplexerTest {
     out.write(getLatin("No newline just yet, so no flushing. "));
     assertArrayEquals(multiplexed.toByteArray(), new byte[0]);
     out.write(getLatin("OK, here we go:\nAnd more to come."));
-
-    String expected = joinLines("@1",
-        "No newline just yet, so no flushing. OK, here we go:", "");
-
-    assertEquals(expected, getLatin(multiplexed.toByteArray()));
-
+    assertMessage(
+        multiplexed.toByteArray(), 0, "No newline just yet, so no flushing. OK, here we go:\n");
+    int firstMessageLength = multiplexed.toByteArray().length;
     out.write((byte) '\n');
-    expected += joinLines("@1", "And more to come.", "");
-
-    assertEquals(expected, getLatin(multiplexed.toByteArray()));
+    assertMessage(multiplexed.toByteArray(), firstMessageLength, "And more to come.\n");
   }
 
   @Test
@@ -122,14 +118,14 @@ public class StreamMultiplexerTest {
     out.write(getLatin("Don't forget to flush!"));
     assertArrayEquals(new byte[0], multiplexed.toByteArray());
     out.flush(); // now the output will appear in multiplexed.
-    assertEquals(joinLines("@1@", "Don't forget to flush!", ""),
-        getLatin(multiplexed.toByteArray()));
+    assertStartsWith(multiplexed.toByteArray(), 1, 0, 0, 0);
+    assertMessage(multiplexed.toByteArray(), 0, "Don't forget to flush!");
   }
 
   @Test
   public void testByteEncoding() throws IOException {
     OutputStream devNull = ByteStreams.nullOutputStream();
-    StreamDemultiplexer demux = new StreamDemultiplexer((byte) '1', devNull);
+    StreamDemultiplexer demux = new StreamDemultiplexer((byte) 1, devNull);
     StreamMultiplexer mux = new StreamMultiplexer(demux);
     OutputStream out = mux.createStdout();
 
@@ -145,4 +141,13 @@ public class StreamMultiplexerTest {
     out.write(10);
   }
 
+  private static void assertStartsWith(byte[] actual, int... expectedPrefix){
+    for (int i = 0; i < expectedPrefix.length; i++) {
+      assertThat(actual[i]).isEqualTo(expectedPrefix[i]);
+    }
+  }
+
+  private static void assertMessage(byte[] actual, int start, String expected) throws Exception {
+    assertThat(Arrays.copyOfRange(actual, start + 5, actual.length)).isEqualTo(getLatin(expected));
+  }
 }