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

ARTEMIS-503 - replace proton-jms with proton-jms from ActiveMQ 

https://issues.apache.org/jira/browse/ARTEMIS-503
This commit is contained in:
Andy Taylor 2016-04-25 09:38:03 +01:00 committed by Martyn Taylor
parent 76d937fab3
commit 9a17681f83
10 changed files with 123 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
artemis-distribution/pom.xml
View File

@ -146,10 +146,6 @@
<dependency>
<groupId>org.jboss.logmanager</groupId>
<artifactId>jboss-logmanager</artifactId>
</dependency>
<dependency>
<groupId>org.apache.qpid</groupId>
<artifactId>proton-jms</artifactId>
</dependency>
<dependency>
<groupId>io.airlift</groupId>
@ -190,6 +186,10 @@
<groupId>io.netty</groupId>
<artifactId>netty-codec-mqtt</artifactId>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-amqp</artifactId>
</dependency>
</dependencies>
<build>

2
artemis-distribution/src/main/assembly/dep.xml
View File

@ -80,7 +80,7 @@
<include>org.jboss.logging:jboss-logging</include>
<include>io.netty:netty-all</include>
<include>org.apache.qpid:proton-j</include>
<include>org.apache.qpid:proton-jms</include>
<include>org.apache.activemq:activemq-amqp</include>
<include>org.apache.activemq:activemq-client</include>
<include>org.slf4j:slf4j-api</include>
<include>io.airlift:airline</include>

8
artemis-protocols/artemis-amqp-protocol/pom.xml
View File

@ -41,6 +41,10 @@
<artifactId>artemis-core-client</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.jboss.logging</groupId>
<artifactId>jboss-logging-processor</artifactId>
@ -83,10 +87,6 @@
<groupId>org.apache.qpid</groupId>
<artifactId>proton-j</artifactId>
</dependency>
<dependency>
<groupId>org.apache.qpid</groupId>
<artifactId>proton-jms</artifactId>
</dependency>
<dependency>
<groupId>org.apache.geronimo.specs</groupId>
<artifactId>geronimo-jms_2.0_spec</artifactId>

56
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/converter/AMQPNativeOutboundTransformer.java Normal file
View File

@ -0,0 +1,56 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
* <p>
* http://www.apache.org/licenses/LICENSE-2.0
* <p>
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.artemis.core.protocol.proton.converter;
import org.apache.activemq.transport.amqp.message.OutboundTransformer;
import org.apache.qpid.proton.amqp.UnsignedInteger;
import org.apache.qpid.proton.amqp.messaging.Header;
import org.apache.qpid.proton.message.ProtonJMessage;
import javax.jms.BytesMessage;
import javax.jms.JMSException;
public class AMQPNativeOutboundTransformer {
static ProtonJMessage transform(OutboundTransformer options, BytesMessage msg) throws JMSException {
byte[] data = new byte[(int) msg.getBodyLength()];
msg.readBytes(data);
msg.reset();
int count = msg.getIntProperty("JMSXDeliveryCount");
// decode...
ProtonJMessage amqp = (ProtonJMessage) org.apache.qpid.proton.message.Message.Factory.create();
int offset = 0;
int len = data.length;
while (len > 0) {
final int decoded = amqp.decode(data, offset, len);
assert decoded > 0 : "Make progress decoding the message";
offset += decoded;
len -= decoded;
}
// Update the DeliveryCount header...
// The AMQP delivery-count field only includes prior failed delivery attempts,
// whereas JMSXDeliveryCount includes the first/current delivery attempt. Subtract 1.
if (amqp.getHeader() == null) {
amqp.setHeader(new Header());
}
amqp.getHeader().setDeliveryCount(new UnsignedInteger(count - 1));
return amqp;
}
}

9
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/converter/ActiveMQJMSVendor.java
View File

