| /* |
| * 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 |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * 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.protocol.amqp.proton; |
| |
| import java.nio.ByteBuffer; |
| import java.util.HashMap; |
| import java.util.Map; |
| import java.util.Objects; |
| import java.util.Set; |
| import java.util.concurrent.atomic.AtomicBoolean; |
| import java.util.concurrent.atomic.AtomicInteger; |
| |
| import io.netty.buffer.ByteBuf; |
| import io.netty.buffer.PooledByteBufAllocator; |
| import org.apache.activemq.artemis.api.core.ActiveMQException; |
| import org.apache.activemq.artemis.api.core.ActiveMQExceptionType; |
| import org.apache.activemq.artemis.api.core.ActiveMQQueueMaxConsumerLimitReached; |
| import org.apache.activemq.artemis.api.core.ActiveMQSecurityException; |
| import org.apache.activemq.artemis.api.core.Message; |
| import org.apache.activemq.artemis.api.core.RoutingType; |
| import org.apache.activemq.artemis.api.core.SimpleString; |
| import org.apache.activemq.artemis.core.io.IOCallback; |
| import org.apache.activemq.artemis.core.message.LargeBodyReader; |
| import org.apache.activemq.artemis.core.persistence.OperationContext; |
| import org.apache.activemq.artemis.core.server.AddressQueryResult; |
| import org.apache.activemq.artemis.core.server.Consumer; |
| import org.apache.activemq.artemis.core.server.MessageReference; |
| import org.apache.activemq.artemis.core.server.QueueQueryResult; |
| import org.apache.activemq.artemis.core.server.ServerConsumer; |
| import org.apache.activemq.artemis.core.server.impl.ServerConsumerImpl; |
| import org.apache.activemq.artemis.protocol.amqp.broker.AMQPLargeMessage; |
| import org.apache.activemq.artemis.protocol.amqp.broker.AMQPMessage; |
| import org.apache.activemq.artemis.protocol.amqp.broker.AMQPMessageBrokerAccessor; |
| import org.apache.activemq.artemis.protocol.amqp.broker.AMQPSessionCallback; |
| import org.apache.activemq.artemis.protocol.amqp.broker.ActiveMQProtonRemotingConnection; |
| import org.apache.activemq.artemis.protocol.amqp.converter.CoreAmqpConverter; |
| import org.apache.activemq.artemis.protocol.amqp.exceptions.ActiveMQAMQPException; |
| import org.apache.activemq.artemis.protocol.amqp.exceptions.ActiveMQAMQPIllegalStateException; |
| 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.exceptions.ActiveMQAMQPResourceLimitExceededException; |
| import org.apache.activemq.artemis.protocol.amqp.logger.ActiveMQAMQPProtocolMessageBundle; |
| import org.apache.activemq.artemis.protocol.amqp.proton.transaction.ProtonTransactionImpl; |
| import org.apache.activemq.artemis.protocol.amqp.util.NettyReadable; |
| import org.apache.activemq.artemis.protocol.amqp.util.NettyWritable; |
| import org.apache.activemq.artemis.protocol.amqp.util.TLSEncode; |
| import org.apache.activemq.artemis.reader.MessageUtil; |
| import org.apache.activemq.artemis.selector.filter.FilterException; |
| import org.apache.activemq.artemis.selector.impl.SelectorParser; |
| import org.apache.activemq.artemis.spi.core.remoting.ReadyListener; |
| import org.apache.activemq.artemis.utils.CompositeAddress; |
| import org.apache.activemq.artemis.utils.DestinationUtil; |
| import org.apache.qpid.proton.amqp.DescribedType; |
| import org.apache.qpid.proton.amqp.Symbol; |
| import org.apache.qpid.proton.amqp.messaging.Accepted; |
| import org.apache.qpid.proton.amqp.messaging.ApplicationProperties; |
| import org.apache.qpid.proton.amqp.messaging.DeliveryAnnotations; |
| import org.apache.qpid.proton.amqp.messaging.Header; |
| import org.apache.qpid.proton.amqp.messaging.MessageAnnotations; |
| import org.apache.qpid.proton.amqp.messaging.Modified; |
| import org.apache.qpid.proton.amqp.messaging.Outcome; |
| import org.apache.qpid.proton.amqp.messaging.Properties; |
| import org.apache.qpid.proton.amqp.messaging.Source; |
| import org.apache.qpid.proton.amqp.messaging.TerminusDurability; |
| import org.apache.qpid.proton.amqp.messaging.TerminusExpiryPolicy; |
| import org.apache.qpid.proton.amqp.transaction.TransactionalState; |
| import org.apache.qpid.proton.amqp.transport.AmqpError; |
| import org.apache.qpid.proton.amqp.transport.DeliveryState; |
| import org.apache.qpid.proton.amqp.transport.DeliveryState.DeliveryStateType; |
| import org.apache.qpid.proton.amqp.transport.ErrorCondition; |
| import org.apache.qpid.proton.amqp.transport.ReceiverSettleMode; |
| import org.apache.qpid.proton.amqp.transport.SenderSettleMode; |
| import org.apache.qpid.proton.codec.ReadableBuffer; |
| import org.apache.qpid.proton.codec.WritableBuffer; |
| import org.apache.qpid.proton.engine.Delivery; |
| import org.apache.qpid.proton.engine.EndpointState; |
| import org.apache.qpid.proton.engine.Link; |
| import org.apache.qpid.proton.engine.Sender; |
| import org.jboss.logging.Logger; |
| |
| /** |
| * This is the Equivalent for the ServerConsumer |
| */ |
| public class ProtonServerSenderContext extends ProtonInitializable implements ProtonDeliveryHandler { |
| |
| private static final Logger log = Logger.getLogger(ProtonServerSenderContext.class); |
| |
| private static final Symbol COPY = Symbol.valueOf("copy"); |
| private static final Symbol TOPIC = Symbol.valueOf("topic"); |
| private static final Symbol QUEUE = Symbol.valueOf("queue"); |
| private static final Symbol SHARED = Symbol.valueOf("shared"); |
| private static final Symbol GLOBAL = Symbol.valueOf("global"); |
| |
| SenderController controller; |
| |
| private final ConnectionFlushIOCallback connectionFlusher = new ConnectionFlushIOCallback(); |
| |
| private Consumer brokerConsumer; |
| private ReadyListener onflowControlReady; |
| protected final AMQPSessionContext protonSession; |
| protected final Sender sender; |
| protected final AMQPConnectionContext connection; |
| protected boolean closed = false; |
| protected final AMQPSessionCallback sessionSPI; |
| |
| private boolean preSettle; |
| |
| private final AtomicBoolean draining = new AtomicBoolean(false); |
| |
| // once a large message is accepted, we shouldn't accept any further messages |
| // as large message could be interrupted due to flow control and resumed at the same message |
| volatile boolean hasLarge = false; |
| volatile LargeMessageDeliveryContext pendingLargeMessage = null; |
| |
| |
| private int credits = 0; |
| |
| private AtomicInteger pending = new AtomicInteger(0); |
| /** |
| * The model proton uses requires us to hold a lock in certain times |
| * to sync the credits we have versus the credits that are being held in proton |
| * */ |
| private final Object creditsLock = new Object(); |
| private final java.util.function.Consumer<? super MessageReference> executeDelivery; |
| private java.util.function.Consumer<? super MessageReference> beforeDelivery; |
| private final boolean amqpTreatRejectAsUnmodifiedDeliveryFailed; |
| |
| public ProtonServerSenderContext(AMQPConnectionContext connection, |
| Sender sender, |
| AMQPSessionContext protonSession, |
| AMQPSessionCallback server) { |
| this(connection, sender, protonSession, server, null); |
| } |
| |
| public ProtonServerSenderContext(AMQPConnectionContext connection, |
| Sender sender, |
| AMQPSessionContext protonSession, |
| AMQPSessionCallback server, |
| SenderController senderController) { |
| super(); |
| this.controller = senderController; |
| this.connection = connection; |
| this.sender = sender; |
| this.protonSession = protonSession; |
| this.sessionSPI = server; |
| this.executeDelivery = this::executeDelivery; |
| amqpTreatRejectAsUnmodifiedDeliveryFailed = this.connection.getProtocolManager() |
| .isAmqpTreatRejectAsUnmodifiedDeliveryFailed(); |
| } |
| |
| public ProtonServerSenderContext setBeforeDelivery(java.util.function.Consumer<? super MessageReference> beforeDelivery) { |
| this.beforeDelivery = beforeDelivery; |
| return this; |
| } |
| |
| public Object getBrokerConsumer() { |
| return brokerConsumer; |
| } |
| |
| @Override |
| public void onFlow(int currentCredits, boolean drain) { |
| connection.requireInHandler(); |
| |
| setupCredit(); |
| |
| ServerConsumerImpl serverConsumer = (ServerConsumerImpl) brokerConsumer; |
| if (drain) { |
| // If the draining is already running, then don't do anything |
| if (draining.compareAndSet(false, true)) { |
| final ProtonServerSenderContext plugSender = (ProtonServerSenderContext) serverConsumer.getProtocolContext(); |
| serverConsumer.forceDelivery(1, new Runnable() { |
| @Override |
| public void run() { |
| try { |
| connection.runNow(() -> { |
| plugSender.reportDrained(); |
| setupCredit(); |
| }); |
| } finally { |
| draining.set(false); |
| } |
| } |
| }); |
| } |
| } else { |
| serverConsumer.receiveCredits(-1); |
| } |
| } |
| |
| public boolean hasCredits() { |
| if (hasLarge) { |
| // we will resume accepting once the large message is finished |
| return false; |
| } |
| |
| if (!connection.flowControl(onflowControlReady)) { |
| return false; |
| } |
| |
| synchronized (creditsLock) { |
| return credits > 0 && sender.getLocalState() != EndpointState.CLOSED; |
| } |
| } |
| |
| private void setupCredit() { |
| synchronized (creditsLock) { |
| this.credits = sender.getCredit() - pending.get(); |
| if (credits < 0) { |
| credits = 0; |
| } |
| } |
| } |
| |
| public Sender getSender() { |
| return sender; |
| } |
| |
| /* |
| * start the session |
| */ |
| public void start() throws ActiveMQAMQPException { |
| sessionSPI.start(); |
| |
| try { |
| // to do whatever you need to make the broker start sending messages to the consumer |
| // this could be null if a link reattach has happened |
| if (brokerConsumer != null) { |
| sessionSPI.startSender(brokerConsumer); |
| } |
| } catch (Exception e) { |
| throw ActiveMQAMQPProtocolMessageBundle.BUNDLE.errorStartingConsumer(e.getMessage()); |
| } |
| } |
| |
| |
| /** |
| * create the actual underlying ActiveMQ Artemis Server Consumer |
| */ |
| @Override |
| public void initialize() throws Exception { |
| super.initialize(); |
| |
| if (controller == null) { |
| controller = new DefaultController(sessionSPI); |
| } |
| |
| try { |
| brokerConsumer = controller.init(this); |
| onflowControlReady = brokerConsumer::promptDelivery; |
| } catch (ActiveMQAMQPResourceLimitExceededException e1) { |
| throw e1; |
| } catch (ActiveMQSecurityException e) { |
| throw ActiveMQAMQPProtocolMessageBundle.BUNDLE.securityErrorCreatingConsumer(e.getMessage()); |
| } catch (ActiveMQQueueMaxConsumerLimitReached e) { |
| throw ActiveMQAMQPProtocolMessageBundle.BUNDLE.errorCreatingConsumer(e.getMessage()); |
| } catch (ActiveMQException e) { |
| throw e; |
| } catch (Exception e) { |
| throw ActiveMQAMQPProtocolMessageBundle.BUNDLE.errorCreatingConsumer(e.getMessage()); |
| } |
| } |
| |
| protected String getClientId() { |
| return connection.getRemoteContainer(); |
| } |
| |
| /* |
| * close the session |
| */ |
| @Override |
| public void close(ErrorCondition condition) throws ActiveMQAMQPException { |
| closed = true; |
| if (condition != null) { |
| sender.setCondition(condition); |
| } |
| protonSession.removeSender(sender); |
| |
| connection.runLater(() -> { |
| sender.close(); |
| try { |
| sessionSPI.closeSender(brokerConsumer); |
| } catch (Exception e) { |
| log.warn(e.getMessage(), e); |
| } |
| sender.close(); |
| connection.flush(); |
| }); |
| } |
| |
| /* |
| * close the session |
| */ |
| @Override |
| public void close(boolean remoteLinkClose) throws ActiveMQAMQPException { |
| // we need to mark closed first to make sure no more adds are accepted |
| closed = true; |
| |
| // MessageReferences are sent to the Connection executor (Netty Loop) |
| // as a result the returning references have to be done later after they |
| // had their chance to finish and clear the runnable |
| connection.runLater(() -> { |
| try { |
| internalClose(remoteLinkClose); |
| } catch (Exception e) { |
| log.warn(e.getMessage(), e); |
| } |
| }); |
| } |
| |
| private void internalClose(boolean remoteLinkClose) throws ActiveMQAMQPException { |
| try { |
| protonSession.removeSender(sender); |
| sessionSPI.closeSender(brokerConsumer); |
| // if this is a link close rather than a connection close or detach, we need to delete |
| // any durable resources for say pub subs |
| if (remoteLinkClose) { |
| controller.close(); |
| |
| } |
| } catch (Exception e) { |
| log.warn(e.getMessage(), e); |
| throw new ActiveMQAMQPInternalErrorException(e.getMessage()); |
| } |
| } |
| |
| @Override |
| public void onMessage(Delivery delivery) throws ActiveMQAMQPException { |
| if (closed) { |
| return; |
| } |
| |
| OperationContext oldContext = sessionSPI.recoverContext(); |
| |
| try { |
| Message message = ((MessageReference) delivery.getContext()).getMessage(); |
| DeliveryState remoteState = delivery.getRemoteState(); |
| |
| if (remoteState != null && remoteState.getType() == DeliveryStateType.Accepted) { |
| // this can happen in the twice ack mode, that is the receiver accepts and settles separately |
| // acking again would show an exception but would have no negative effect but best to handle anyway. |
| if (!delivery.isSettled()) { |
| doAck(message); |
| |
| delivery.settle(); |
| } |
| } else { |
| handleExtendedDeliveryOutcomes(message, delivery, remoteState); |
| } |
| |
| if (!preSettle) { |
| protonSession.replaceTag(delivery.getTag()); |
| } |
| } finally { |
| sessionSPI.afterIO(connectionFlusher); |
| sessionSPI.resetContext(oldContext); |
| } |
| } |
| |
| protected void doAck(Message message) throws ActiveMQAMQPIllegalStateException { |
| // we have to individual ack as we can't guarantee we will get the delivery updates |
| // (including acks) in order from dealer, a performance hit but a must |
| try { |
| sessionSPI.ack(null, brokerConsumer, message); |
| } catch (Exception e) { |
| log.warn(e.toString(), e); |
| throw ActiveMQAMQPProtocolMessageBundle.BUNDLE.errorAcknowledgingMessage(message.toString(), e.getMessage()); |
| } |
| } |
| |
| private boolean handleExtendedDeliveryOutcomes(Message message, Delivery delivery, DeliveryState remoteState) throws ActiveMQAMQPException { |
| boolean settleImmediate = true; |
| boolean handled = true; |
| |
| if (remoteState == null) { |
| log.debug("Received null disposition for delivery update: " + remoteState); |
| return true; |
| } |
| |
| switch (remoteState.getType()) { |
| case Transactional: |
| // When the message arrives with a TransactionState disposition the ack should |
| // enlist the message into the transaction associated with the given txn ID. |
| TransactionalState txState = (TransactionalState) remoteState; |
| ProtonTransactionImpl tx = (ProtonTransactionImpl) this.sessionSPI.getTransaction(txState.getTxnId(), false); |
| |
| if (txState.getOutcome() != null) { |
| settleImmediate = false; |
| Outcome outcome = txState.getOutcome(); |
| if (outcome instanceof Accepted) { |
| if (!delivery.remotelySettled()) { |
| TransactionalState txAccepted = new TransactionalState(); |
| txAccepted.setOutcome(Accepted.getInstance()); |
| txAccepted.setTxnId(txState.getTxnId()); |
| delivery.disposition(txAccepted); |
| } |
| // we have to individual ack as we can't guarantee we will get the delivery |
| // (including acks) in order from dealer, a performance hit but a must |
| try { |
| sessionSPI.ack(tx, brokerConsumer, message); |
| tx.addDelivery(delivery, this); |
| } catch (Exception e) { |
| throw ActiveMQAMQPProtocolMessageBundle.BUNDLE.errorAcknowledgingMessage(message.toString(), e.getMessage()); |
| } |
| } |
| } |
| break; |
| case Released: |
| try { |
| sessionSPI.cancel(brokerConsumer, message, false); |
| } catch (Exception e) { |
| throw ActiveMQAMQPProtocolMessageBundle.BUNDLE.errorCancellingMessage(message.toString(), e.getMessage()); |
| } |
| break; |
| case Rejected: |
| try { |
| if (amqpTreatRejectAsUnmodifiedDeliveryFailed) { |
| // We could be more discriminating - for instance check for AmqpError#RESOURCE_LIMIT_EXCEEDED |
| sessionSPI.cancel(brokerConsumer, message, true); |
| } else { |
| sessionSPI.reject(brokerConsumer, message); |
| } |
| } catch (Exception e) { |
| throw ActiveMQAMQPProtocolMessageBundle.BUNDLE.errorCancellingMessage(message.toString(), e.getMessage()); |
| } |
| break; |
| case Modified: |
| try { |
| Modified modification = (Modified) remoteState; |
| |
| if (Boolean.TRUE.equals(modification.getUndeliverableHere())) { |
| message.rejectConsumer(brokerConsumer.sequentialID()); |
| } |
| |
| if (Boolean.TRUE.equals(modification.getDeliveryFailed())) { |
| sessionSPI.cancel(brokerConsumer, message, true); |
| } else { |
| sessionSPI.cancel(brokerConsumer, message, false); |
| } |
| } catch (Exception e) { |
| throw ActiveMQAMQPProtocolMessageBundle.BUNDLE.errorCancellingMessage(message.toString(), e.getMessage()); |
| } |
| break; |
| default: |
| log.debug("Received null or unknown disposition for delivery update: " + remoteState); |
| handled = false; |
| } |
| |
| if (settleImmediate) { |
| delivery.settle(); |
| } |
| |
| return handled; |
| } |
| |
| private final class ConnectionFlushIOCallback implements IOCallback { |
| @Override |
| public void done() { |
| connection.flush(); |
| } |
| |
| @Override |
| public void onError(int errorCode, String errorMessage) { |
| connection.flush(); |
| } |
| } |
| |
| public void settle(Delivery delivery) { |
| connection.requireInHandler(); |
| delivery.settle(); |
| } |
| |
| public synchronized void checkState() { |
| sessionSPI.resumeDelivery(brokerConsumer); |
| } |
| |
| /** |
| * handle an out going message from ActiveMQ Artemis, send via the Proton Sender |
| */ |
| public int deliverMessage(final MessageReference messageReference, final ServerConsumer consumer) throws Exception { |
| |
| if (closed) { |
| return 0; |
| } |
| |
| if (beforeDelivery != null) { |
| beforeDelivery.accept(messageReference); |
| } |
| |
| try { |
| synchronized (creditsLock) { |
| if (sender.getLocalState() == EndpointState.CLOSED) { |
| return 0; |
| } |
| pending.incrementAndGet(); |
| credits--; |
| } |
| |
| if (messageReference.getMessage() instanceof AMQPLargeMessage) { |
| hasLarge = true; |
| } |
| |
| if (messageReference instanceof Runnable && consumer.allowReferenceCallback()) { |
| messageReference.onDelivery(executeDelivery); |
| connection.runNow((Runnable) messageReference); |
| } else { |
| connection.runNow(() -> executeDelivery(messageReference)); |
| } |
| |
| // This is because on AMQP we only send messages based in credits, not bytes |
| return 1; |
| } finally { |
| |
| } |
| } |
| |
| private void executeDelivery(MessageReference messageReference) { |
| |
| try { |
| if (sender.getLocalState() == EndpointState.CLOSED) { |
| log.debug("Not delivering message " + messageReference + " as the sender is closed and credits were available, if you see too many of these it means clients are issuing credits and closing the connection with pending credits a lot of times"); |
| return; |
| } |
| AMQPMessage message = CoreAmqpConverter.checkAMQP(messageReference.getMessage(), sessionSPI.getStorageManager()); |
| |
| if (sessionSPI.invokeOutgoing(message, (ActiveMQProtonRemotingConnection) sessionSPI.getTransportConnection().getProtocolConnection()) != null) { |
| return; |
| } |
| if (message instanceof AMQPLargeMessage) { |
| deliverLarge(messageReference, (AMQPLargeMessage) message); |
| } else { |
| deliverStandard(messageReference, message); |
| } |
| |
| } catch (Exception e) { |
| log.warn(e.getMessage(), e); |
| brokerConsumer.errorProcessing(e, messageReference); |
| } |
| } |
| |
| private class LargeMessageDeliveryContext { |
| |
| LargeMessageDeliveryContext(MessageReference reference, AMQPLargeMessage message, Delivery delivery) { |
| this.position = 0L; |
| this.reference = reference; |
| this.message = message; |
| this.delivery = delivery; |
| } |
| |
| long position; |
| final MessageReference reference; |
| final AMQPLargeMessage message; |
| final Delivery delivery; |
| |
| void resume() { |
| connection.runNow(this::deliver); |
| } |
| |
| void deliver() { |
| |
| // This is discounting some bytes due to Transfer payload |
| int frameSize = protonSession.session.getConnection().getTransport().getOutboundFrameSizeLimit() - 50 - (delivery.getTag() != null ? delivery.getTag().length : 0); |
| |
| DeliveryAnnotations deliveryAnnotationsToEncode; |
| |
| message.checkReference(reference); |
| |
| if (reference.getProtocolData() != null && reference.getProtocolData() instanceof DeliveryAnnotations) { |
| deliveryAnnotationsToEncode = (DeliveryAnnotations)reference.getProtocolData(); |
| } else { |
| deliveryAnnotationsToEncode = null; |
| } |
| |
| LargeBodyReader context = message.getLargeBodyReader(); |
| try { |
| context.open(); |
| try { |
| context.position(position); |
| long bodySize = context.getSize(); |
| |
| ByteBuffer buf = ByteBuffer.allocate(frameSize); |
| |
| for (; position < bodySize; ) { |
| if (!connection.flowControl(this::resume)) { |
| context.close(); |
| return; |
| } |
| buf.clear(); |
| int size = 0; |
| |
| try { |
| if (position == 0) { |
| replaceInitialHeader(deliveryAnnotationsToEncode, context, WritableBuffer.ByteBufferWrapper.wrap(buf)); |
| } |
| size = context.readInto(buf); |
| |
| sender.send(new ReadableBuffer.ByteBufferReader(buf)); |
| position += size; |
| } catch (java.nio.BufferOverflowException overflowException) { |
| if (position == 0) { |
| if (log.isDebugEnabled()) { |
| log.debug("Delivery of message failed with an overFlowException, retrying again with expandable buffer"); |
| } |
| // on the very first packet, if the initial header was replaced with a much bigger header (re-encoding) |
| // we could recover the situation with a retry using an expandable buffer. |
| // this is tested on org.apache.activemq.artemis.tests.integration.amqp.AmqpMessageDivertsTest |
| size = retryInitialPacketWithExpandableBuffer(deliveryAnnotationsToEncode, context, buf); |
| } else { |
| // if this is not the position 0, something is going on |
| // we just forward the exception as this is not supposed to happen |
| throw overflowException; |
| } |
| } |
| |
| if (size > 0) { |
| |
| if (position < bodySize) { |
| connection.instantFlush(); |
| } |
| } |
| } |
| } finally { |
| context.close(); |
| } |
| |
| if (preSettle) { |
| // Presettled means the client implicitly accepts any delivery we send it. |
| try { |
| sessionSPI.ack(null, brokerConsumer, reference.getMessage()); |
| } catch (Exception e) { |
| log.debug(e.getMessage(), e); |
| } |
| delivery.settle(); |
| } else { |
| sender.advance(); |
| } |
| |
| connection.instantFlush(); |
| |
| synchronized (creditsLock) { |
| pending.decrementAndGet(); |
| } |
| |
| finishLargeMessage(); |
| } catch (Exception e) { |
| log.warn(e.getMessage(), e); |
| brokerConsumer.errorProcessing(e, reference); |
| } |
| } |
| |
| /** |
| * This is a retry logic when either the delivery annotations or re-encoded buffer is bigger than the frame size |
| * This will create one expandable buffer. |
| * It will then let Proton to do the framing correctly |
| */ |
| private int retryInitialPacketWithExpandableBuffer(DeliveryAnnotations deliveryAnnotationsToEncode, |
| LargeBodyReader context, |
| ByteBuffer buf) throws Exception { |
| int size; |
| buf.clear(); |
| // if the buffer overflow happened during the initial position |
| // this means the replaced headers are bigger then the frame size |
| // on this case we do with an expandable netty buffer |
| ByteBuf nettyBuffer = PooledByteBufAllocator.DEFAULT.buffer(AMQPMessageBrokerAccessor.getRemainingBodyPosition(message) * 2); |
| try { |
| replaceInitialHeader(deliveryAnnotationsToEncode, context, new NettyWritable(nettyBuffer)); |
| size = context.readInto(buf); |
| position += size; |
| |
| nettyBuffer.writeBytes(buf); |
| |
| ByteBuffer nioBuffer = nettyBuffer.nioBuffer(); |
| nioBuffer.position(nettyBuffer.writerIndex()); |
| nioBuffer = (ByteBuffer) nioBuffer.flip(); |
| sender.send(new ReadableBuffer.ByteBufferReader(nioBuffer)); |
| } finally { |
| nettyBuffer.release(); |
| } |
| return size; |
| } |
| |
| private int replaceInitialHeader(DeliveryAnnotations deliveryAnnotationsToEncode, |
| LargeBodyReader context, |
| WritableBuffer buf) throws Exception { |
| TLSEncode.getEncoder().setByteBuffer(buf); |
| try { |
| int proposedPosition = writeHeaderAndAnnotations(context, deliveryAnnotationsToEncode); |
| if (message.isReencoded()) { |
| proposedPosition = writeMessageAnnotationsPropertiesAndApplicationProperties(context, message); |
| } |
| |
| context.position(proposedPosition); |
| position = proposedPosition; |
| return (int)position; |
| } finally { |
| |
| TLSEncode.getEncoder().setByteBuffer((WritableBuffer)null); |
| } |
| } |
| |
| /** |
| * Write properties and application properties when the message is flagged as re-encoded. |
| */ |
| private int writeMessageAnnotationsPropertiesAndApplicationProperties(LargeBodyReader context, AMQPLargeMessage message) throws Exception { |
| int bodyPosition = AMQPMessageBrokerAccessor.getRemainingBodyPosition(message); |
| assert bodyPosition > 0; |
| writeMessageAnnotationsPropertiesAndApplicationPropertiesInternal(message); |
| return bodyPosition; |
| } |
| |
| private void writeMessageAnnotationsPropertiesAndApplicationPropertiesInternal(AMQPLargeMessage message) { |
| MessageAnnotations messageAnnotations = AMQPMessageBrokerAccessor.getDecodedMessageAnnotations(message); |
| |
| if (messageAnnotations != null) { |
| TLSEncode.getEncoder().writeObject(messageAnnotations); |
| } |
| |
| Properties amqpProperties = AMQPMessageBrokerAccessor.getCurrentProperties(message); |
| if (amqpProperties != null) { |
| TLSEncode.getEncoder().writeObject(amqpProperties); |
| } |
| |
| ApplicationProperties applicationProperties = AMQPMessageBrokerAccessor.getDecodedApplicationProperties(message); |
| |
| if (applicationProperties != null) { |
| TLSEncode.getEncoder().writeObject(applicationProperties); |
| } |
| } |
| |
| private int writeHeaderAndAnnotations(LargeBodyReader context, |
| DeliveryAnnotations deliveryAnnotationsToEncode) throws ActiveMQException { |
| Header header = AMQPMessageBrokerAccessor.getCurrentHeader(message); |
| if (header != null) { |
| TLSEncode.getEncoder().writeObject(header); |
| } |
| if (deliveryAnnotationsToEncode != null) { |
| TLSEncode.getEncoder().writeObject(deliveryAnnotationsToEncode); |
| } |
| return message.getPositionAfterDeliveryAnnotations(); |
| } |
| } |
| |
| private void finishLargeMessage() { |
| pendingLargeMessage = null; |
| hasLarge = false; |
| brokerConsumer.promptDelivery(); |
| } |
| |
| private void deliverLarge(MessageReference messageReference, AMQPLargeMessage message) { |
| |
| // we only need a tag if we are going to settle later |
| byte[] tag = preSettle ? new byte[0] : protonSession.getTag(); |
| |
| final Delivery delivery; |
| delivery = sender.delivery(tag, 0, tag.length); |
| delivery.setMessageFormat((int) message.getMessageFormat()); |
| delivery.setContext(messageReference); |
| |
| pendingLargeMessage = new LargeMessageDeliveryContext(messageReference, message, delivery); |
| pendingLargeMessage.deliver(); |
| |
| } |
| |
| private void deliverStandard(MessageReference messageReference, AMQPMessage message) { |
| // Let the Message decide how to present the message bytes |
| ReadableBuffer sendBuffer = message.getSendBuffer(messageReference.getDeliveryCount(), messageReference); |
| // we only need a tag if we are going to settle later |
| byte[] tag = preSettle ? new byte[0] : protonSession.getTag(); |
| |
| boolean releaseRequired = sendBuffer instanceof NettyReadable; |
| final Delivery delivery; |
| delivery = sender.delivery(tag, 0, tag.length); |
| delivery.setMessageFormat((int) message.getMessageFormat()); |
| delivery.setContext(messageReference); |
| |
| try { |
| |
| if (releaseRequired) { |
| sender.send(sendBuffer); |
| // Above send copied, so release now if needed |
| releaseRequired = false; |
| ((NettyReadable) sendBuffer).getByteBuf().release(); |
| } else { |
| // Don't have pooled content, no need to release or copy. |
| sender.sendNoCopy(sendBuffer); |
| } |
| |
| if (preSettle) { |
| // Presettled means the client implicitly accepts any delivery we send it. |
| try { |
| sessionSPI.ack(null, brokerConsumer, messageReference.getMessage()); |
| } catch (Exception e) { |
| log.debug(e.getMessage(), e); |
| } |
| delivery.settle(); |
| } else { |
| sender.advance(); |
| } |
| |
| connection.flush(); |
| } finally { |
| synchronized (creditsLock) { |
| pending.decrementAndGet(); |
| } |
| if (releaseRequired) { |
| ((NettyReadable) sendBuffer).getByteBuf().release(); |
| } |
| } |
| } |
| |
| private static boolean hasCapabilities(Symbol symbol, Source source) { |
| if (source != null) { |
| if (source.getCapabilities() != null) { |
| for (Symbol cap : source.getCapabilities()) { |
| if (symbol.equals(cap)) { |
| return true; |
| } |
| } |
| } |
| } |
| return false; |
| } |
| |
| private static boolean hasRemoteDesiredCapability(Link link, Symbol capability) { |
| Symbol[] remoteDesiredCapabilities = link.getRemoteDesiredCapabilities(); |
| if (remoteDesiredCapabilities != null) { |
| for (Symbol cap : remoteDesiredCapabilities) { |
| if (capability.equals(cap)) { |
| return true; |
| } |
| } |
| } |
| return false; |
| } |
| |
| private static SimpleString createQueueName(boolean useCoreSubscriptionNaming, |
| String clientId, |
| String pubId, |
| boolean shared, |
| boolean global, |
| boolean isVolatile) { |
| if (useCoreSubscriptionNaming) { |
| final boolean durable = !isVolatile; |
| final String subscriptionName = pubId.contains("|") ? pubId.split("\\|")[0] : pubId; |
| final String clientID = clientId == null || clientId.isEmpty() || global ? null : clientId; |
| return DestinationUtil.createQueueNameForSubscription(durable, clientID, subscriptionName); |
| } else { |
| String queue = clientId == null || clientId.isEmpty() || global ? pubId : clientId + "." + pubId; |
| if (shared) { |
| if (queue.contains("|")) { |
| queue = queue.split("\\|")[0]; |
| } |
| if (isVolatile) { |
| queue += ":shared-volatile"; |
| } |
| if (global) { |
| queue += ":global"; |
| } |
| } |
| return SimpleString.toSimpleString(queue); |
| } |
| } |
| |
| /** |
| * Update link state to reflect that the previous drain attempt has completed. |
| */ |
| public void reportDrained() { |
| connection.requireInHandler(); |
| sender.drained(); |
| connection.instantFlush(); |
| } |
| |
| public AMQPSessionContext getSessionContext() { |
| return protonSession; |
| } |
| |
| class DefaultController implements SenderController { |
| |
| |
| private boolean shared = false; |
| boolean global = false; |
| boolean multicast; |
| final AMQPSessionCallback sessionSPI; |
| SimpleString queue = null; |
| SimpleString tempQueueName; |
| String selector; |
| |
| private final RoutingType defaultRoutingType = RoutingType.ANYCAST; |
| private RoutingType routingTypeToUse = RoutingType.ANYCAST; |
| |
| private boolean isVolatile = false; |
| |
| DefaultController(AMQPSessionCallback sessionSPI) { |
| this.sessionSPI = sessionSPI; |
| |
| } |
| |
| @Override |
| public Consumer init(ProtonServerSenderContext senderContext) throws Exception { |
| Source source = (Source) sender.getRemoteSource(); |
| final Map<Symbol, Object> supportedFilters = new HashMap<>(); |
| |
| // Match the settlement mode of the remote instead of relying on the default of MIXED. |
| sender.setSenderSettleMode(sender.getRemoteSenderSettleMode()); |
| |
| // We don't currently support SECOND so enforce that the answer is anlways FIRST |
| sender.setReceiverSettleMode(ReceiverSettleMode.FIRST); |
| |
| if (source != null) { |
| // We look for message selectors on every receiver, while in other cases we might only |
| // consume the filter depending on the subscription type. |
| Map.Entry<Symbol, DescribedType> filter = AmqpSupport.findFilter(source.getFilter(), AmqpSupport.JMS_SELECTOR_FILTER_IDS); |
| if (filter != null) { |
| selector = filter.getValue().getDescribed().toString(); |
| // Validate the Selector. |
| try { |
| SelectorParser.parse(selector); |
| } catch (FilterException e) { |
| throw new ActiveMQAMQPException(AmqpError.INVALID_FIELD, "Invalid filter", ActiveMQExceptionType.INVALID_FILTER_EXPRESSION); |
| } |
| |
| supportedFilters.put(filter.getKey(), filter.getValue()); |
| } |
| } |
| |
| if (source == null) { |
| // Attempt to recover a previous subscription happens when a link reattach happens on a |
| // subscription queue |
| String clientId = getClientId(); |
| String pubId = sender.getName(); |
| global = hasRemoteDesiredCapability(sender, GLOBAL); |
| shared = hasRemoteDesiredCapability(sender, SHARED); |
| queue = createQueueName(connection.isUseCoreSubscriptionNaming(), clientId, pubId, true, global, false); |
| QueueQueryResult result = sessionSPI.queueQuery(queue, RoutingType.MULTICAST, false); |
| multicast = true; |
| routingTypeToUse = RoutingType.MULTICAST; |
| |
| // Once confirmed that the address exists we need to return a Source that reflects |
| // the lifetime policy and capabilities of the new subscription. |
| if (result.isExists()) { |
| source = new org.apache.qpid.proton.amqp.messaging.Source(); |
| source.setAddress(queue.toString()); |
| source.setDurable(TerminusDurability.UNSETTLED_STATE); |
| source.setExpiryPolicy(TerminusExpiryPolicy.NEVER); |
| source.setDistributionMode(COPY); |
| source.setCapabilities(TOPIC); |
| |
| SimpleString filterString = result.getFilterString(); |
| if (filterString != null) { |
| selector = filterString.toString(); |
| boolean noLocal = false; |
| |
| String remoteContainerId = sender.getSession().getConnection().getRemoteContainer(); |
| String noLocalFilter = MessageUtil.CONNECTION_ID_PROPERTY_NAME.toString() + "<>'" + remoteContainerId + "'"; |
| |
| if (selector.endsWith(noLocalFilter)) { |
| if (selector.length() > noLocalFilter.length()) { |
| noLocalFilter = " AND " + noLocalFilter; |
| selector = selector.substring(0, selector.length() - noLocalFilter.length()); |
| } else { |
| selector = null; |
| } |
| |
| noLocal = true; |
| } |
| |
| if (noLocal) { |
| supportedFilters.put(AmqpSupport.NO_LOCAL_NAME, AmqpNoLocalFilter.NO_LOCAL); |
| } |
| |
| if (selector != null && !selector.trim().isEmpty()) { |
| supportedFilters.put(AmqpSupport.JMS_SELECTOR_NAME, new AmqpJmsSelectorFilter(selector)); |
| } |
| } |
| |
| sender.setSource(source); |
| } else { |
| throw new ActiveMQAMQPNotFoundException("Unknown subscription link: " + sender.getName()); |
| } |
| } else if (source.getDynamic()) { |
| // if dynamic we have to create the node (queue) and set the address on the target, the |
| // node is temporary and will be deleted on closing of the session |
| queue = SimpleString.toSimpleString(java.util.UUID.randomUUID().toString()); |
| tempQueueName = queue; |
| try { |
| sessionSPI.createTemporaryQueue(queue, RoutingType.ANYCAST); |
| // protonSession.getServerSession().createQueue(queue, queue, null, true, false); |
| } catch (Exception e) { |
| throw ActiveMQAMQPProtocolMessageBundle.BUNDLE.errorCreatingTemporaryQueue(e.getMessage()); |
| } |
| source.setAddress(queue.toString()); |
| } else { |
| SimpleString addressToUse; |
| SimpleString queueNameToUse = null; |
| shared = hasCapabilities(SHARED, source); |
| global = hasCapabilities(GLOBAL, source); |
| |
| //find out if we have an address made up of the address and queue name, if yes then set queue name |
| if (CompositeAddress.isFullyQualified(source.getAddress())) { |
| addressToUse = SimpleString.toSimpleString(CompositeAddress.extractAddressName(source.getAddress())); |
| queueNameToUse = SimpleString.toSimpleString(CompositeAddress.extractQueueName(source.getAddress())); |
| } else { |
| addressToUse = SimpleString.toSimpleString(source.getAddress()); |
| } |
| //check to see if the client has defined how we act |
| boolean clientDefined = hasCapabilities(TOPIC, source) || hasCapabilities(QUEUE, source); |
| if (clientDefined) { |
| multicast = hasCapabilities(TOPIC, source); |
| AddressQueryResult addressQueryResult = null; |
| try { |
| addressQueryResult = sessionSPI.addressQuery(addressToUse, multicast ? RoutingType.MULTICAST : RoutingType.ANYCAST, true); |
| } catch (ActiveMQSecurityException e) { |
| throw ActiveMQAMQPProtocolMessageBundle.BUNDLE.securityErrorCreatingConsumer(e.getMessage()); |
| } catch (ActiveMQAMQPException e) { |
| throw e; |
| } catch (Exception e) { |
| throw new ActiveMQAMQPInternalErrorException(e.getMessage(), e); |
| } |
| |
| if (!addressQueryResult.isExists()) { |
| throw ActiveMQAMQPProtocolMessageBundle.BUNDLE.sourceAddressDoesntExist(); |
| } |
| |
| Set<RoutingType> routingTypes = addressQueryResult.getRoutingTypes(); |
| |
| //if the client defines 1 routing type and the broker another then throw an exception |
| if (multicast && !routingTypes.contains(RoutingType.MULTICAST)) { |
| throw new ActiveMQAMQPIllegalStateException("Address " + addressToUse + " is not configured for topic support"); |
| } else if (!multicast && !routingTypes.contains(RoutingType.ANYCAST)) { |
| //if client specifies fully qualified name that's allowed, don't throw exception. |
| if (queueNameToUse == null) { |
| throw new ActiveMQAMQPIllegalStateException("Address " + addressToUse + " is not configured for queue support"); |
| } |
| } |
| } else { |
| // if not we look up the address |
| AddressQueryResult addressQueryResult = null; |
| try { |
| addressQueryResult = sessionSPI.addressQuery(addressToUse, defaultRoutingType, true); |
| } catch (ActiveMQSecurityException e) { |
| throw ActiveMQAMQPProtocolMessageBundle.BUNDLE.securityErrorCreatingConsumer(e.getMessage()); |
| } catch (ActiveMQAMQPException e) { |
| throw e; |
| } catch (Exception e) { |
| throw new ActiveMQAMQPInternalErrorException(e.getMessage(), e); |
| } |
| |
| if (!addressQueryResult.isExists()) { |
| throw ActiveMQAMQPProtocolMessageBundle.BUNDLE.sourceAddressDoesntExist(); |
| } |
| |
| Set<RoutingType> routingTypes = addressQueryResult.getRoutingTypes(); |
| if (routingTypes.contains(RoutingType.MULTICAST) && routingTypes.size() == 1) { |
| multicast = true; |
| } else { |
| //todo add some checks if both routing types are supported |
| multicast = false; |
| } |
| } |
| routingTypeToUse = multicast ? RoutingType.MULTICAST : RoutingType.ANYCAST; |
| // if not dynamic then we use the target's address as the address to forward the |
| // messages to, however there has to be a queue bound to it so we need to check this. |
| if (multicast) { |
| Map.Entry<Symbol, DescribedType> filter = AmqpSupport.findFilter(source.getFilter(), AmqpSupport.NO_LOCAL_FILTER_IDS); |
| if (filter != null) { |
| String remoteContainerId = sender.getSession().getConnection().getRemoteContainer(); |
| String noLocalFilter = MessageUtil.CONNECTION_ID_PROPERTY_NAME.toString() + "<>'" + remoteContainerId + "'"; |
| if (selector != null) { |
| selector += " AND " + noLocalFilter; |
| } else { |
| selector = noLocalFilter; |
| } |
| |
| supportedFilters.put(filter.getKey(), filter.getValue()); |
| } |
| |
| queue = getMatchingQueue(queueNameToUse, addressToUse, RoutingType.MULTICAST); |
| SimpleString simpleStringSelector = SimpleString.toSimpleString(selector); |
| |
| //if the address specifies a broker configured queue then we always use this, treat it as a queue |
| if (queue != null) { |
| multicast = false; |
| } else if (TerminusDurability.UNSETTLED_STATE.equals(source.getDurable()) || TerminusDurability.CONFIGURATION.equals(source.getDurable())) { |
| |
| // if we are a subscription and durable create a durable queue using the container |
| // id and link name |
| String clientId = getClientId(); |
| String pubId = sender.getName(); |
| queue = createQueueName(connection.isUseCoreSubscriptionNaming(), clientId, pubId, shared, global, false); |
| QueueQueryResult result = sessionSPI.queueQuery(queue, routingTypeToUse, false); |
| if (result.isExists()) { |
| /* |
| * If a client reattaches to a durable subscription with a different filter or address then we must |
| * recreate the queue (JMS semantics). However, if the corresponding queue is managed via the |
| * configuration then we don't want to change it |
| */ |
| if (!result.isConfigurationManaged() && (!Objects.equals(result.getAddress(), addressToUse) || !Objects.equals(result.getFilterString(), simpleStringSelector))) { |
| |
| if (result.getConsumerCount() == 0) { |
| sessionSPI.deleteQueue(queue); |
| sessionSPI.createUnsharedDurableQueue(addressToUse, RoutingType.MULTICAST, queue, simpleStringSelector); |
| } else { |
| throw new ActiveMQAMQPIllegalStateException("Unable to recreate subscription, consumers already exist"); |
| } |
| } |
| } else { |
| if (shared) { |
| sessionSPI.createSharedDurableQueue(addressToUse, RoutingType.MULTICAST, queue, simpleStringSelector); |
| } else { |
| sessionSPI.createUnsharedDurableQueue(addressToUse, RoutingType.MULTICAST, queue, simpleStringSelector); |
| } |
| } |
| } else { |
| // otherwise we are a volatile subscription |
| isVolatile = true; |
| if (shared && sender.getName() != null) { |
| queue = createQueueName(connection.isUseCoreSubscriptionNaming(), getClientId(), sender.getName(), shared, global, isVolatile); |
| QueueQueryResult result = sessionSPI.queueQuery(queue, routingTypeToUse, false); |
| if ((!result.isExists() || !Objects.equals(result.getAddress(), addressToUse) || !Objects.equals(result.getFilterString(), simpleStringSelector)) && !result.isConfigurationManaged()) { |
| sessionSPI.createSharedVolatileQueue(addressToUse, RoutingType.MULTICAST, queue, simpleStringSelector); |
| } |
| } else { |
| queue = SimpleString.toSimpleString(java.util.UUID.randomUUID().toString()); |
| tempQueueName = queue; |
| try { |
| sessionSPI.createTemporaryQueue(addressToUse, queue, RoutingType.MULTICAST, simpleStringSelector); |
| } catch (Exception e) { |
| throw ActiveMQAMQPProtocolMessageBundle.BUNDLE.errorCreatingTemporaryQueue(e.getMessage()); |
| } |
| } |
| } |
| } else { |
| if (queueNameToUse != null) { |
| //a queue consumer can receive from a multicast queue if it uses a fully qualified name |
| //setting routingType to null means do not check the routingType against the Queue's routing type. |
| routingTypeToUse = null; |
| SimpleString matchingAnycastQueue = getMatchingQueue(queueNameToUse, addressToUse, null); |
| if (matchingAnycastQueue != null) { |
| queue = matchingAnycastQueue; |
| } else { |
| throw ActiveMQAMQPProtocolMessageBundle.BUNDLE.sourceAddressDoesntExist(); |
| } |
| } else { |
| SimpleString matchingAnycastQueue = sessionSPI.getMatchingQueue(addressToUse, RoutingType.ANYCAST); |
| if (matchingAnycastQueue != null) { |
| queue = matchingAnycastQueue; |
| } else { |
| queue = addressToUse; |
| } |
| } |
| |
| } |
| |
| if (queue == null) { |
| throw ActiveMQAMQPProtocolMessageBundle.BUNDLE.sourceAddressNotSet(); |
| } |
| |
| try { |
| if (!sessionSPI.queueQuery(queue, routingTypeToUse, !multicast).isExists()) { |
| throw ActiveMQAMQPProtocolMessageBundle.BUNDLE.sourceAddressDoesntExist(); |
| } |
| } catch (ActiveMQAMQPNotFoundException e) { |
| throw e; |
| } catch (Exception e) { |
| throw new ActiveMQAMQPInternalErrorException(e.getMessage(), e); |
| } |
| } |
| |
| // Detect if sender is in pre-settle mode. |
| preSettle = sender.getRemoteSenderSettleMode() == SenderSettleMode.SETTLED; |
| |
| // We need to update the source with any filters we support otherwise the client |
| // is free to consider the attach as having failed if we don't send back what we |
| // do support or if we send something we don't support the client won't know we |
| // have not honored what it asked for. |
| source.setFilter(supportedFilters.isEmpty() ? null : supportedFilters); |
| |
| boolean browseOnly = !multicast && source.getDistributionMode() != null && source.getDistributionMode().equals(COPY); |
| |
| return (Consumer) sessionSPI.createSender(senderContext, queue, multicast ? null : selector, browseOnly); |
| } |
| |
| |
| private SimpleString getMatchingQueue(SimpleString queueName, SimpleString address, RoutingType routingType) throws Exception { |
| if (queueName != null) { |
| QueueQueryResult result = sessionSPI.queueQuery(CompositeAddress.toFullyQualified(address, queueName), routingType, true); |
| if (!result.isExists()) { |
| throw new ActiveMQAMQPNotFoundException("Queue: '" + queueName + "' does not exist"); |
| } else { |
| if (!result.getAddress().equals(address)) { |
| throw new ActiveMQAMQPNotFoundException("Queue: '" + queueName + "' does not exist for address '" + address + "'"); |
| } |
| return sessionSPI.getMatchingQueue(address, queueName, routingType); |
| } |
| } |
| return null; |
| } |
| |
| |
| @Override |
| public void close() throws Exception { |
| Source source = (Source) sender.getSource(); |
| if (source != null && source.getAddress() != null && multicast) { |
| SimpleString queueName = SimpleString.toSimpleString(source.getAddress()); |
| QueueQueryResult result = sessionSPI.queueQuery(queueName, routingTypeToUse, false); |
| if (result.isExists() && source.getDynamic()) { |
| sessionSPI.deleteQueue(queueName); |
| } else { |
| if (source.getDurable() == TerminusDurability.NONE && tempQueueName != null && (source.getExpiryPolicy() == TerminusExpiryPolicy.LINK_DETACH || source.getExpiryPolicy() == TerminusExpiryPolicy.SESSION_END)) { |
| sessionSPI.removeTemporaryQueue(tempQueueName); |
| } else { |
| String clientId = getClientId(); |
| String pubId = sender.getName(); |
| if (pubId.contains("|")) { |
| pubId = pubId.split("\\|")[0]; |
| } |
| SimpleString queue = createQueueName(connection.isUseCoreSubscriptionNaming(), clientId, pubId, shared, global, isVolatile); |
| result = sessionSPI.queueQuery(queue, multicast ? RoutingType.MULTICAST : RoutingType.ANYCAST, false); |
| //only delete if it isn't volatile and has no consumers |
| if (result.isExists() && !isVolatile && result.getConsumerCount() == 0) { |
| sessionSPI.deleteQueue(queue); |
| } |
| } |
| } |
| } else if (source != null && source.getDynamic() && (source.getExpiryPolicy() == TerminusExpiryPolicy.LINK_DETACH || source.getExpiryPolicy() == TerminusExpiryPolicy.SESSION_END)) { |
| try { |
| sessionSPI.removeTemporaryQueue(SimpleString.toSimpleString(source.getAddress())); |
| } catch (Exception e) { |
| //ignore on close, its temp anyway and will be removed later |
| } |
| } |
| } |
| |
| } |
| } |
