diff --git a/scm-plugins/scm-hg-plugin/src/main/java/sonia/scm/repository/hooks/Sockets.java b/scm-plugins/scm-hg-plugin/src/main/java/sonia/scm/repository/hooks/Sockets.java index 76fe3f9f88..4a0389a33c 100644 --- a/scm-plugins/scm-hg-plugin/src/main/java/sonia/scm/repository/hooks/Sockets.java +++ b/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 receive(InputStream in, Class 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; - } - } diff --git a/scm-plugins/scm-hg-plugin/src/test/java/sonia/scm/repository/hooks/SocketsTest.java b/scm-plugins/scm-hg-plugin/src/test/java/sonia/scm/repository/hooks/SocketsTest.java index f20c2e93db..bb5b48a420 100644 --- a/scm-plugins/scm-hg-plugin/src/test/java/sonia/scm/repository/hooks/SocketsTest.java +++ b/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 {