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

ARTEMIS-1038 Make usage of Delivery.available and upgrade proton

This commit is contained in:
Clebert Suconic 2017-03-14 11:38:40 -04:00
parent 9e6c40a8de
commit ae34b01065
3 changed files with 10 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/ProtonServerReceiverContext.java
View File

@ -19,8 +19,6 @@ package org.apache.activemq.artemis.protocol.amqp.proton;
import java.util.Arrays;
import java.util.List;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.core.transaction.Transaction;
import org.apache.activemq.artemis.protocol.amqp.broker.AMQPSessionCallback;
@ -28,7 +26,6 @@ import org.apache.activemq.artemis.protocol.amqp.exceptions.ActiveMQAMQPExceptio
import org.apache.activemq.artemis.protocol.amqp.exceptions.ActiveMQAMQPInternalErrorException;
import org.apache.activemq.artemis.protocol.amqp.exceptions.ActiveMQAMQPNotFoundException;
import org.apache.activemq.artemis.protocol.amqp.logger.ActiveMQAMQPProtocolMessageBundle;
import org.apache.activemq.artemis.protocol.amqp.util.DeliveryUtil;
import org.apache.qpid.proton.amqp.Symbol;
import org.apache.qpid.proton.amqp.messaging.Rejected;
import org.apache.qpid.proton.amqp.messaging.TerminusExpiryPolicy;
@ -134,7 +131,6 @@ public class ProtonServerReceiverContext extends ProtonInitializable implements
@Override
public void onMessage(Delivery delivery) throws ActiveMQAMQPException {
Receiver receiver;
ByteBuf buffer = null;
try {
receiver = ((Receiver) delivery.getLink());
@ -145,20 +141,17 @@ public class ProtonServerReceiverContext extends ProtonInitializable implements
if (delivery.isPartial()) {
return;
}
// This should be used if getDataLength was avilable
// byte[] data = new byte[delivery.getDataLength()];
buffer = PooledByteBufAllocator.DEFAULT.heapBuffer(10 * 1024);
Transaction tx = null;
byte[] data;
synchronized (connection.getLock()) {
DeliveryUtil.readDelivery(receiver, buffer);
data = new byte[delivery.available()];
receiver.recv(data, 0, data.length);
receiver.advance();
}
byte[] data = new byte[buffer.writerIndex()];
buffer.readBytes(data);
if (delivery.getRemoteState() instanceof TransactionalState) {
TransactionalState txState = (TransactionalState) delivery.getRemoteState();
@ -179,10 +172,6 @@ public class ProtonServerReceiverContext extends ProtonInitializable implements
rejected.setError(condition);
delivery.disposition(rejected);
delivery.settle();
} finally {
if (buffer != null) {
buffer.release();
}
}
}

2
pom.xml
View File

@ -83,7 +83,7 @@
<jgroups.version>3.6.9.Final</jgroups.version>
<maven.assembly.plugin.version>2.4</maven.assembly.plugin.version>
<netty.version>4.1.6.Final</netty.version>
<proton.version>0.16.0</proton.version>
<proton.version>0.18.0</proton.version>
<resteasy.version>3.0.19.Final</resteasy.version>
<slf4j.version>1.7.21</slf4j.version>
<qpid.jms.version>0.20.0</qpid.jms.version>

5
tests/artemis-test-support/src/main/java/org/apache/activemq/transport/amqp/client/util/UnmodifiableDelivery.java
View File

@ -57,6 +57,11 @@ public class UnmodifiableDelivery implements Delivery {
return delivery.getDataLength();
} */
@Override
public int available() {
return delivery.available();
}
@Override
public DeliveryState getLocalState() {
return delivery.getLocalState();