@ -26,7 +26,6 @@ import javax.jms.TextMessage;
import org.apache.activemq.artemis.core.protocol.proton.converter.jms.ServerDestination;
import org.apache.activemq.artemis.jms.client.ActiveMQDestination;
import org.apache.qpid.proton.jms.JMSVendor;
import org.apache.activemq.artemis.core.buffers.impl.ResetLimitWrappedActiveMQBuffer;
import org.apache.activemq.artemis.core.protocol.proton.converter.jms.ServerJMSBytesMessage;
import org.apache.activemq.artemis.core.protocol.proton.converter.jms.ServerJMSMapMessage;
@ -36,8 +35,9 @@ import org.apache.activemq.artemis.core.protocol.proton.converter.jms.ServerJMST
import org.apache.activemq.artemis.core.server.ServerMessage;
import org.apache.activemq.artemis.core.server.impl.ServerMessageImpl;
import org.apache.activemq.artemis.utils.IDGenerator;
import org.apache.activemq.transport.amqp.message.JMSVendor;
public class ActiveMQJMSVendor extends JMSVendor {
public class ActiveMQJMSVendor implements JMSVendor {
private final IDGenerator serverGenerator;
@ -85,11 +85,6 @@ public class ActiveMQJMSVendor extends JMSVendor {
return new ServerDestination(name);
}
@Override
public <T extends Destination> T createDestination(String name, Class<T> kind) {
return super.createDestination(name, kind);
}
@Override
public void setJMSXGroupID(Message message, String s) {

51
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/converter/ProtonMessageConverter.java
View File

@ -16,23 +16,30 @@
*/
package org.apache.activemq.artemis.core.protocol.proton.converter;
import org.apache.qpid.proton.jms.EncodedMessage;
import org.apache.qpid.proton.jms.InboundTransformer;
import org.apache.qpid.proton.jms.JMSMappingInboundTransformer;
import org.apache.qpid.proton.jms.JMSMappingOutboundTransformer;
import org.apache.activemq.artemis.core.client.ActiveMQClientLogger;
import org.apache.activemq.transport.amqp.message.EncodedMessage;
import org.apache.activemq.transport.amqp.message.InboundTransformer;
import org.apache.activemq.transport.amqp.message.JMSMappingInboundTransformer;
import org.apache.activemq.transport.amqp.message.JMSMappingOutboundTransformer;
import org.apache.activemq.artemis.core.protocol.proton.converter.jms.ServerJMSMessage;
import org.apache.activemq.artemis.core.server.ServerMessage;
import org.apache.activemq.artemis.spi.core.protocol.MessageConverter;
import org.apache.activemq.artemis.utils.IDGenerator;
import javax.jms.BytesMessage;
import java.io.IOException;
public class ProtonMessageConverter implements MessageConverter {
ActiveMQJMSVendor activeMQJMSVendor;
private final String prefixVendor;
public ProtonMessageConverter(IDGenerator idGenerator) {
activeMQJMSVendor = new ActiveMQJMSVendor(idGenerator);
inboundTransformer = new JMSMappingInboundTransformer(activeMQJMSVendor);
outboundTransformer = new JMSMappingOutboundTransformer(activeMQJMSVendor);
prefixVendor = outboundTransformer.getPrefixVendor();
}
private final InboundTransformer inboundTransformer;
@ -50,11 +57,30 @@ public class ProtonMessageConverter implements MessageConverter {
*
* @param messageSource
* @return
* @throws Exception
* @throws Exception https://issues.jboss.org/browse/ENTMQ-1560
*/
public ServerJMSMessage inboundJMSType(EncodedMessage messageSource) throws Exception {
EncodedMessage encodedMessageSource = messageSource;
ServerJMSMessage transformedMessage = (ServerJMSMessage) inboundTransformer.transform(encodedMessageSource);
ServerJMSMessage transformedMessage = null;
InboundTransformer transformer = inboundTransformer;
while (transformer != null) {
try {
transformedMessage = (ServerJMSMessage) transformer.transform(encodedMessageSource);
break;
}
catch (Exception e) {
ActiveMQClientLogger.LOGGER.debug("Transform of message using [{}] transformer, failed" + inboundTransformer.getTransformerName());
ActiveMQClientLogger.LOGGER.trace("Transformation error:", e);
transformer = transformer.getFallbackTransformer();
}
}
if (transformedMessage == null) {
throw new IOException("Failed to transform incoming delivery, skipping.");
}
transformedMessage.encode();
@ -64,8 +90,19 @@ public class ProtonMessageConverter implements MessageConverter {
@Override
public Object outbound(ServerMessage messageOutbound, int deliveryCount) throws Exception {
ServerJMSMessage jmsMessage = activeMQJMSVendor.wrapMessage(messageOutbound.getType(), messageOutbound, deliveryCount);
jmsMessage.decode();
return outboundTransformer.convert(jmsMessage);
if (jmsMessage.getBooleanProperty(prefixVendor + "NATIVE")) {
if (jmsMessage instanceof BytesMessage) {
return AMQPNativeOutboundTransformer.transform(outboundTransformer, (BytesMessage) jmsMessage);
}
else {
return null;
}
}
else {
return outboundTransformer.convert(jmsMessage);
}
}
}

2
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/proton/plug/ProtonSessionIntegrationCallback.java
View File

@ -24,13 +24,13 @@ import org.apache.activemq.artemis.core.server.MessageReference;
import org.apache.activemq.artemis.core.transaction.Transaction;
import org.apache.activemq.artemis.spi.core.remoting.Connection;
import org.apache.activemq.artemis.spi.core.remoting.ReadyListener;
import org.apache.activemq.transport.amqp.message.EncodedMessage;
import org.apache.qpid.proton.amqp.Binary;
import org.apache.qpid.proton.amqp.transport.AmqpError;
import org.apache.qpid.proton.amqp.transport.ErrorCondition;
import org.apache.qpid.proton.engine.Delivery;
import org.apache.qpid.proton.engine.Link;
import org.apache.qpid.proton.engine.Receiver;
import org.apache.qpid.proton.jms.EncodedMessage;
import org.apache.qpid.proton.message.ProtonJMessage;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.api.core.client.ActiveMQClient;

2
artemis-protocols/artemis-amqp-protocol/src/test/java/org/apache/activemq/artemis/core/protocol/proton/TestConversions.java
View File

@ -26,12 +26,12 @@ import java.util.Map;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.transport.amqp.message.EncodedMessage;
import org.apache.qpid.proton.amqp.Binary;
import org.apache.qpid.proton.amqp.messaging.AmqpSequence;
import org.apache.qpid.proton.amqp.messaging.AmqpValue;
import org.apache.qpid.proton.amqp.messaging.ApplicationProperties;
import org.apache.qpid.proton.amqp.messaging.Data;
import org.apache.qpid.proton.jms.EncodedMessage;
import org.apache.qpid.proton.message.Message;
import org.apache.qpid.proton.message.ProtonJMessage;
import org.apache.qpid.proton.message.impl.MessageImpl;

14
pom.xml
View File

@ -408,15 +408,21 @@
<!-- License: Apache 2.0 -->
</dependency>
<dependency>
<groupId>org.apache.qpid</groupId>
<artifactId>proton-jms</artifactId>
<version>${proton.version}</version>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-client</artifactId>
<version>${activemq5-version}</version>
<!-- License: Apache 2.0 -->
</dependency>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>activemq-client</artifactId>
<artifactId>activemq-amqp</artifactId>
<version>${activemq5-version}</version>
<exclusions>
<exclusion>
<groupId>org.apache.geronimo.specs</groupId>
<artifactId>geronimo-jms_1.1_spec</artifactId>
</exclusion>
</exclusions>
<!-- License: Apache 2.0 -->
</dependency>
<dependency>

4
tests/integration-tests/pom.xml
View File

@ -194,10 +194,6 @@
<groupId>org.apache.qpid</groupId>
<artifactId>proton-j</artifactId>
</dependency>
<dependency>
<groupId>org.apache.qpid</groupId>
<artifactId>proton-jms</artifactId>
</dependency>
<dependency>
<groupId>org.apache.qpid</groupId>
<artifactId>qpid-client</artifactId>