Use DataInputStream and DataOutputStream instead of manual reading and writing

scm-plugins/scm-hg-plugin/src/main/java/sonia/scm/repository/hooks/Sockets.java

@@ -28,7 +28,8 @@ import com.fasterxml.jackson.databind.ObjectMapper;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.io.EOFException;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
@@ -47,53 +48,34 @@ class Sockets {
   static void send(OutputStream out, Object object) throws IOException {
     byte[] bytes = objectMapper.writeValueAsBytes(object);
     LOG.trace("send message length of {} to socket", bytes.length);
-    writeInt(out, bytes.length);
+
+    DataOutputStream dataOutputStream = new DataOutputStream(out);
+    dataOutputStream.writeInt(bytes.length);
+
     LOG.trace("send message to socket");
-    out.write(bytes);
+    dataOutputStream.write(bytes);
+
     LOG.trace("flush socket");
     out.flush();
   }
 
   static <T> T receive(InputStream in, Class<T> type) throws IOException {
     LOG.trace("read {} from socket", type);
-    int length = readInt(in);
+
+    DataInputStream dataInputStream = new DataInputStream(in);
+
+    int length = dataInputStream.readInt();
     LOG.trace("read message length of {} from socket", length);
     if (length > READ_LIMIT) {
       String message = String.format("received length of %d, which exceeds the limit of %d", length, READ_LIMIT);
       throw new IOException(message);
     }
-    byte[] data = read(in, length);
+
+    byte[] data = new byte[length];
+    dataInputStream.readFully(data);
+
     LOG.trace("convert message to {}", type);
     return objectMapper.readValue(data, type);
   }
 
-  private static void writeInt(OutputStream out, int value) throws IOException {
-    out.write((value >>> 24) & 0xFF);
-    out.write((value >>> 16) & 0xFF);
-    out.write((value >>>  8) & 0xFF);
-    out.write(value & 0xFF);
-  }
-
-  private static int readInt(InputStream in) throws IOException {
-    int b1 = in.read();
-    int b2 = in.read();
-    int b3 = in.read();
-    int b4 = in.read();
-
-    if ((b1 | b2 | b3 | b4) < 0) {
-      throw new EOFException("failed to read int from socket");
-    }
-
-    return ((b1 << 24) + (b2 << 16) + (b3 << 8) + b4);
-  }
-
-  private static byte[] read(InputStream in, int length) throws IOException {
-    byte[] buffer = new byte[length];
-    int read = in.read(buffer);
-    if (read < length) {
-      throw new EOFException("failed to read bytes from socket");
-    }
-    return buffer;
-  }
-
 }

scm-plugins/scm-hg-plugin/src/test/java/sonia/scm/repository/hooks/SocketsTest.java

@@ -30,8 +30,12 @@ import org.junit.jupiter.api.Test;
 
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
+import java.io.EOFException;
 import java.io.IOException;
+import java.io.InputStream;
 
+import static java.util.stream.Collectors.joining;
+import static java.util.stream.Stream.generate;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
@@ -53,8 +57,7 @@ class SocketsTest {
     output.write((512 >>> 16) & 0xFF);
 
     ByteArrayInputStream input = new ByteArrayInputStream(output.toByteArray());
-    IOException ex = assertThrows(IOException.class, () -> Sockets.receive(input, TestValue.class));
-    assertThat(ex.getMessage()).contains("int");
+    assertThrows(EOFException.class, () -> Sockets.receive(input, TestValue.class));
   }
 
   @Test
@@ -66,8 +69,7 @@ class SocketsTest {
     output.write(16 & 0xFF);
 
     ByteArrayInputStream input = new ByteArrayInputStream(output.toByteArray());
-    IOException ex = assertThrows(IOException.class, () -> Sockets.receive(input, TestValue.class));
-    assertThat(ex.getMessage()).contains("bytes");
+    assertThrows(EOFException.class, () -> Sockets.receive(input, TestValue.class));
   }
 
   @Test
@@ -83,6 +85,22 @@ class SocketsTest {
     assertThat(ex.getMessage()).contains("9216");
   }
 
+  @Test
+  void shouldSendAndReceiveWithChunks() throws IOException {
+    String stringValue = generate(() -> "a").limit(1024).collect(joining());
+
+    ByteArrayOutputStream output = new ByteArrayOutputStream();
+    Sockets.send(output, new TestValue(stringValue));
+    InputStream input = new ByteArrayInputStream(output.toByteArray()) {
+      @Override
+      public synchronized int read(byte[] b, int off, int len) {
+        return super.read(b, off, Math.min(8, len));
+      }
+    };
+    TestValue value = Sockets.receive(input, TestValue.class);
+    assertThat(value.value).hasSize(1024);
+  }
+
   @Data
   @AllArgsConstructor
   public static class TestValue {