| /* |
| * |
| * 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.qpid.server.protocol.v1_0; |
| |
| import java.security.AccessControlException; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collections; |
| import java.util.EnumSet; |
| import java.util.HashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.ConcurrentLinkedQueue; |
| import java.util.regex.Pattern; |
| |
| import com.google.common.util.concurrent.ListenableFuture; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import org.apache.qpid.server.consumer.ConsumerOption; |
| import org.apache.qpid.server.filter.FilterManager; |
| import org.apache.qpid.server.filter.JMSSelectorFilter; |
| import org.apache.qpid.server.filter.SelectorParsingException; |
| import org.apache.qpid.server.filter.selector.ParseException; |
| import org.apache.qpid.server.filter.selector.TokenMgrError; |
| import org.apache.qpid.server.message.MessageInstance; |
| import org.apache.qpid.server.message.MessageInstanceConsumer; |
| import org.apache.qpid.server.message.MessageSource; |
| import org.apache.qpid.server.model.NamedAddressSpace; |
| import org.apache.qpid.server.model.NotFoundException; |
| import org.apache.qpid.server.model.Queue; |
| import org.apache.qpid.server.protocol.v1_0.type.AmqpErrorException; |
| import org.apache.qpid.server.protocol.v1_0.type.Binary; |
| import org.apache.qpid.server.protocol.v1_0.type.DeliveryState; |
| import org.apache.qpid.server.protocol.v1_0.type.Outcome; |
| import org.apache.qpid.server.protocol.v1_0.type.Symbol; |
| import org.apache.qpid.server.protocol.v1_0.type.UnsignedInteger; |
| import org.apache.qpid.server.protocol.v1_0.type.messaging.Accepted; |
| import org.apache.qpid.server.protocol.v1_0.type.messaging.Filter; |
| import org.apache.qpid.server.protocol.v1_0.type.messaging.Modified; |
| import org.apache.qpid.server.protocol.v1_0.type.messaging.NoLocalFilter; |
| import org.apache.qpid.server.protocol.v1_0.type.messaging.Rejected; |
| import org.apache.qpid.server.protocol.v1_0.type.messaging.Released; |
| import org.apache.qpid.server.protocol.v1_0.type.messaging.Source; |
| import org.apache.qpid.server.protocol.v1_0.type.messaging.StdDistMode; |
| import org.apache.qpid.server.protocol.v1_0.type.messaging.Target; |
| import org.apache.qpid.server.protocol.v1_0.type.messaging.TerminusDurability; |
| import org.apache.qpid.server.protocol.v1_0.type.messaging.TerminusExpiryPolicy; |
| import org.apache.qpid.server.protocol.v1_0.type.transaction.TransactionError; |
| import org.apache.qpid.server.protocol.v1_0.type.transport.AmqpError; |
| import org.apache.qpid.server.protocol.v1_0.type.transport.Attach; |
| import org.apache.qpid.server.protocol.v1_0.type.transport.Detach; |
| import org.apache.qpid.server.protocol.v1_0.type.transport.Error; |
| import org.apache.qpid.server.protocol.v1_0.type.transport.Flow; |
| import org.apache.qpid.server.protocol.v1_0.type.transport.Role; |
| import org.apache.qpid.server.protocol.v1_0.type.transport.Transfer; |
| import org.apache.qpid.server.txn.AsyncAutoCommitTransaction; |
| import org.apache.qpid.server.txn.AsyncCommand; |
| import org.apache.qpid.server.txn.AutoCommitTransaction; |
| import org.apache.qpid.server.txn.ServerTransaction; |
| import org.apache.qpid.server.util.Action; |
| import org.apache.qpid.server.util.ConnectionScopedRuntimeException; |
| import org.apache.qpid.server.virtualhost.LinkRegistryModel; |
| import org.apache.qpid.server.virtualhost.QueueManagingVirtualHost; |
| |
| public class SendingLinkEndpoint extends AbstractLinkEndpoint<Source, Target> |
| implements AsyncAutoCommitTransaction.FutureRecorder |
| { |
| private static final Logger LOGGER = LoggerFactory.getLogger(SendingLinkEndpoint.class); |
| private static final Symbol PRIORITY = Symbol.valueOf("priority"); |
| private static final Pattern ANY_CONTAINER_ID = Pattern.compile(".*"); |
| |
| private final List<Binary> _resumeAcceptedTransfers = new ArrayList<>(); |
| private final List<MessageInstance> _resumeFullTransfers = new ArrayList<>(); |
| private final Map<Binary, OutgoingDelivery> _unsettled = new ConcurrentHashMap<>(); |
| private final AsyncAutoCommitTransaction _asyncAutoCommitTransaction; |
| private final java.util.Queue<AsyncCommand> _unfinishedCommandsQueue = new ConcurrentLinkedQueue<>(); |
| |
| // TODO: QPID-7845 : remove after implementation of link resuming |
| private final Action<Session_1_0> _cleanUpUnsettledDeliveryTask = object -> cleanUpUnsettledDeliveries(); |
| |
| private volatile Binary _transactionId; |
| private volatile Integer _priority; |
| private volatile boolean _draining = false; |
| private volatile SendingDestination _destination; |
| private volatile EnumSet<ConsumerOption> _consumerOptions; |
| private volatile FilterManager _consumerFilters; |
| private volatile ConsumerTarget_1_0 _consumerTarget; |
| private volatile MessageInstanceConsumer<ConsumerTarget_1_0> _consumer; |
| |
| public SendingLinkEndpoint(final Session_1_0 session, final LinkImpl<Source, Target> link) |
| { |
| super(session, link); |
| setDeliveryCount(new SequenceNumber(0)); |
| setAvailable(UnsignedInteger.valueOf(0)); |
| setCapabilities(Collections.singletonList(AMQPConnection_1_0.SHARED_SUBSCRIPTIONS)); |
| _asyncAutoCommitTransaction = |
| new AsyncAutoCommitTransaction(getSession().getConnection().getAddressSpace().getMessageStore(), this); |
| } |
| |
| @Override |
| public void start() |
| { |
| } |
| |
| private void prepareConsumerOptionsAndFilters(final SendingDestination destination) throws AmqpErrorException |
| { |
| // TODO QPID-7952: this method might modify the source. this is not good encapsulation. furthermore if it does so then it should inform the link/linkregistry about it! |
| _destination = destination; |
| final Source source = getSource(); |
| |
| EnumSet<ConsumerOption> options = EnumSet.noneOf(ConsumerOption.class); |
| |
| boolean noLocal = false; |
| JMSSelectorFilter messageFilter = null; |
| |
| if(destination instanceof ExchangeSendingDestination) |
| { |
| options.add(ConsumerOption.ACQUIRES); |
| options.add(ConsumerOption.SEES_REQUEUES); |
| } |
| else if(destination instanceof StandardSendingDestination) |
| { |
| MessageSource messageSource = _destination.getMessageSource(); |
| |
| if(messageSource instanceof Queue && ((Queue<?>)messageSource).getAvailableAttributes().contains("topic")) |
| { |
| source.setDistributionMode(StdDistMode.COPY); |
| } |
| |
| Map<Symbol,Filter> filters = source.getFilter(); |
| |
| Map<Symbol,Filter> actualFilters = new HashMap<>(); |
| |
| if(filters != null) |
| { |
| for(Map.Entry<Symbol,Filter> entry : filters.entrySet()) |
| { |
| if(entry.getValue() instanceof NoLocalFilter) |
| { |
| actualFilters.put(entry.getKey(), entry.getValue()); |
| noLocal = true; |
| } |
| else if(messageFilter == null && entry.getValue() instanceof org.apache.qpid.server.protocol.v1_0.type.messaging.JMSSelectorFilter) |
| { |
| |
| org.apache.qpid.server.protocol.v1_0.type.messaging.JMSSelectorFilter selectorFilter = (org.apache.qpid.server.protocol.v1_0.type.messaging.JMSSelectorFilter) entry.getValue(); |
| try |
| { |
| messageFilter = new JMSSelectorFilter(selectorFilter.getValue()); |
| |
| actualFilters.put(entry.getKey(), entry.getValue()); |
| } |
| catch (ParseException | SelectorParsingException | TokenMgrError e) |
| { |
| Error error = new Error(); |
| error.setCondition(AmqpError.INVALID_FIELD); |
| error.setDescription("Invalid JMS Selector: " + selectorFilter.getValue()); |
| error.setInfo(Collections.singletonMap(Symbol.valueOf("field"), Symbol.valueOf("filter"))); |
| throw new AmqpErrorException(error); |
| } |
| |
| |
| } |
| else if (entry.getValue() instanceof Filter.InvalidFilter) |
| { |
| Error error = new Error(); |
| error.setCondition(AmqpError.NOT_IMPLEMENTED); |
| error.setDescription("Unsupported filter type: " + ((Filter.InvalidFilter)entry.getValue()).getDescriptor()); |
| error.setInfo(Collections.singletonMap(Symbol.valueOf("field"), Symbol.valueOf("filter"))); |
| throw new AmqpErrorException(error); |
| } |
| } |
| } |
| source.setFilter(actualFilters.isEmpty() ? null : actualFilters); |
| |
| if(source.getDistributionMode() != StdDistMode.COPY) |
| { |
| options.add(ConsumerOption.ACQUIRES); |
| options.add(ConsumerOption.SEES_REQUEUES); |
| } |
| } |
| else |
| { |
| throw new ConnectionScopedRuntimeException("Unknown destination type"); |
| } |
| if(noLocal) |
| { |
| options.add(ConsumerOption.NO_LOCAL); |
| } |
| |
| FilterManager filters = null; |
| if(messageFilter != null) |
| { |
| filters = new FilterManager(); |
| filters.add(messageFilter.getName(), messageFilter); |
| } |
| _consumerOptions = options; |
| _consumerFilters = filters; |
| } |
| |
| private void createConsumerTarget() throws AmqpErrorException |
| { |
| final Source source = getSource(); |
| _consumerTarget = new ConsumerTarget_1_0(this, |
| _destination instanceof ExchangeSendingDestination |
| || source.getDistributionMode() != StdDistMode.COPY); |
| try |
| { |
| final String name = getTarget().getAddress() == null ? getLinkName() : getTarget().getAddress(); |
| _consumer = _destination.getMessageSource() |
| .addConsumer(_consumerTarget, |
| _consumerFilters, |
| Message_1_0.class, |
| name, |
| _consumerOptions, |
| getPriority()); |
| _consumerTarget.updateNotifyWorkDesired(); |
| } |
| catch (MessageSource.ExistingExclusiveConsumer e) |
| { |
| String msg = "Cannot add a consumer to the destination as there is already an exclusive consumer"; |
| throw new AmqpErrorException(new Error(AmqpError.RESOURCE_LOCKED, msg), e); |
| } |
| catch (MessageSource.ExistingConsumerPreventsExclusive e) |
| { |
| String msg = "Cannot add an exclusive consumer to the destination as there is already a consumer"; |
| throw new AmqpErrorException(new Error(AmqpError.RESOURCE_LOCKED, msg), e); |
| } |
| catch (MessageSource.ConsumerAccessRefused e) |
| { |
| String msg = "Cannot add an exclusive consumer to the destination as there is an incompatible exclusivity policy"; |
| throw new AmqpErrorException(new Error(AmqpError.RESOURCE_LOCKED, msg), e); |
| } |
| catch (MessageSource.QueueDeleted e) |
| { |
| String msg = "Cannot add a consumer to the destination as the destination has been deleted"; |
| throw new AmqpErrorException(new Error(AmqpError.RESOURCE_DELETED, msg), e); |
| } |
| catch (AccessControlException e) |
| { |
| throw new AmqpErrorException(new Error(AmqpError.UNAUTHORIZED_ACCESS, e.getMessage())); |
| } |
| } |
| |
| |
| @Override |
| protected Map<Symbol, Object> initProperties(final Attach attach) |
| { |
| |
| Map<Symbol, Object> peerProperties = attach.getProperties(); |
| if(peerProperties != null) |
| { |
| Map<Symbol, Object> actualProperties = new HashMap<>(); |
| if(peerProperties.containsKey(PRIORITY)) |
| { |
| Object value = peerProperties.get(PRIORITY); |
| if(value instanceof Number) |
| { |
| _priority = ((Number)value).intValue(); |
| } |
| else if(value instanceof String) |
| { |
| try |
| { |
| _priority = Integer.parseInt(value.toString()); |
| } |
| catch (NumberFormatException e) |
| { |
| // ignore |
| } |
| } |
| if(_priority != null) |
| { |
| actualProperties.put(PRIORITY, _priority); |
| } |
| } |
| return actualProperties; |
| } |
| else |
| { |
| |
| return Collections.emptyMap(); |
| } |
| } |
| |
| @Override |
| protected void reattachLink(final Attach attach) throws AmqpErrorException |
| { |
| if (getSource() == null) |
| { |
| throw new IllegalStateException("Terminus should be set when resuming a Link."); |
| } |
| if (attach.getSource() == null) |
| { |
| throw new IllegalStateException("Attach.getSource should not be null when resuming a Link. That would be recovering the Link."); |
| } |
| |
| Source newSource = (Source) attach.getSource(); |
| Source oldSource = getSource(); |
| |
| final SendingDestination destination = getSession().getSendingDestination(getLink(), oldSource); |
| prepareConsumerOptionsAndFilters(destination); |
| |
| if (getDestination() instanceof ExchangeSendingDestination && !Boolean.TRUE.equals(newSource.getDynamic())) |
| { |
| final SendingDestination newDestination = |
| getSession().getSendingDestination(getLink(), newSource); |
| if (getSession().updateSourceForSubscription(this, newSource, newDestination)) |
| { |
| setDestination(newDestination); |
| } |
| } |
| |
| attachReceived(attach); |
| } |
| |
| @Override |
| protected void resumeLink(final Attach attach) throws AmqpErrorException |
| { |
| if (getSource() == null) |
| { |
| throw new IllegalStateException("Terminus should be set when resuming a Link."); |
| } |
| if (attach.getSource() == null) |
| { |
| throw new IllegalStateException("Attach.getSource should not be null when resuming a Link. That would be recovering the Link."); |
| } |
| |
| Source newSource = (Source) attach.getSource(); |
| Source oldSource = getSource(); |
| |
| final SendingDestination destination = getSession().getSendingDestination(getLink(), oldSource); |
| prepareConsumerOptionsAndFilters(destination); |
| |
| if (getDestination() instanceof ExchangeSendingDestination && !Boolean.TRUE.equals(newSource.getDynamic())) |
| { |
| final SendingDestination newDestination = |
| getSession().getSendingDestination(getLink(), newSource); |
| if (getSession().updateSourceForSubscription(this, newSource, newDestination)) |
| { |
| setDestination(newDestination); |
| } |
| } |
| |
| attachReceived(attach); |
| } |
| |
| @Override |
| protected void establishLink(final Attach attach) throws AmqpErrorException |
| { |
| if (getSource() != null || getTarget() != null) |
| { |
| throw new IllegalStateException("LinkEndpoint and Termini should be null when establishing a Link."); |
| } |
| |
| attachReceived(attach); |
| } |
| |
| @Override |
| protected void recoverLink(final Attach attach) throws AmqpErrorException |
| { |
| Source source = getSource(); |
| if (source == null && attach.getDesiredCapabilities() != null) |
| { |
| List<Symbol> capabilities = Arrays.asList(attach.getDesiredCapabilities()); |
| if (capabilities.contains(Session_1_0.GLOBAL_CAPABILITY) |
| && capabilities.contains(Session_1_0.SHARED_CAPABILITY) |
| && getLinkName().endsWith("|global")) |
| { |
| final NamedAddressSpace namedAddressSpace = getSession().getConnection().getAddressSpace(); |
| final Pattern linkNamePattern = Pattern.compile("^" + Pattern.quote(getLinkName()) + "$"); |
| namedAddressSpace.visitSendingLinks((LinkRegistryModel.LinkVisitor<Link_1_0<Source, Target>>) link -> { |
| if (link.getSource() != null |
| && ANY_CONTAINER_ID.matcher(link.getRemoteContainerId()).matches() |
| && linkNamePattern.matcher(link.getName()).matches()) |
| { |
| getLink().setSource(new Source(link.getSource())); |
| return true; |
| } |
| return false; |
| }); |
| } |
| } |
| |
| source = getSource(); |
| if (source == null) |
| { |
| throw new AmqpErrorException(new Error(AmqpError.NOT_FOUND, "Link not found")); |
| } |
| |
| attach.setSource(source); |
| receiveAttach(attach); |
| } |
| |
| @Override |
| public Role getRole() |
| { |
| return Role.SENDER; |
| } |
| |
| private Integer getPriority() |
| { |
| return _priority; |
| } |
| |
| void transfer(final Transfer xfr, final boolean decrementCredit) |
| { |
| Session_1_0 s = getSession(); |
| xfr.setMessageFormat(UnsignedInteger.ZERO); |
| if(decrementCredit) |
| { |
| setLinkCredit(getLinkCredit().subtract(UnsignedInteger.ONE)); |
| } |
| |
| getDeliveryCount().incr(); |
| |
| xfr.setHandle(getLocalHandle()); |
| |
| s.sendTransfer(xfr, this); |
| } |
| |
| |
| boolean drained() |
| { |
| if (_draining) |
| { |
| getDeliveryCount().add(getLinkCredit().intValue()); |
| setLinkCredit(UnsignedInteger.ZERO); |
| sendFlow(); |
| _draining = false; |
| return true; |
| } |
| else |
| { |
| return false; |
| } |
| } |
| |
| @Override |
| public void receiveFlow(final Flow flow) |
| { |
| UnsignedInteger receiverDeliveryCount = flow.getDeliveryCount(); |
| UnsignedInteger receiverLinkCredit = flow.getLinkCredit(); |
| setDrain(flow.getDrain()); |
| |
| Map<Symbol, Object> properties = flow.getProperties(); |
| if (properties != null) |
| { |
| final Binary transactionId = (Binary) properties.get(Symbol.valueOf("txn-id")); |
| if (transactionId != null) |
| { |
| try |
| { |
| getSession().getTransaction(transactionId); |
| } |
| catch (UnknownTransactionException e) |
| { |
| close(new Error(TransactionError.UNKNOWN_ID, e.getMessage())); |
| return; |
| } |
| } |
| |
| _transactionId = transactionId; |
| } |
| |
| if(receiverDeliveryCount == null) |
| { |
| setLinkCredit(receiverLinkCredit); |
| } |
| else |
| { |
| // 2.6.7 Flow Control : link_credit_snd := delivery_count_rcv + link_credit_rcv - delivery_count_snd |
| SequenceNumber limit = |
| new SequenceNumber(receiverDeliveryCount.intValue()).add(receiverLinkCredit.intValue()); |
| if (limit.compareTo(getDeliveryCount()) <= 0) |
| { |
| setLinkCredit(UnsignedInteger.valueOf(0)); |
| } |
| else |
| { |
| setLinkCredit(limit.subtract(getDeliveryCount().intValue()).unsignedIntegerValue()); |
| } |
| } |
| |
| // send flow when echo=true or drain=true but link credit is zero |
| boolean sendFlow = Boolean.TRUE.equals(flow.getEcho()) || |
| ( Boolean.TRUE.equals(flow.getDrain()) && getLinkCredit().equals(UnsignedInteger.ZERO)); |
| |
| flowStateChanged(); |
| |
| if (sendFlow) |
| { |
| sendFlow(); |
| } |
| } |
| |
| @Override |
| public void flowStateChanged() |
| { |
| if(Boolean.TRUE.equals(getDrain())) |
| { |
| if(getLinkCredit().compareTo(UnsignedInteger.ZERO) > 0) |
| { |
| _draining = true; |
| getSession().notifyWork(getConsumerTarget()); |
| } |
| } |
| else |
| { |
| _draining = false; |
| } |
| |
| while(!_resumeAcceptedTransfers.isEmpty() && hasCreditToSend()) |
| { |
| Accepted accepted = new Accepted(); |
| Transfer xfr = new Transfer(); |
| Binary dt = _resumeAcceptedTransfers.remove(0); |
| xfr.setDeliveryTag(dt); |
| xfr.setState(accepted); |
| xfr.setResume(Boolean.TRUE); |
| transfer(xfr, true); |
| xfr.dispose(); |
| } |
| if(_resumeAcceptedTransfers.isEmpty()) |
| { |
| getConsumerTarget().flowStateChanged(); |
| } |
| } |
| |
| |
| @Override |
| protected void remoteDetachedPerformDetach(final Detach detach) |
| { |
| TerminusExpiryPolicy expiryPolicy = getSource().getExpiryPolicy(); |
| if (Boolean.TRUE.equals(detach.getClosed()) |
| || TerminusExpiryPolicy.LINK_DETACH.equals(expiryPolicy) |
| || ((expiryPolicy == null || TerminusExpiryPolicy.SESSION_END.equals(expiryPolicy)) && getSession().isClosing()) |
| || (TerminusExpiryPolicy.CONNECTION_CLOSE.equals(expiryPolicy) && getSession().getConnection().isClosing())) |
| { |
| cleanUpUnsettledDeliveries(); |
| |
| close(); |
| } |
| else if (detach.getError() != null) |
| { |
| cleanUpUnsettledDeliveries(); |
| detach(); |
| destroy(); |
| getConsumerTarget().updateNotifyWorkDesired(); |
| } |
| else |
| { |
| detach(); |
| |
| // TODO: QPID-7845 : Resuming links is unsupported at the moment. Destroying link unconditionally. |
| destroy(); |
| getConsumerTarget().updateNotifyWorkDesired(); |
| } |
| } |
| |
| private void cleanUpUnsettledDeliveries() |
| { |
| getSession().removeDeleteTask(_cleanUpUnsettledDeliveryTask); |
| Modified state = new Modified(); |
| state.setDeliveryFailed(true); |
| |
| for (OutgoingDelivery delivery : _unsettled.values()) |
| { |
| UnsettledAction action = delivery.getAction(); |
| if (action != null) |
| { |
| action.process(state, Boolean.TRUE); |
| delivery.setAction(null); |
| } |
| } |
| _unsettled.clear(); |
| } |
| |
| void addUnsettled(final Binary tag, final UnsettledAction unsettledAction, final MessageInstance messageInstance) |
| { |
| _unsettled.put(tag, new OutgoingDelivery(messageInstance, unsettledAction, null)); |
| } |
| |
| @Override |
| protected void handleDeliveryState(final Binary deliveryTag, final DeliveryState state, final Boolean settled) |
| { |
| OutgoingDelivery outgoingDelivery = _unsettled.get(deliveryTag); |
| boolean localSettle = false; |
| if(outgoingDelivery != null && outgoingDelivery.getAction() != null) |
| { |
| UnsettledAction action = outgoingDelivery.getAction(); |
| localSettle = action.process(state, settled); |
| if(localSettle && !Boolean.TRUE.equals(settled)) |
| { |
| updateDisposition(deliveryTag, state, true); |
| } |
| } |
| if(Boolean.TRUE.equals(settled) || localSettle) |
| { |
| _unsettled.remove(deliveryTag); |
| } |
| } |
| |
| public ServerTransaction getTransaction(Binary transactionId) |
| { |
| Session_1_0 session = getSession(); |
| return session == null ? null : session.getTransaction(transactionId); |
| } |
| |
| AsyncAutoCommitTransaction getAsyncAutoCommitTransaction() |
| { |
| return _asyncAutoCommitTransaction; |
| } |
| |
| public boolean hasCreditToSend() |
| { |
| UnsignedInteger linkCredit = getLinkCredit(); |
| return linkCredit != null && (linkCredit.compareTo(UnsignedInteger.valueOf(0)) > 0) |
| && getSession().hasCreditToSend(); |
| } |
| |
| void updateDisposition(final Binary deliveryTag, DeliveryState state, boolean settled) |
| { |
| if (settled && (_unsettled.remove(deliveryTag) != null)) |
| { |
| getSession().updateDisposition(this, deliveryTag, state, settled); |
| } |
| } |
| |
| public Binary getTransactionId() |
| { |
| return _transactionId; |
| } |
| |
| @Override |
| public void attachReceived(final Attach attach) throws AmqpErrorException |
| { |
| super.attachReceived(attach); |
| |
| Target target = (Target) attach.getTarget(); |
| Source source = getSource(); |
| if (source == null) |
| { |
| source = new Source(); |
| Source attachSource = (Source) attach.getSource(); |
| |
| final Modified defaultOutcome = new Modified(); |
| defaultOutcome.setDeliveryFailed(true); |
| source.setDefaultOutcome(defaultOutcome); |
| source.setOutcomes(Accepted.ACCEPTED_SYMBOL, Released.RELEASED_SYMBOL, Rejected.REJECTED_SYMBOL); |
| source.setAddress(attachSource.getAddress()); |
| source.setDynamic(attachSource.getDynamic()); |
| if (Boolean.TRUE.equals(attachSource.getDynamic()) && attachSource.getDynamicNodeProperties() != null) |
| { |
| Map<Symbol, Object> dynamicNodeProperties = new HashMap<>(); |
| if (attachSource.getDynamicNodeProperties().containsKey(Session_1_0.LIFETIME_POLICY)) |
| { |
| dynamicNodeProperties.put(Session_1_0.LIFETIME_POLICY, |
| attachSource.getDynamicNodeProperties().get(Session_1_0.LIFETIME_POLICY)); |
| } |
| source.setDynamicNodeProperties(dynamicNodeProperties); |
| } |
| source.setDurable(TerminusDurability.min(attachSource.getDurable(), |
| getLink().getHighestSupportedTerminusDurability())); |
| source.setExpiryPolicy(attachSource.getExpiryPolicy()); |
| source.setDistributionMode(attachSource.getDistributionMode()); |
| source.setFilter(attachSource.getFilter()); |
| source.setCapabilities(attachSource.getCapabilities()); |
| final SendingDestination destination = getSession().getSendingDestination(getLink(), source); |
| source.setCapabilities(destination.getCapabilities()); |
| getLink().setSource(source); |
| prepareConsumerOptionsAndFilters(destination); |
| } |
| |
| getLink().setTarget(target); |
| |
| |
| final MessageInstanceConsumer oldConsumer = getConsumer(); |
| createConsumerTarget(); |
| _resumeAcceptedTransfers.clear(); |
| _resumeFullTransfers.clear(); |
| final NamedAddressSpace addressSpace = getSession().getConnection().getAddressSpace(); |
| |
| // TODO: QPID-7845 : Resuming links is unsupported at the moment. Thus, cleaning up unsettled deliveries unconditionally. |
| cleanUpUnsettledDeliveries(); |
| getSession().addDeleteTask(_cleanUpUnsettledDeliveryTask); |
| |
| Map<Binary, OutgoingDelivery> unsettledCopy = new HashMap<>(_unsettled); |
| Map<Binary, DeliveryState> remoteUnsettled = |
| attach.getUnsettled() == null ? Collections.emptyMap() : new HashMap<>(attach.getUnsettled()); |
| |
| final boolean isUnsettledComplete = !Boolean.TRUE.equals(attach.getIncompleteUnsettled()); |
| for (Map.Entry<Binary, OutgoingDelivery> entry : unsettledCopy.entrySet()) |
| { |
| Binary deliveryTag = entry.getKey(); |
| final MessageInstance queueEntry = entry.getValue().getMessageInstance(); |
| if (!remoteUnsettled.containsKey(deliveryTag) && isUnsettledComplete) |
| { |
| queueEntry.setRedelivered(); |
| queueEntry.release(oldConsumer); |
| _unsettled.remove(deliveryTag); |
| } |
| else if (remoteUnsettled.get(deliveryTag) instanceof Outcome) |
| { |
| Outcome outcome = (Outcome) remoteUnsettled.get(deliveryTag); |
| |
| if (outcome instanceof Accepted) |
| { |
| if (oldConsumer.acquires()) |
| { |
| AutoCommitTransaction txn = new AutoCommitTransaction(addressSpace.getMessageStore()); |
| if (queueEntry.acquire() || queueEntry.isAcquired()) |
| { |
| txn.dequeue(Collections.singleton(queueEntry), |
| new ServerTransaction.Action() |
| { |
| @Override |
| public void postCommit() |
| { |
| queueEntry.delete(); |
| } |
| |
| @Override |
| public void onRollback() |
| { |
| } |
| }); |
| } |
| } |
| } |
| else if (outcome instanceof Released) |
| { |
| if (oldConsumer.acquires()) |
| { |
| AutoCommitTransaction txn = new AutoCommitTransaction(addressSpace.getMessageStore()); |
| txn.dequeue(Collections.singleton(queueEntry), |
| new ServerTransaction.Action() |
| { |
| @Override |
| public void postCommit() |
| { |
| queueEntry.release(oldConsumer); |
| } |
| |
| @Override |
| public void onRollback() |
| { |
| } |
| }); |
| } |
| } |
| |
| // TODO: QPID-7845: Handle rejected and modified outcome |
| |
| remoteUnsettled.remove(deliveryTag); |
| _resumeAcceptedTransfers.add(deliveryTag); |
| } |
| else |
| { |
| _resumeFullTransfers.add(queueEntry); |
| |
| // TODO:QPID-7845: exists in receivers map, but not yet got an outcome ... should resend with resume = true |
| } |
| } |
| |
| getConsumerTarget().updateNotifyWorkDesired(); |
| } |
| |
| @Override |
| protected Map<Binary, DeliveryState> getLocalUnsettled() |
| { |
| Map<Binary, DeliveryState> unsettled = new HashMap<>(); |
| for (Map.Entry<Binary, OutgoingDelivery> entry : _unsettled.entrySet()) |
| { |
| unsettled.put(entry.getKey(), entry.getValue().getLocalState()); |
| } |
| return unsettled; |
| } |
| |
| private MessageInstanceConsumer<ConsumerTarget_1_0> getConsumer() |
| { |
| return _consumer; |
| } |
| |
| ConsumerTarget_1_0 getConsumerTarget() |
| { |
| return _consumerTarget; |
| } |
| |
| public SendingDestination getDestination() |
| { |
| return _destination; |
| } |
| |
| public void setDestination(final SendingDestination destination) |
| { |
| _destination = destination; |
| } |
| |
| @Override |
| protected void detach(Error error, final boolean close) |
| { |
| if (_consumerTarget != null) |
| { |
| _consumerTarget.close(); |
| } |
| |
| Source source = getSource(); |
| TerminusExpiryPolicy expiryPolicy = source.getExpiryPolicy(); |
| NamedAddressSpace addressSpace = getSession().getConnection().getAddressSpace(); |
| List<Symbol> sourceCapabilities = source.getCapabilities() == null ? Collections.emptyList() : Arrays.asList(source.getCapabilities()); |
| |
| if (close |
| || TerminusExpiryPolicy.LINK_DETACH.equals(expiryPolicy) |
| || ((expiryPolicy == null || TerminusExpiryPolicy.SESSION_END.equals(expiryPolicy)) && getSession().isClosing()) |
| || (TerminusExpiryPolicy.CONNECTION_CLOSE.equals(expiryPolicy) && getSession().getConnection().isClosing())) |
| { |
| cleanUpUnsettledDeliveries(); |
| } |
| |
| if (close) |
| { |
| Error closingError = null; |
| if (getDestination() instanceof ExchangeSendingDestination |
| && addressSpace instanceof QueueManagingVirtualHost && TerminusExpiryPolicy.NEVER.equals(expiryPolicy)) |
| { |
| try |
| { |
| ((QueueManagingVirtualHost) addressSpace).removeSubscriptionQueue( |
| ((ExchangeSendingDestination) getDestination()).getQueue().getName()); |
| |
| TerminusDurability sourceDurability = source.getDurable(); |
| if (sourceDurability != null |
| && !TerminusDurability.NONE.equals(sourceDurability) |
| && sourceCapabilities.contains(Session_1_0.SHARED_CAPABILITY) |
| && sourceCapabilities.contains(ExchangeSendingDestination.TOPIC_CAPABILITY)) |
| { |
| final Pattern containerIdPattern = sourceCapabilities.contains(Session_1_0.GLOBAL_CAPABILITY) |
| ? ANY_CONTAINER_ID |
| : Pattern.compile("^" + Pattern.quote(getSession().getConnection().getRemoteContainerId()) + "$"); |
| final Pattern linkNamePattern = Pattern.compile("^" + Pattern.quote(getLinkName()) + "\\|?\\d*$"); |
| |
| addressSpace.visitSendingLinks((LinkRegistryModel.LinkVisitor<Link_1_0<Source, Target>>) link -> { |
| if (containerIdPattern.matcher(link.getRemoteContainerId()).matches() |
| && linkNamePattern.matcher(link.getName()).matches()) |
| { |
| link.linkClosed(); |
| } |
| return false; |
| }); |
| } |
| } |
| catch (AccessControlException e) |
| { |
| LOGGER.error("Error unregistering subscription", e); |
| closingError = new Error(AmqpError.NOT_ALLOWED, "Error unregistering subscription"); |
| } |
| catch (IllegalStateException e) |
| { |
| String message; |
| if(sourceCapabilities.contains(Session_1_0.SHARED_CAPABILITY) |
| && sourceCapabilities.contains(ExchangeSendingDestination.TOPIC_CAPABILITY)) |
| { |
| String subscriptionName = getLinkName(); |
| int separator = subscriptionName.indexOf("|"); |
| if (separator > 0) |
| { |
| subscriptionName = subscriptionName.substring(0, separator); |
| } |
| message = "There are active consumers on the shared subscription '"+subscriptionName+"'"; |
| } |
| else |
| { |
| message = e.getMessage(); |
| } |
| closingError = new Error(AmqpError.RESOURCE_LOCKED, message); |
| } |
| catch (NotFoundException e) |
| { |
| closingError = new Error(AmqpError.NOT_FOUND, e.getMessage()); |
| } |
| } |
| if (error == null) |
| { |
| error = closingError; |
| } |
| else |
| { |
| LOGGER.warn("Unexpected error on detaching endpoint {}: {}", getLinkName(), error); |
| } |
| } |
| else if (addressSpace instanceof QueueManagingVirtualHost |
| && ((QueueManagingVirtualHost) addressSpace).isDiscardGlobalSharedSubscriptionLinksOnDetach() |
| && sourceCapabilities.contains(Session_1_0.SHARED_CAPABILITY) |
| && sourceCapabilities.contains(Session_1_0.GLOBAL_CAPABILITY) |
| && sourceCapabilities.contains(ExchangeSendingDestination.TOPIC_CAPABILITY) |
| && !getLinkName().endsWith("|global")) |
| { |
| // For JMS 2.0 global shared subscriptions we do not want to keep the links hanging around. |
| // However, we keep one link (ending with "|global") to perform a null-source lookup upon un-subscription. |
| getLink().linkClosed(); |
| } |
| super.detach(error, close); |
| } |
| |
| @Override |
| public void receiveComplete() |
| { |
| AsyncCommand cmd; |
| while((cmd = _unfinishedCommandsQueue.poll()) != null) |
| { |
| cmd.complete(); |
| } |
| } |
| |
| @Override |
| public void recordFuture(final ListenableFuture<Void> future, final ServerTransaction.Action action) |
| { |
| _unfinishedCommandsQueue.add(new AsyncCommand(future, action)); |
| } |
| |
| private static class OutgoingDelivery |
| { |
| private final MessageInstance _messageInstance; |
| private volatile UnsettledAction _action; |
| private volatile DeliveryState _localState; |
| |
| public OutgoingDelivery(final MessageInstance messageInstance, |
| final UnsettledAction action, |
| final DeliveryState localState) |
| { |
| _messageInstance = messageInstance; |
| _action = action; |
| _localState = localState; |
| } |
| |
| public MessageInstance getMessageInstance() |
| { |
| return _messageInstance; |
| } |
| |
| public UnsettledAction getAction() |
| { |
| return _action; |
| } |
| |
| public DeliveryState getLocalState() |
| { |
| return _localState; |
| } |
| |
| public void setLocalState(final DeliveryState localState) |
| { |
| _localState = localState; |
| } |
| |
| public void setAction(final UnsettledAction action) |
| { |
| _action = action; |
| } |
| } |
| } |
