diff --git a/src/main/java/com/rabbitmq/client/RpcServer.java b/src/main/java/com/rabbitmq/client/RpcServer.java index 4ec31a801..10afad05d 100644 --- a/src/main/java/com/rabbitmq/client/RpcServer.java +++ b/src/main/java/com/rabbitmq/client/RpcServer.java @@ -17,6 +17,8 @@ package com.rabbitmq.client; import com.rabbitmq.utility.Utility; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.io.IOException; import java.util.concurrent.BlockingQueue; @@ -27,6 +29,9 @@ * The class is agnostic about the format of RPC arguments / return values. */ public class RpcServer { + + private static final Logger LOGGER = LoggerFactory.getLogger(RpcServer.class); + /** Channel we are communicating on */ private final Channel _channel; /** Queue to receive requests from */ @@ -119,8 +124,15 @@ public ShutdownSignalException mainloop() _mainloopRunning = false; continue; } - processRequest(request); - _channel.basicAck(request.getEnvelope().getDeliveryTag(), false); + try { + processRequest(request); + _channel.basicAck(request.getEnvelope().getDeliveryTag(), false); + } catch (ShutdownSignalException sse) { + throw sse; + } catch (RuntimeException e) { + LOGGER.warn("Discarding request that could not be processed", e); + _channel.basicReject(request.getEnvelope().getDeliveryTag(), false); + } } return null; } catch (ShutdownSignalException sse) { diff --git a/src/main/java/com/rabbitmq/client/impl/ValueReader.java b/src/main/java/com/rabbitmq/client/impl/ValueReader.java index 91eb09c3e..7d8cb9cda 100644 --- a/src/main/java/com/rabbitmq/client/impl/ValueReader.java +++ b/src/main/java/com/rabbitmq/client/impl/ValueReader.java @@ -52,6 +52,9 @@ private static long unsignedExtend(int value) return extended & INT_MASK; } + /** Maximum length, in bytes, of a shortstr on the wire. */ + private static final int MAX_SHORTSTR_LENGTH = 255; + /** The stream we are reading from. */ private final DataInputStream in; @@ -71,7 +74,25 @@ private static String readShortstr(DataInputStream in) { byte [] b = new byte[in.readUnsignedByte()]; in.readFully(b); - return new String(b, StandardCharsets.UTF_8); + return truncateToMaxUtf8Length(new String(b, StandardCharsets.UTF_8), MAX_SHORTSTR_LENGTH); + } + + private static String truncateToMaxUtf8Length(String s, int maxBytes) { + if (s.length() <= maxBytes / 3 || s.indexOf('\uFFFD') < 0) { + return s; + } + int bytes = 0; + int i = 0; + while (i < s.length()) { + int codePoint = s.codePointAt(i); + int width = codePoint < 0x80 ? 1 : codePoint < 0x800 ? 2 : codePoint < 0x10000 ? 3 : 4; + if (bytes + width > maxBytes) { + return s.substring(0, i); + } + bytes += width; + i += Character.charCount(codePoint); + } + return s; } /** Public API - reads a short string. */ diff --git a/src/test/java/com/rabbitmq/client/test/ClientTestSuite.java b/src/test/java/com/rabbitmq/client/test/ClientTestSuite.java index 7a979c7c2..a83eedfd9 100644 --- a/src/test/java/com/rabbitmq/client/test/ClientTestSuite.java +++ b/src/test/java/com/rabbitmq/client/test/ClientTestSuite.java @@ -30,6 +30,7 @@ @Suite @SelectClasses({ TableTest.class, + ShortstrRoundTripTest.class, LongStringTest.class, BlockingCellTest.class, TruncatedInputStreamTest.class, diff --git a/src/test/java/com/rabbitmq/client/test/RpcTest.java b/src/test/java/com/rabbitmq/client/test/RpcTest.java index d4cfd3207..d79021879 100644 --- a/src/test/java/com/rabbitmq/client/test/RpcTest.java +++ b/src/test/java/com/rabbitmq/client/test/RpcTest.java @@ -330,6 +330,100 @@ public void handleRecoveryStarted(Recoverable recoverable) { client.close(); } + @Test + public void serverKeepsRunningWhenHandlerFails() throws Exception { + rpcServer = new FailingRpcServer(serverChannel, queue); + Thread serverThread = new Thread(() -> { + try { + rpcServer.mainloop(); + } catch (Exception e) { + } + }); + serverThread.start(); + RpcClient client = new RpcClient(new RpcClientParams() + .channel(clientChannel).exchange("").routingKey(queue).timeout(1000)); + + try { + client.doCall(null, "boom".getBytes()); + fail("The handler failed, the call should have timed out"); + } catch (TimeoutException e) { + } + + RpcClient.Response response = client.doCall(null, "hello".getBytes()); + assertEquals("*** hello ***", new String(response.getBody())); + assertTrue(serverThread.isAlive()); + + client.close(); + } + + @Test + public void requestThatCannotBeProcessedIsNotRequeued() throws Exception { + rpcServer = new FailingRpcServer(serverChannel, queue); + Thread serverThread = new Thread(() -> { + try { + rpcServer.mainloop(); + } catch (Exception e) { + } + }); + serverThread.start(); + RpcClient client = new RpcClient(new RpcClientParams() + .channel(clientChannel).exchange("").routingKey(queue).timeout(1000)); + + try { + client.doCall(null, "boom".getBytes()); + fail("The handler failed, the call should have timed out"); + } catch (TimeoutException e) { + } + + waitAtMost(Duration.ofSeconds(5), () -> clientChannel.messageCount(queue) == 0); + assertTrue(serverThread.isAlive()); + + client.close(); + } + + @Test + public void serverKeepsRunningWhenSeveralHandlerCallsFail() throws Exception { + rpcServer = new FailingRpcServer(serverChannel, queue); + Thread serverThread = new Thread(() -> { + try { + rpcServer.mainloop(); + } catch (Exception e) { + } + }); + serverThread.start(); + RpcClient client = new RpcClient(new RpcClientParams() + .channel(clientChannel).exchange("").routingKey(queue).timeout(1000)); + + for (int i = 0; i < 5; i++) { + try { + client.doCall(null, "boom".getBytes()); + fail("The handler failed, the call should have timed out"); + } catch (TimeoutException e) { + } + } + + RpcClient.Response response = client.doCall(null, "hello".getBytes()); + assertEquals("*** hello ***", new String(response.getBody())); + assertTrue(serverThread.isAlive()); + + client.close(); + } + + private static class FailingRpcServer extends TestRpcServer { + + public FailingRpcServer(Channel channel, String queueName) throws IOException { + super(channel, queueName); + } + + @Override + public byte[] handleCall(Delivery request, AMQP.BasicProperties replyProperties) { + if ("boom".equals(new String(request.getBody()))) { + throw new IllegalArgumentException("cannot handle this request"); + } + return super.handleCall(request, replyProperties); + } + } + private static class TestRpcServer extends RpcServer { public TestRpcServer(Channel channel, String queueName) throws IOException { diff --git a/src/test/java/com/rabbitmq/client/test/ShortstrRoundTripTest.java b/src/test/java/com/rabbitmq/client/test/ShortstrRoundTripTest.java new file mode 100644 index 000000000..435f06a89 --- /dev/null +++ b/src/test/java/com/rabbitmq/client/test/ShortstrRoundTripTest.java @@ -0,0 +1,144 @@ +// Copyright (c) 2007-2026 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries. +// +// This software, the RabbitMQ Java client library, is triple-licensed under the +// Mozilla Public License 2.0 ("MPL"), the GNU General Public License version 2 +// ("GPL") and the Apache License version 2 ("ASL"). For the MPL, please see +// LICENSE-MPL-RabbitMQ. For the GPL, please see LICENSE-GPL2. For the ASL, +// please see LICENSE-APACHE2. +// +// This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND, +// either express or implied. See the LICENSE file for specific language governing +// rights and limitations of this software. +// +// If you have any questions regarding licensing, please contact us at +// info@rabbitmq.com. + +package com.rabbitmq.client.test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; + +import com.rabbitmq.client.AMQP; +import com.rabbitmq.client.impl.ValueReader; +import com.rabbitmq.client.impl.ValueWriter; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import org.junit.jupiter.api.Test; + +public class ShortstrRoundTripTest { + + private static String read(byte[] payload) throws IOException { + ByteArrayOutputStream frame = new ByteArrayOutputStream(); + frame.write(payload.length); + frame.write(payload); + return new ValueReader(new DataInputStream(new ByteArrayInputStream(frame.toByteArray()))) + .readShortstr(); + } + + private static byte[] write(String s) throws IOException { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + new ValueWriter(new DataOutputStream(out)).writeShortstr(s); + return out.toByteArray(); + } + + @Test + public void valueReadFromWireCanBeWrittenBack() throws IOException { + byte[] payload = new byte[255]; + Arrays.fill(payload, (byte) 0xFF); + String decoded = read(payload); + assertThat(decoded.getBytes(StandardCharsets.UTF_8).length).isLessThanOrEqualTo(255); + assertThatCode(() -> write(decoded)).doesNotThrowAnyException(); + } + + @Test + public void wellFormedValuesArePreserved() throws IOException { + for (String value : new String[] {"", "hello", "santé", "你好", "😀"}) { + assertThat(read(value.getBytes(StandardCharsets.UTF_8))).isEqualTo(value); + } + StringBuilder sb = new StringBuilder(); + for (int i = 0; i < 85; i++) { + sb.append("你"); + } + String maxLength = sb.toString(); + assertThat(maxLength.getBytes(StandardCharsets.UTF_8).length).isEqualTo(255); + assertThat(read(maxLength.getBytes(StandardCharsets.UTF_8))).isEqualTo(maxLength); + } + + @Test + public void partiallyMalformedValueStaysWithinLimit() throws IOException { + byte[] payload = new byte[255]; + Arrays.fill(payload, (byte) 'a'); + for (int i = 100; i < 255; i++) { + payload[i] = (byte) 0xFF; + } + String decoded = read(payload); + assertThat(decoded.getBytes(StandardCharsets.UTF_8).length).isLessThanOrEqualTo(255); + assertThatCode(() -> write(decoded)).doesNotThrowAnyException(); + assertThat(decoded).startsWith("aaaa"); + } + + @Test + public void messagePropertiesReadFromWireCanBeWrittenBack() throws IOException { + byte[] malformed = new byte[255]; + Arrays.fill(malformed, (byte) 0xFF); + + ByteArrayOutputStream header = new ByteArrayOutputStream(); + DataOutputStream out = new DataOutputStream(header); + out.writeShort(0); + out.writeLong(6); + out.writeShort((1 << 10) | (1 << 9) | (1 << 7) | (1 << 5) | (1 << 4)); + out.writeByte(malformed.length); + out.write(malformed); + out.writeByte(malformed.length); + out.write(malformed); + out.writeByte(malformed.length); + out.write(malformed); + out.writeByte(malformed.length); + out.write(malformed); + out.writeByte(malformed.length); + out.write(malformed); + out.flush(); + + AMQP.BasicProperties properties = + new AMQP.BasicProperties( + new DataInputStream(new ByteArrayInputStream(header.toByteArray()))); + + assertThat(properties.getCorrelationId()).isNotNull(); + assertThat(properties.getReplyTo()).isNotNull(); + assertThat(properties.getMessageId()).isNotNull(); + assertThat(properties.getType()).isNotNull(); + assertThat(properties.getUserId()).isNotNull(); + + AMQP.BasicProperties echoed = + new AMQP.BasicProperties.Builder() + .correlationId(properties.getCorrelationId()) + .replyTo(properties.getReplyTo()) + .messageId(properties.getMessageId()) + .type(properties.getType()) + .userId(properties.getUserId()) + .build(); + + assertThatCode( + () -> { + ByteArrayOutputStream sink = new ByteArrayOutputStream(); + echoed.writePropertiesTo( + new com.rabbitmq.client.impl.ContentHeaderPropertyWriter( + new DataOutputStream(sink))); + }) + .doesNotThrowAnyException(); + } + + @Test + public void oversizedValuesAreStillRejected() { + StringBuilder sb = new StringBuilder(); + for (int i = 0; i < 256; i++) { + sb.append('a'); + } + assertThatCode(() -> write(sb.toString())).isInstanceOf(IllegalArgumentException.class); + } +}