Apache
/
activemq-artemis
mirror of https://github.com/apache/activemq-artemis.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

This closes #942 ARTEMIS-902 OpenWire Compression Issue

This commit is contained in:
Andy Taylor 2016-12-23 15:08:03 +00:00
parent 22f0fcf08b eecbbb18db
commit 79b48767d4
2 changed files with 50 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireMessageConverter.java
View File

@ -490,10 +490,11 @@ public class OpenWireMessageConverter implements MessageConverter {
ByteArrayOutputStream bytesOut = new ByteArrayOutputStream(text.length() + 4); ByteArrayOutputStream bytesOut = new ByteArrayOutputStream(text.length() + 4);
OutputStream out = bytesOut; OutputStream out = bytesOut;
if (isCompressed) { if (isCompressed) {
out = new DeflaterOutputStream(out); out = new DeflaterOutputStream(out, true);
} }
try (DataOutputStream dataOut = new DataOutputStream(out)) { try (DataOutputStream dataOut = new DataOutputStream(out)) {
MarshallingSupport.writeUTF8(dataOut, text.toString()); MarshallingSupport.writeUTF8(dataOut, text.toString());
dataOut.flush();
bytes = bytesOut.toByteArray(); bytes = bytesOut.toByteArray();
} }
} }
@ -506,10 +507,11 @@ public class OpenWireMessageConverter implements MessageConverter {
ByteArrayOutputStream out = new ByteArrayOutputStream(mapData.getEncodeSize()); ByteArrayOutputStream out = new ByteArrayOutputStream(mapData.getEncodeSize());
OutputStream os = out; OutputStream os = out;
if (isCompressed) { if (isCompressed) {
os = new DeflaterOutputStream(os); os = new DeflaterOutputStream(os, true);
} }
try (DataOutputStream dataOut = new DataOutputStream(os)) { try (DataOutputStream dataOut = new DataOutputStream(os)) {
MarshallingSupport.marshalPrimitiveMap(map, dataOut); MarshallingSupport.marshalPrimitiveMap(map, dataOut);
dataOut.flush();
} }
bytes = out.toByteArray(); bytes = out.toByteArray();
} }
@ -520,8 +522,9 @@ public class OpenWireMessageConverter implements MessageConverter {
buffer.readBytes(bytes); buffer.readBytes(bytes);
if (isCompressed) { if (isCompressed) {
ByteArrayOutputStream bytesOut = new ByteArrayOutputStream(); ByteArrayOutputStream bytesOut = new ByteArrayOutputStream();
try (DeflaterOutputStream out = new DeflaterOutputStream(bytesOut)) { try (DeflaterOutputStream out = new DeflaterOutputStream(bytesOut, true)) {
out.write(bytes); out.write(bytes);
out.flush();
} }
bytes = bytesOut.toByteArray(); bytes = bytesOut.toByteArray();
} }
@ -529,7 +532,7 @@ public class OpenWireMessageConverter implements MessageConverter {
org.apache.activemq.util.ByteArrayOutputStream bytesOut = new org.apache.activemq.util.ByteArrayOutputStream(); org.apache.activemq.util.ByteArrayOutputStream bytesOut = new org.apache.activemq.util.ByteArrayOutputStream();
OutputStream out = bytesOut; OutputStream out = bytesOut;
if (isCompressed) { if (isCompressed) {
out = new DeflaterOutputStream(bytesOut); out = new DeflaterOutputStream(bytesOut, true);
} }
try (DataOutputStream dataOut = new DataOutputStream(out)) { try (DataOutputStream dataOut = new DataOutputStream(out)) {
@ -583,6 +586,7 @@ public class OpenWireMessageConverter implements MessageConverter {
stop = true; stop = true;
break; break;
} }
dataOut.flush();
} }
} }
bytes = bytesOut.toByteArray(); bytes = bytesOut.toByteArray();
@ -602,6 +606,7 @@ public class OpenWireMessageConverter implements MessageConverter {
int count = deflater.deflate(bytesBuf); int count = deflater.deflate(bytesBuf);
compressed.write(bytesBuf, 0, count); compressed.write(bytesBuf, 0, count);
} }
compressed.flush();
ByteSequence byteSeq = compressed.toByteSequence(); ByteSequence byteSeq = compressed.toByteSequence();
ByteSequenceData.writeIntBig(byteSeq, length); ByteSequenceData.writeIntBig(byteSeq, length);
bytes = Arrays.copyOfRange(byteSeq.data, 0, byteSeq.length); bytes = Arrays.copyOfRange(byteSeq.data, 0, byteSeq.length);
@ -615,8 +620,9 @@ public class OpenWireMessageConverter implements MessageConverter {
buffer.readBytes(bytes); buffer.readBytes(bytes);
if (isCompressed) { if (isCompressed) {
try (ByteArrayOutputStream bytesOut = new ByteArrayOutputStream(); try (ByteArrayOutputStream bytesOut = new ByteArrayOutputStream();
DeflaterOutputStream out = new DeflaterOutputStream(bytesOut)) { DeflaterOutputStream out = new DeflaterOutputStream(bytesOut, true)) {
out.write(bytes); out.write(bytes);
out.flush();
bytes = bytesOut.toByteArray(); bytes = bytesOut.toByteArray();
} }
} }

69
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/interop/CompressedInteropTest.java
View File

@ -18,6 +18,7 @@ package org.apache.activemq.artemis.tests.integration.openwire.interop;
import javax.jms.BytesMessage; import javax.jms.BytesMessage;
import javax.jms.Connection; import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.MapMessage; import javax.jms.MapMessage;
import javax.jms.Message; import javax.jms.Message;
import javax.jms.MessageConsumer; import javax.jms.MessageConsumer;
@ -63,24 +64,32 @@ public class CompressedInteropTest extends BasicOpenWireTest {
xaFactory.setUseCompression(true); xaFactory.setUseCompression(true);
} }
@Test @Test
public void testCoreReceiveOpenWireCompressedMessages() throws Exception { public void testCoreReceiveOpenWireCompressedMessages() throws Exception {
testCompressedMessageSendReceive(true);
}
@Test
public void testOpenWireReceiveOpenWireCompressedMessages() throws Exception {
testCompressedMessageSendReceive(false);
}
private void testCompressedMessageSendReceive(boolean useCore) throws Exception {
//TextMessage //TextMessage
sendCompressedTextMessageUsingOpenWire(); sendCompressedTextMessageUsingOpenWire();
receiveTextMessageUsingCore(); receiveTextMessage(useCore);
//BytesMessage //BytesMessage
sendCompressedBytesMessageUsingOpenWire(); sendCompressedBytesMessageUsingOpenWire();
receiveBytesMessageUsingCore(); receiveBytesMessage(useCore);
//MapMessage //MapMessage
sendCompressedMapMessageUsingOpenWire(); sendCompressedMapMessageUsingOpenWire();
receiveMapMessageUsingCore(); receiveMapMessage(useCore);
//StreamMessage //StreamMessage
sendCompressedStreamMessageUsingOpenWire(); sendCompressedStreamMessageUsingOpenWire();
receiveStreamMessageUsingCore(); receiveStreamMessage(useCore);
//ObjectMessage //ObjectMessage
sendCompressedObjectMessageUsingOpenWire(); sendCompressedObjectMessageUsingOpenWire();
receiveObjectMessageUsingCore(); receiveObjectMessage(useCore);
} }
private void sendCompressedStreamMessageUsingOpenWire() throws Exception { private void sendCompressedStreamMessageUsingOpenWire() throws Exception {
@ -106,8 +115,8 @@ public class CompressedInteropTest extends BasicOpenWireTest {
producer.send(streamMessage); producer.send(streamMessage);
} }
private void receiveStreamMessageUsingCore() throws Exception { private void receiveStreamMessage(boolean useCore) throws Exception {
StreamMessage streamMessage = (StreamMessage) receiveMessageUsingCore(); StreamMessage streamMessage = (StreamMessage) receiveMessage(useCore);
boolean booleanVal = streamMessage.readBoolean(); boolean booleanVal = streamMessage.readBoolean();
assertTrue(booleanVal); assertTrue(booleanVal);
byte byteVal = streamMessage.readByte(); byte byteVal = streamMessage.readByte();
@ -149,8 +158,8 @@ public class CompressedInteropTest extends BasicOpenWireTest {
producer.send(objectMessage); producer.send(objectMessage);
} }
private void receiveObjectMessageUsingCore() throws Exception { private void receiveObjectMessage(boolean useCore) throws Exception {
ObjectMessage objectMessage = (ObjectMessage) receiveMessageUsingCore(); ObjectMessage objectMessage = (ObjectMessage) receiveMessage(useCore);
Object objectVal = objectMessage.getObject(); Object objectVal = objectMessage.getObject();
assertEquals(TEXT, objectVal); assertEquals(TEXT, objectVal);
} }
@ -178,8 +187,8 @@ public class CompressedInteropTest extends BasicOpenWireTest {
producer.send(mapMessage); producer.send(mapMessage);
} }
private void receiveMapMessageUsingCore() throws Exception { private void receiveMapMessage(boolean useCore) throws Exception {
MapMessage mapMessage = (MapMessage) receiveMessageUsingCore(); MapMessage mapMessage = (MapMessage) receiveMessage(useCore);
boolean booleanVal = mapMessage.getBoolean("boolean-type"); boolean booleanVal = mapMessage.getBoolean("boolean-type");
assertTrue(booleanVal); assertTrue(booleanVal);
@ -222,8 +231,8 @@ public class CompressedInteropTest extends BasicOpenWireTest {
producer.send(bytesMessage); producer.send(bytesMessage);
} }
private void receiveBytesMessageUsingCore() throws Exception { private void receiveBytesMessage(boolean useCore) throws Exception {
BytesMessage bytesMessage = (BytesMessage) receiveMessageUsingCore(); BytesMessage bytesMessage = (BytesMessage) receiveMessage(useCore);
byte[] bytes = new byte[TEXT.getBytes(StandardCharsets.UTF_8).length]; byte[] bytes = new byte[TEXT.getBytes(StandardCharsets.UTF_8).length];
bytesMessage.readBytes(bytes); bytesMessage.readBytes(bytes);
@ -233,16 +242,28 @@ public class CompressedInteropTest extends BasicOpenWireTest {
assertEquals(TEXT, rcvString); assertEquals(TEXT, rcvString);
} }
private void receiveTextMessageUsingCore() throws Exception { private void receiveTextMessage(boolean useCore) throws Exception {
TextMessage txtMessage = (TextMessage) receiveMessageUsingCore(); TextMessage txtMessage = (TextMessage) receiveMessage(useCore);
assertEquals(TEXT, txtMessage.getText()); assertEquals(TEXT, txtMessage.getText());
} }
private Message receiveMessageUsingCore() throws Exception { private void sendCompressedTextMessageUsingOpenWire() throws Exception {
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
ActiveMQDestination destination = createDestination(session, ActiveMQDestination.QUEUE_TYPE);
final ActiveMQMessageProducer producer = (ActiveMQMessageProducer) session.createProducer(destination);
TextMessage textMessage = session.createTextMessage(TEXT);
producer.send(textMessage);
}
private Message receiveMessage(boolean useCore) throws Exception {
ConnectionFactory factoryToUse = useCore ? coreCf : factory;
Connection jmsConn = null; Connection jmsConn = null;
Message message = null; Message message = null;
try { try {
jmsConn = coreCf.createConnection(); jmsConn = factoryToUse.createConnection();
jmsConn.start(); jmsConn.start();
Session session = jmsConn.createSession(false, Session.AUTO_ACKNOWLEDGE); Session session = jmsConn.createSession(false, Session.AUTO_ACKNOWLEDGE);
@ -257,16 +278,4 @@ public class CompressedInteropTest extends BasicOpenWireTest {
} }
return message; return message;
} }
private void sendCompressedTextMessageUsingOpenWire() throws Exception {
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
ActiveMQDestination destination = createDestination(session, ActiveMQDestination.QUEUE_TYPE);
final ActiveMQMessageProducer producer = (ActiveMQMessageProducer) session.createProducer(destination);
TextMessage textMessage = session.createTextMessage(TEXT);
producer.send(textMessage);
}
} }