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 #1132

This commit is contained in:
Clebert Suconic 2017-03-24 17:46:01 -04:00
parent 52a2c5dd06 40b9ac0a20
commit f2e0891b0e
3 changed files with 37 additions and 22 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/AMQPConnectionContext.java
View File

@ -16,6 +16,12 @@
*/
package org.apache.activemq.artemis.protocol.amqp.proton;
import static org.apache.activemq.artemis.protocol.amqp.proton.AmqpSupport.FAILOVER_SERVER_LIST;
import static org.apache.activemq.artemis.protocol.amqp.proton.AmqpSupport.HOSTNAME;
import static org.apache.activemq.artemis.protocol.amqp.proton.AmqpSupport.NETWORK_HOST;
import static org.apache.activemq.artemis.protocol.amqp.proton.AmqpSupport.PORT;
import static org.apache.activemq.artemis.protocol.amqp.proton.AmqpSupport.SCHEME;
import java.net.URI;
import java.util.Arrays;
import java.util.HashMap;
@ -50,12 +56,6 @@ import org.jboss.logging.Logger;
import io.netty.buffer.ByteBuf;
import static org.apache.activemq.artemis.protocol.amqp.proton.AmqpSupport.FAILOVER_SERVER_LIST;
import static org.apache.activemq.artemis.protocol.amqp.proton.AmqpSupport.HOSTNAME;
import static org.apache.activemq.artemis.protocol.amqp.proton.AmqpSupport.NETWORK_HOST;
import static org.apache.activemq.artemis.protocol.amqp.proton.AmqpSupport.PORT;
import static org.apache.activemq.artemis.protocol.amqp.proton.AmqpSupport.SCHEME;
public class AMQPConnectionContext extends ProtonInitializable {
private static final Logger log = Logger.getLogger(AMQPConnectionContext.class);
@ -210,7 +210,6 @@ public class AMQPConnectionContext extends ProtonInitializable {
} else {
Sender sender = (Sender) link;
protonSession.addSender(sender);
sender.offer(1);
}
}
@ -421,8 +420,10 @@ public class AMQPConnectionContext extends ProtonInitializable {
@Override
public void onRemoteClose(Link link) throws Exception {
synchronized (getLock()) {
link.close();
link.free();
}
ProtonDeliveryHandler linkContext = (ProtonDeliveryHandler) link.getContext();
if (linkContext != null) {
linkContext.close(true);
@ -431,10 +432,14 @@ public class AMQPConnectionContext extends ProtonInitializable {
@Override
public void onRemoteDetach(Link link) throws Exception {
synchronized (getLock()) {
link.detach();
link.free();
}
flush();
}
@Override
public void onLocalDetach(Link link) throws Exception {
Object context = link.getContext();

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

@ -147,9 +147,11 @@ public class AMQPSessionContext extends ProtonInitializable {
coordinator.setCapabilities(Symbol.getSymbol("amqp:local-transactions"), Symbol.getSymbol("amqp:multi-txns-per-ssn"), Symbol.getSymbol("amqp:multi-ssns-per-txn"));
receiver.setContext(transactionHandler);
synchronized (connection.getLock()) {
receiver.open();
receiver.flow(ProtonTransactionHandler.DEFAULT_COORDINATOR_CREDIT);
}
}
public void addSender(Sender sender) throws Exception {
// TODO: Remove this check when we have support for global link names
@ -161,15 +163,19 @@ public class AMQPSessionContext extends ProtonInitializable {
senders.put(sender, protonSender);
serverSenders.put(protonSender.getBrokerConsumer(), protonSender);
sender.setContext(protonSender);
synchronized (connection.getLock()) {
sender.open();
}
protonSender.start();
} catch (ActiveMQAMQPException e) {
senders.remove(sender);
sender.setSource(null);
sender.setCondition(new ErrorCondition(e.getAmqpError(), e.getMessage()));
synchronized (connection.getLock()) {
sender.close();
}
}
}
public void removeSender(Sender sender) throws ActiveMQAMQPException {
senders.remove(sender);
@ -185,12 +191,16 @@ public class AMQPSessionContext extends ProtonInitializable {
protonReceiver.initialise();
receivers.put(receiver, protonReceiver);
receiver.setContext(protonReceiver);
synchronized (connection.getLock()) {
receiver.open();
}
} catch (ActiveMQAMQPException e) {
receivers.remove(receiver);
receiver.setTarget(null);
receiver.setCondition(new ErrorCondition(e.getAmqpError(), e.getMessage()));
synchronized (connection.getLock()) {
receiver.close();
}
}
}
}

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

@ -170,10 +170,12 @@ public class ProtonServerReceiverContext extends ProtonInitializable implements
condition.setCondition(Symbol.valueOf("failed"));
condition.setDescription(e.getMessage());
rejected.setError(condition);
synchronized (connection.getLock()) {
delivery.disposition(rejected);
delivery.settle();
}
}
}
@Override
public void close(boolean remoteLinkClose) throws ActiveMQAMQPException {
@ -204,7 +206,6 @@ public class ProtonServerReceiverContext extends ProtonInitializable implements
}
connection.flush();
}
}
public void drain(int credits) {
@ -221,5 +222,4 @@ public class ProtonServerReceiverContext extends ProtonInitializable implements
public boolean isDraining() {
return receiver.draining();
}
}