Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

ARTEMIS-1268 Fix LargeMessages over STOMP #1394

Closed
wants to merge 1 commit into from
Closed

ARTEMIS-1268 Fix LargeMessages over STOMP #1394

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
5 changes: 3 additions & 2 deletions ...-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/stomp/StompSession.java
View file
Open in desktop
Expand Up @@ -389,8 +389,9 @@ public void sendInternalLarge(CoreMessage message, boolean direct) throws Except
long id = storageManager.generateID();
LargeServerMessage largeMessage = storageManager.createLargeMessage(id, message);

byte[] bytes = new byte[message.getBodyBuffer().writerIndex() - CoreMessage.BODY_OFFSET];
message.getBodyBuffer().readBytes(bytes);
ActiveMQBuffer body = message.getReadOnlyBodyBuffer();
byte[] bytes = new byte[body.readableBytes()];
body.readBytes(bytes);

largeMessage.addBytes(bytes);

Expand Down
2 changes: 0 additions & 2 deletions ...main/java/org/apache/activemq/artemis/core/protocol/stomp/VersionedStompFrameHandler.java
View file
Open in desktop
Expand Up @@ -299,8 +299,6 @@ public StompFrame createMessageFrame(ICoreMessage serverMessage,

ActiveMQBuffer buffer = bodyBuffer != null ? bodyBuffer : serverMessage.getReadOnlyBodyBuffer();

int bodyPos = (serverMessage).getEndOfBodyPosition() == -1 ? buffer.writerIndex() : (serverMessage).getEndOfBodyPosition();

int size = buffer.writerIndex();

byte[] data = new byte[size];
Expand Down
54 changes: 42 additions & 12 deletions ...on-tests/src/test/java/org/apache/activemq/artemis/tests/integration/stomp/StompTest.java
View file
Open in desktop
Expand Up @@ -16,6 +16,12 @@
*/
package org.apache.activemq.artemis.tests.integration.stomp;

import javax.jms.BytesMessage;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.TextMessage;
import java.io.ByteArrayOutputStream;
import java.nio.charset.StandardCharsets;
import java.util.HashSet;
Expand All @@ -26,13 +32,6 @@
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import javax.jms.BytesMessage;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.TextMessage;

import org.apache.activemq.artemis.api.config.ActiveMQDefaultConfiguration;
import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.api.core.SimpleString;
Expand Down Expand Up @@ -159,12 +158,16 @@ public void onMessage(Message arg0) {
.setMaxUsage(0)
.tick();

for (int i = 1; i <= count; i++) {
// Thread.sleep(1);
// log.info(">>> " + i);
send(conn, getQueuePrefix() + getQueueName(), null, "Hello World!");
// Connection should be closed by broker when disk is full and attempt to send
Exception e = null;
try {
for (int i = 1; i <= count; i++) {
send(conn, getQueuePrefix() + getQueueName(), null, "Hello World!");
}
} catch (Exception se) {
e = se;
}

assertNotNull(e);
// It should encounter the exception on logs
AssertionLoggerHandler.findText("AMQ119119");
} finally {
Expand Down Expand Up @@ -254,6 +257,33 @@ public void sendSTOMPReceiveMQTT() throws Exception {
clientProvider.disconnect();
}

@Test
public void testSendReceiveLargeMessage() throws Exception {
String address = "testLargeMessageAddress";
server.getActiveMQServer().createQueue(SimpleString.toSimpleString(address), RoutingType.ANYCAST, SimpleString.toSimpleString(address), null, true, false);

// STOMP default is UTF-8 == 1 byte per char.
int largeMessageStringSize = 10 * 1024 * 1024; // 10MB
StringBuilder b = new StringBuilder(largeMessageStringSize);
for (int i = 0; i < largeMessageStringSize; i++) {
b.append('t');
}
String payload = b.toString();

// Set up STOMP subscription
conn.connect(defUser, defPass);
subscribe(conn, null, Stomp.Headers.Subscribe.AckModeValues.AUTO, null, null, address, true);

// Send Large Message
System.out.println("Sending Message Size: " + largeMessageStringSize);
send(conn, address, null, payload);

// Receive STOMP Message
ClientStompFrame frame = conn.receiveFrame();
System.out.println(frame.getBody().length());
assertTrue(frame.getBody().equals(payload));
}

@Test
public void sendMQTTReceiveSTOMP() throws Exception {
String payload = "This is a test message";
Expand Down
3 changes: 3 additions & 0 deletions ...ests/src/test/java/org/apache/activemq/artemis/tests/integration/stomp/StompTestBase.java
View file
Open in desktop
Expand Up @@ -137,6 +137,7 @@ public void setUp() throws Exception {
connection.start();
}


/**
* @return
* @throws Exception
Expand Down Expand Up @@ -168,6 +169,8 @@ protected JMSServerManager createServer() throws Exception {
config.setOutgoingInterceptorClassNames(getOutgoingInterceptors());
}

config.setPersistenceEnabled(true);

ActiveMQServer activeMQServer = addServer(ActiveMQServers.newActiveMQServer(config, defUser, defPass));

if (isSecurityEnabled()) {
Expand Down