| /* |
| * |
| * 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 static org.apache.qpid.server.logging.subjects.LogSubjectFormat.CHANNEL_FORMAT; |
| import static org.apache.qpid.server.protocol.v1_0.ExchangeDestination.SHARED_CAPABILITY; |
| |
| import java.security.AccessControlContext; |
| import java.security.AccessControlException; |
| import java.security.AccessController; |
| import java.security.PrivilegedAction; |
| import java.text.MessageFormat; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.EnumSet; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.Iterator; |
| import java.util.LinkedHashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| import java.util.UUID; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.CopyOnWriteArrayList; |
| import java.util.concurrent.atomic.AtomicBoolean; |
| |
| import javax.security.auth.Subject; |
| |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import org.apache.qpid.bytebuffer.QpidByteBuffer; |
| import org.apache.qpid.common.AMQPFilterTypes; |
| import org.apache.qpid.exchange.ExchangeDefaults; |
| import org.apache.qpid.filter.SelectorParsingException; |
| import org.apache.qpid.filter.selector.ParseException; |
| import org.apache.qpid.filter.selector.TokenMgrError; |
| import org.apache.qpid.server.logging.LogMessage; |
| import org.apache.qpid.server.logging.LogSubject; |
| import org.apache.qpid.server.logging.messages.ChannelMessages; |
| import org.apache.qpid.server.message.MessageDestination; |
| 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.AbstractConfiguredObject; |
| import org.apache.qpid.server.model.Binding; |
| import org.apache.qpid.server.model.Consumer; |
| import org.apache.qpid.server.model.Exchange; |
| import org.apache.qpid.server.model.ExclusivityPolicy; |
| 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.model.Session; |
| import org.apache.qpid.server.model.State; |
| import org.apache.qpid.server.protocol.CapacityChecker; |
| import org.apache.qpid.server.protocol.LinkRegistry; |
| import org.apache.qpid.server.protocol.v1_0.codec.QpidByteBufferUtils; |
| import org.apache.qpid.server.protocol.v1_0.framing.OversizeFrameException; |
| 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.ErrorCondition; |
| import org.apache.qpid.server.protocol.v1_0.type.FrameBody; |
| import org.apache.qpid.server.protocol.v1_0.type.LifetimePolicy; |
| 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.DeleteOnClose; |
| import org.apache.qpid.server.protocol.v1_0.type.messaging.DeleteOnNoLinks; |
| import org.apache.qpid.server.protocol.v1_0.type.messaging.DeleteOnNoLinksOrMessages; |
| import org.apache.qpid.server.protocol.v1_0.type.messaging.DeleteOnNoMessages; |
| import org.apache.qpid.server.protocol.v1_0.type.messaging.ExactSubjectFilter; |
| import org.apache.qpid.server.protocol.v1_0.type.messaging.Filter; |
| import org.apache.qpid.server.protocol.v1_0.type.messaging.JMSSelectorFilter; |
| import org.apache.qpid.server.protocol.v1_0.type.messaging.MatchingSubjectFilter; |
| import org.apache.qpid.server.protocol.v1_0.type.messaging.NoLocalFilter; |
| 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.Coordinator; |
| import org.apache.qpid.server.protocol.v1_0.type.transaction.TxnCapability; |
| 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.Begin; |
| import org.apache.qpid.server.protocol.v1_0.type.transport.ConnectionError; |
| import org.apache.qpid.server.protocol.v1_0.type.transport.Detach; |
| import org.apache.qpid.server.protocol.v1_0.type.transport.Disposition; |
| import org.apache.qpid.server.protocol.v1_0.type.transport.End; |
| 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.LinkError; |
| 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.security.SecurityToken; |
| import org.apache.qpid.server.session.AbstractAMQPSession; |
| import org.apache.qpid.server.store.TransactionLogResource; |
| import org.apache.qpid.server.transport.AMQPConnection; |
| 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.QueueManagingVirtualHost; |
| import org.apache.qpid.transport.network.Ticker; |
| |
| public class Session_1_0 extends AbstractAMQPSession<Session_1_0, ConsumerTarget_1_0> |
| implements LogSubject, org.apache.qpid.server.util.Deletable<Session_1_0> |
| { |
| public static final Symbol DELAYED_DELIVERY = Symbol.valueOf("DELAYED_DELIVERY"); |
| private static final Logger _logger = LoggerFactory.getLogger(Session_1_0.class); |
| private static final Symbol LIFETIME_POLICY = Symbol.valueOf("lifetime-policy"); |
| private static final EnumSet<SessionState> END_STATES = |
| EnumSet.of(SessionState.END_RECVD, SessionState.END_PIPE, SessionState.END_SENT, SessionState.ENDED); |
| private AutoCommitTransaction _transaction; |
| |
| private final LinkedHashMap<Integer, ServerTransaction> _openTransactions = |
| new LinkedHashMap<Integer, ServerTransaction>(); |
| |
| private final AMQPConnection_1_0 _connection; |
| private AtomicBoolean _closed = new AtomicBoolean(); |
| |
| private final CopyOnWriteArrayList<Consumer<?, ConsumerTarget_1_0>> _consumers = new CopyOnWriteArrayList<>(); |
| |
| private Session<?> _modelObject = this; |
| |
| private SessionState _sessionState; |
| |
| private final Map<String, SendingLinkEndpoint> _sendingLinkMap = new HashMap<>(); |
| private final Map<String, ReceivingLinkEndpoint> _receivingLinkMap = new HashMap<>(); |
| private final Map<LinkEndpoint, UnsignedInteger> _localLinkEndpoints = new HashMap<>(); |
| private final Map<UnsignedInteger, LinkEndpoint> _remoteLinkEndpoints = new HashMap<>(); |
| private long _lastAttachedTime; |
| |
| private short _receivingChannel; |
| private final short _sendingChannel; |
| |
| private final CapacityCheckAction _capacityCheckAction = new CapacityCheckAction(); |
| |
| |
| // has to be a power of two |
| private static final int DEFAULT_SESSION_BUFFER_SIZE = 1 << 11; |
| private static final int BUFFER_SIZE_MASK = DEFAULT_SESSION_BUFFER_SIZE - 1; |
| |
| |
| |
| private int _nextOutgoingDeliveryId; |
| |
| private UnsignedInteger _outgoingSessionCredit; |
| private UnsignedInteger _initialOutgoingId = UnsignedInteger.valueOf(0); |
| private SequenceNumber _nextIncomingTransferId; |
| private SequenceNumber _nextOutgoingTransferId = new SequenceNumber(_initialOutgoingId.intValue()); |
| |
| private LinkedHashMap<UnsignedInteger,Delivery> _outgoingUnsettled = new LinkedHashMap<>(DEFAULT_SESSION_BUFFER_SIZE); |
| private LinkedHashMap<UnsignedInteger,Delivery> _incomingUnsettled = new LinkedHashMap<>(DEFAULT_SESSION_BUFFER_SIZE); |
| |
| private int _availableIncomingCredit = DEFAULT_SESSION_BUFFER_SIZE; |
| private int _availableOutgoingCredit = DEFAULT_SESSION_BUFFER_SIZE; |
| private UnsignedInteger _lastSentIncomingLimit; |
| |
| private final Error _sessionEndedLinkError = |
| new Error(LinkError.DETACH_FORCED, |
| "Force detach the link because the session is remotely ended."); |
| |
| private final String _primaryDomain; |
| private final Set<Object> _blockingEntities = Collections.newSetFromMap(new ConcurrentHashMap<Object,Boolean>()); |
| private volatile long _startedTransactions; |
| private volatile long _committedTransactions; |
| private volatile long _rolledBackTransactions; |
| private volatile int _unacknowledgedMessages; |
| |
| public Session_1_0(final AMQPConnection_1_0 connection, Begin begin, short sendingChannelId) |
| { |
| super(connection, sendingChannelId); |
| _sendingChannel = sendingChannelId; |
| _sessionState = SessionState.BEGIN_RECVD; |
| _nextIncomingTransferId = new SequenceNumber(begin.getNextOutgoingId().intValue()); |
| _connection = connection; |
| _primaryDomain = getPrimaryDomain(); |
| } |
| |
| public void setReceivingChannel(final short receivingChannel) |
| { |
| _receivingChannel = receivingChannel; |
| switch(_sessionState) |
| { |
| case INACTIVE: |
| _sessionState = SessionState.BEGIN_RECVD; |
| break; |
| case BEGIN_SENT: |
| _sessionState = SessionState.ACTIVE; |
| break; |
| case END_PIPE: |
| _sessionState = SessionState.END_SENT; |
| break; |
| default: |
| // TODO error |
| |
| } |
| } |
| |
| public void sendDetach(final Detach detach) |
| { |
| send(detach); |
| } |
| |
| public void receiveAttach(final Attach attach) |
| { |
| if(_sessionState == SessionState.ACTIVE) |
| { |
| UnsignedInteger handle = attach.getHandle(); |
| if(_remoteLinkEndpoints.containsKey(handle)) |
| { |
| // TODO - Error - handle busy? |
| } |
| else |
| { |
| Map<String, ? extends LinkEndpoint> linkMap = |
| attach.getRole() == Role.RECEIVER ? _sendingLinkMap : _receivingLinkMap; |
| LinkEndpoint endpoint = linkMap.get(attach.getName()); |
| if(endpoint == null) |
| { |
| endpoint = attach.getRole() == Role.RECEIVER |
| ? new SendingLinkEndpoint(this, attach) |
| : new ReceivingLinkEndpoint(this, attach); |
| |
| if(_blockingEntities.contains(this) && attach.getRole() == Role.SENDER) |
| { |
| endpoint.setStopped(true); |
| } |
| |
| // TODO : fix below - distinguish between local and remote owned |
| endpoint.setSource(attach.getSource()); |
| endpoint.setTarget(attach.getTarget()); |
| ((Map<String,LinkEndpoint>)linkMap).put(attach.getName(), endpoint); |
| } |
| else |
| { |
| endpoint.receiveAttach(attach); |
| } |
| |
| if(attach.getRole() == Role.SENDER) |
| { |
| endpoint.setDeliveryCount(attach.getInitialDeliveryCount()); |
| } |
| |
| _remoteLinkEndpoints.put(handle, endpoint); |
| |
| if(!_localLinkEndpoints.containsKey(endpoint)) |
| { |
| UnsignedInteger localHandle = findNextAvailableHandle(); |
| endpoint.setLocalHandle(localHandle); |
| _localLinkEndpoints.put(endpoint, localHandle); |
| |
| remoteLinkCreation(endpoint, attach); |
| } |
| else |
| { |
| // TODO - error already attached |
| } |
| } |
| } |
| } |
| |
| public void updateDisposition(final Role role, |
| final UnsignedInteger first, |
| final UnsignedInteger last, |
| final DeliveryState state, final boolean settled) |
| { |
| |
| |
| Disposition disposition = new Disposition(); |
| disposition.setRole(role); |
| disposition.setFirst(first); |
| disposition.setLast(last); |
| disposition.setSettled(settled); |
| |
| disposition.setState(state); |
| |
| if (settled) |
| { |
| final LinkedHashMap<UnsignedInteger, Delivery> unsettled = |
| role == Role.RECEIVER ? _incomingUnsettled : _outgoingUnsettled; |
| SequenceNumber pos = new SequenceNumber(first.intValue()); |
| SequenceNumber end = new SequenceNumber(last.intValue()); |
| while (pos.compareTo(end) <= 0) |
| { |
| unsettled.remove(new UnsignedInteger(pos.intValue())); |
| pos.incr(); |
| } |
| } |
| |
| send(disposition); |
| //TODO - check send flow |
| } |
| |
| public boolean hasCreditToSend() |
| { |
| boolean b = _outgoingSessionCredit != null && _outgoingSessionCredit.intValue() > 0; |
| boolean b1 = getOutgoingWindowSize() != null && getOutgoingWindowSize().compareTo(UnsignedInteger.ZERO) > 0; |
| return b && b1; |
| } |
| |
| public void end() |
| { |
| end(new End()); |
| } |
| |
| public void sendTransfer(final Transfer xfr, final SendingLinkEndpoint endpoint, final boolean newDelivery) |
| { |
| _nextOutgoingTransferId.incr(); |
| UnsignedInteger deliveryId; |
| final boolean settled = Boolean.TRUE.equals(xfr.getSettled()); |
| if (newDelivery) |
| { |
| deliveryId = UnsignedInteger.valueOf(_nextOutgoingDeliveryId++); |
| endpoint.setLastDeliveryId(deliveryId); |
| if (!settled) |
| { |
| final Delivery delivery = new Delivery(xfr, endpoint); |
| _outgoingUnsettled.put(deliveryId, delivery); |
| _outgoingSessionCredit = _outgoingSessionCredit.subtract(UnsignedInteger.ONE); |
| endpoint.addUnsettled(delivery); |
| } |
| } |
| else |
| { |
| deliveryId = endpoint.getLastDeliveryId(); |
| final Delivery delivery = _outgoingUnsettled.get(deliveryId); |
| if (delivery != null) |
| { |
| if (!settled) |
| { |
| delivery.addTransfer(xfr); |
| _outgoingSessionCredit = _outgoingSessionCredit.subtract(UnsignedInteger.ONE); |
| } |
| else |
| { |
| _outgoingSessionCredit = _outgoingSessionCredit.add(new UnsignedInteger(delivery.getNumberOfTransfers())); |
| endpoint.settle(delivery.getDeliveryTag()); |
| _outgoingUnsettled.remove(deliveryId); |
| } |
| } |
| } |
| xfr.setDeliveryId(deliveryId); |
| |
| try |
| { |
| List<QpidByteBuffer> payload = xfr.getPayload(); |
| final long remaining = QpidByteBufferUtils.remaining(payload); |
| int payloadSent = _connection.sendFrame(_sendingChannel, xfr, payload); |
| |
| if(payload != null && payloadSent < remaining && payloadSent >= 0) |
| { |
| // TODO - should make this iterative and not recursive |
| |
| Transfer secondTransfer = new Transfer(); |
| |
| secondTransfer.setDeliveryTag(xfr.getDeliveryTag()); |
| secondTransfer.setHandle(xfr.getHandle()); |
| secondTransfer.setSettled(xfr.getSettled()); |
| secondTransfer.setState(xfr.getState()); |
| secondTransfer.setMessageFormat(xfr.getMessageFormat()); |
| secondTransfer.setPayload(payload); |
| |
| sendTransfer(secondTransfer, endpoint, false); |
| |
| secondTransfer.dispose(); |
| } |
| |
| if (payload != null) |
| { |
| for (QpidByteBuffer buf : payload) |
| { |
| buf.dispose(); |
| } |
| } |
| } |
| catch (OversizeFrameException e) |
| { |
| throw new ConnectionScopedRuntimeException(e); |
| } |
| } |
| |
| public boolean isActive() |
| { |
| return _sessionState == SessionState.ACTIVE; |
| } |
| |
| public void receiveEnd(final End end) |
| { |
| switch (_sessionState) |
| { |
| case END_SENT: |
| _sessionState = SessionState.ENDED; |
| break; |
| case ACTIVE: |
| detachLinks(); |
| remoteEnd(end); |
| short sendChannel = _sendingChannel; |
| _connection.sendEnd(sendChannel, new End(), true); |
| _sessionState = SessionState.ENDED; |
| break; |
| default: |
| sendChannel = _sendingChannel; |
| End reply = new End(); |
| Error error = new Error(); |
| error.setCondition(AmqpError.ILLEGAL_STATE); |
| error.setDescription("END called on Session which has not been opened"); |
| reply.setError(error); |
| _connection.sendEnd(sendChannel, reply, true); |
| break; |
| } |
| } |
| |
| public UnsignedInteger getNextOutgoingId() |
| { |
| return UnsignedInteger.valueOf(_nextOutgoingTransferId.intValue()); |
| } |
| |
| public void sendFlowConditional() |
| { |
| if(_nextIncomingTransferId != null) |
| { |
| UnsignedInteger clientsCredit = |
| _lastSentIncomingLimit.subtract(UnsignedInteger.valueOf(_nextIncomingTransferId.intValue())); |
| int i = UnsignedInteger.valueOf(_availableIncomingCredit).subtract(clientsCredit).compareTo(clientsCredit); |
| if (i >= 0) |
| { |
| sendFlow(); |
| } |
| } |
| |
| } |
| |
| public UnsignedInteger getOutgoingWindowSize() |
| { |
| return UnsignedInteger.valueOf(_availableOutgoingCredit); |
| } |
| |
| public void receiveFlow(final Flow flow) |
| { |
| UnsignedInteger handle = flow.getHandle(); |
| final LinkEndpoint endpoint = handle == null ? null : _remoteLinkEndpoints.get(handle); |
| |
| final UnsignedInteger nextOutgoingId = |
| flow.getNextIncomingId() == null ? _initialOutgoingId : flow.getNextIncomingId(); |
| int limit = (nextOutgoingId.intValue() + flow.getIncomingWindow().intValue()); |
| _outgoingSessionCredit = UnsignedInteger.valueOf(limit - _nextOutgoingTransferId.intValue()); |
| |
| if (endpoint != null) |
| { |
| endpoint.receiveFlow(flow); |
| } |
| else |
| { |
| final Collection<LinkEndpoint> allLinkEndpoints = _remoteLinkEndpoints.values(); |
| for (LinkEndpoint le : allLinkEndpoints) |
| { |
| le.flowStateChanged(); |
| } |
| } |
| } |
| |
| public void setNextIncomingId(final UnsignedInteger nextIncomingId) |
| { |
| _nextIncomingTransferId = new SequenceNumber(nextIncomingId.intValue()); |
| |
| } |
| |
| public void receiveDisposition(final Disposition disposition) |
| { |
| Role dispositionRole = disposition.getRole(); |
| |
| LinkedHashMap<UnsignedInteger, Delivery> unsettledTransfers; |
| |
| if(dispositionRole == Role.RECEIVER) |
| { |
| unsettledTransfers = _outgoingUnsettled; |
| } |
| else |
| { |
| unsettledTransfers = _incomingUnsettled; |
| |
| } |
| |
| UnsignedInteger deliveryId = disposition.getFirst(); |
| UnsignedInteger last = disposition.getLast(); |
| if(last == null) |
| { |
| last = deliveryId; |
| } |
| |
| |
| while(deliveryId.compareTo(last)<=0) |
| { |
| |
| Delivery delivery = unsettledTransfers.get(deliveryId); |
| if(delivery != null) |
| { |
| delivery.getLinkEndpoint().receiveDeliveryState(delivery, |
| disposition.getState(), |
| disposition.getSettled()); |
| if (Boolean.TRUE.equals(disposition.getSettled())) |
| { |
| unsettledTransfers.remove(deliveryId); |
| } |
| } |
| deliveryId = deliveryId.add(UnsignedInteger.ONE); |
| } |
| if(Boolean.TRUE.equals(disposition.getSettled())) |
| { |
| //TODO - check send flow |
| } |
| |
| } |
| |
| public SessionState getSessionState() |
| { |
| return _sessionState; |
| } |
| |
| public void sendFlow() |
| { |
| sendFlow(new Flow()); |
| } |
| |
| public void setSendingChannel(final short sendingChannel) |
| { |
| switch(_sessionState) |
| { |
| case INACTIVE: |
| _sessionState = SessionState.BEGIN_SENT; |
| break; |
| case BEGIN_RECVD: |
| _sessionState = SessionState.ACTIVE; |
| break; |
| default: |
| // TODO error |
| |
| } |
| |
| AccessController.doPrivileged((new PrivilegedAction<Object>() |
| { |
| @Override |
| public Object run() |
| { |
| _connection.getEventLogger().message(ChannelMessages.CREATE()); |
| |
| return null; |
| } |
| }), _accessControllerContext); |
| } |
| |
| public void sendFlow(final Flow flow) |
| { |
| if(_nextIncomingTransferId != null) |
| { |
| final int nextIncomingId = _nextIncomingTransferId.intValue(); |
| flow.setNextIncomingId(UnsignedInteger.valueOf(nextIncomingId)); |
| _lastSentIncomingLimit = UnsignedInteger.valueOf(nextIncomingId + _availableIncomingCredit); |
| } |
| flow.setIncomingWindow(UnsignedInteger.valueOf(_availableIncomingCredit)); |
| |
| flow.setNextOutgoingId(UnsignedInteger.valueOf(_nextOutgoingTransferId.intValue())); |
| flow.setOutgoingWindow(UnsignedInteger.valueOf(_availableOutgoingCredit)); |
| send(flow); |
| } |
| |
| public void setOutgoingSessionCredit(final UnsignedInteger outgoingSessionCredit) |
| { |
| _outgoingSessionCredit = outgoingSessionCredit; |
| } |
| |
| public void receiveDetach(final Detach detach) |
| { |
| UnsignedInteger handle = detach.getHandle(); |
| detach(handle, detach); |
| } |
| |
| public void sendAttach(final Attach attach) |
| { |
| send(attach); |
| } |
| |
| private void send(final FrameBody frameBody) |
| { |
| _connection.sendFrame(_sendingChannel, frameBody); |
| } |
| |
| public boolean isSyntheticError(final Error error) |
| { |
| return error == _sessionEndedLinkError; |
| } |
| |
| public void end(final End end) |
| { |
| switch (_sessionState) |
| { |
| case BEGIN_SENT: |
| _connection.sendEnd(_sendingChannel, end, false); |
| _sessionState = SessionState.END_PIPE; |
| break; |
| case ACTIVE: |
| detachLinks(); |
| short sendChannel = _sendingChannel; |
| _connection.sendEnd(sendChannel, end, true); |
| _sessionState = SessionState.END_SENT; |
| break; |
| default: |
| sendChannel = _sendingChannel; |
| End reply = new End(); |
| Error error = new Error(); |
| error.setCondition(AmqpError.ILLEGAL_STATE); |
| error.setDescription("END called on Session which has not been opened"); |
| reply.setError(error); |
| _connection.sendEnd(sendChannel, reply, true); |
| break; |
| |
| |
| } |
| } |
| |
| public void receiveTransfer(final Transfer transfer) |
| { |
| _nextIncomingTransferId.incr(); |
| |
| UnsignedInteger handle = transfer.getHandle(); |
| |
| |
| LinkEndpoint linkEndpoint = _remoteLinkEndpoints.get(handle); |
| |
| if (linkEndpoint == null) |
| { |
| Error error = new Error(); |
| error.setCondition(AmqpError.ILLEGAL_STATE); |
| error.setDescription("TRANSFER called on Session for link handle " + handle + " which is not attached"); |
| _connection.close(error); |
| |
| } |
| else if(!(linkEndpoint instanceof ReceivingLinkEndpoint)) |
| { |
| |
| Error error = new Error(); |
| error.setCondition(ConnectionError.FRAMING_ERROR); |
| error.setDescription("TRANSFER called on Session for link handle " + handle + " which is a sending ink not a receiving link"); |
| _connection.close(error); |
| |
| } |
| else |
| { |
| ReceivingLinkEndpoint endpoint = ((ReceivingLinkEndpoint) linkEndpoint); |
| |
| UnsignedInteger deliveryId = transfer.getDeliveryId(); |
| if (deliveryId == null) |
| { |
| deliveryId = endpoint.getLastDeliveryId(); |
| } |
| |
| Delivery delivery = _incomingUnsettled.get(deliveryId); |
| if (delivery == null) |
| { |
| delivery = new Delivery(transfer, endpoint); |
| _incomingUnsettled.put(deliveryId, delivery); |
| |
| if (Boolean.TRUE.equals(transfer.getMore())) |
| { |
| endpoint.setLastDeliveryId(transfer.getDeliveryId()); |
| } |
| } |
| else |
| { |
| if (delivery.getDeliveryId().equals(deliveryId)) |
| { |
| delivery.addTransfer(transfer); |
| |
| if (!Boolean.TRUE.equals(transfer.getMore())) |
| { |
| endpoint.setLastDeliveryId(null); |
| } |
| } |
| else |
| { |
| End reply = new End(); |
| |
| Error error = new Error(); |
| error.setCondition(AmqpError.ILLEGAL_STATE); |
| error.setDescription("TRANSFER called on Session for link handle " |
| + handle |
| + " with incorrect delivery id " |
| + transfer.getDeliveryId()); |
| reply.setError(error); |
| _connection.sendEnd(_sendingChannel, reply, true); |
| |
| return; |
| |
| } |
| } |
| |
| Error error = endpoint.receiveTransfer(transfer, delivery); |
| if(error != null) |
| { |
| endpoint.close(error); |
| } |
| if ((delivery.isComplete() && delivery.isSettled() || Boolean.TRUE.equals(transfer.getAborted()))) |
| { |
| _incomingUnsettled.remove(deliveryId); |
| } |
| } |
| } |
| |
| private Collection<LinkEndpoint> getLocalLinkEndpoints() |
| { |
| return new ArrayList<>(_localLinkEndpoints.keySet()); |
| } |
| |
| boolean isEnded() |
| { |
| return _sessionState == SessionState.ENDED || _connection.isClosed(); |
| } |
| |
| UnsignedInteger getIncomingWindowSize() |
| { |
| return UnsignedInteger.valueOf(_availableIncomingCredit); |
| } |
| |
| AccessControlContext getAccessControllerContext() |
| { |
| return _accessControllerContext; |
| } |
| |
| public void remoteLinkCreation(final LinkEndpoint endpoint, Attach attach) |
| { |
| Link_1_0 link = null; |
| Error error = null; |
| Set<Symbol> capabilities = new HashSet<>(); |
| try |
| { |
| if (endpoint.getRole() == Role.SENDER) |
| { |
| link = createSendingLink(endpoint, attach); |
| if (link != null) |
| { |
| capabilities.add(AMQPConnection_1_0Impl.SHARED_SUBSCRIPTIONS); |
| } |
| } |
| else if (endpoint.getTarget() instanceof Coordinator) |
| { |
| link = createCoordinatorLink(endpoint); |
| } |
| else // standard (non-Coordinator) receiver |
| { |
| link = createReceivingLink(endpoint, capabilities); |
| } |
| } |
| catch (AmqpErrorException e) |
| { |
| |
| if (e.getError() == null || e.getError().getCondition() == AmqpError.INTERNAL_ERROR) |
| { |
| _logger.error("Could not create link", e); |
| } |
| else |
| { |
| _logger.debug("Could not create link", e); |
| } |
| if (endpoint.getRole() == Role.SENDER) |
| { |
| endpoint.setSource(null); |
| } |
| else |
| { |
| endpoint.setTarget(null); |
| } |
| error = e.getError(); |
| } |
| |
| endpoint.setCapabilities(capabilities); |
| endpoint.attach(); |
| |
| if (link == null) |
| { |
| if (error == null) |
| { |
| error = new Error(); |
| error.setCondition(AmqpError.NOT_FOUND); |
| } |
| endpoint.close(error); |
| } |
| else |
| { |
| link.start(); |
| } |
| } |
| |
| private Link_1_0 createReceivingLink(final LinkEndpoint endpoint, |
| final Set<Symbol> capabilities) |
| { |
| Link_1_0 link = null; |
| Destination destination; |
| final LinkRegistry linkRegistry = getAddressSpace().getLinkRegistry(getConnection().getRemoteContainerId()); |
| StandardReceivingLink_1_0 previousLink = |
| (StandardReceivingLink_1_0) linkRegistry.getDurableReceivingLink(endpoint.getName()); |
| |
| if (previousLink == null) |
| { |
| |
| Target target = (Target) endpoint.getTarget(); |
| |
| if (target != null) |
| { |
| if (Boolean.TRUE.equals(target.getDynamic())) |
| { |
| |
| MessageDestination tempQueue = createDynamicDestination(target.getDynamicNodeProperties()); |
| target.setAddress(tempQueue.getName()); |
| } |
| |
| String addr = target.getAddress(); |
| if (addr == null || "".equals(addr.trim())) |
| { |
| MessageDestination messageDestination = getAddressSpace().getDefaultDestination(); |
| destination = new NodeReceivingDestination(messageDestination, target.getDurable(), |
| target.getExpiryPolicy(), "", |
| target.getCapabilities(), |
| _connection.getEventLogger()); |
| target.setCapabilities(destination.getCapabilities()); |
| |
| if (_blockingEntities.contains(messageDestination)) |
| { |
| endpoint.setStopped(true); |
| } |
| } |
| else if (!addr.startsWith("/") && addr.contains("/")) |
| { |
| String[] parts = addr.split("/", 2); |
| Exchange<?> exchange = getExchange(parts[0]); |
| if (exchange != null) |
| { |
| Symbol[] capabilities1 = target.getCapabilities(); |
| ExchangeDestination exchangeDestination = new ExchangeDestination(exchange, |
| null, |
| target.getDurable(), |
| target.getExpiryPolicy(), |
| parts[0], |
| parts[1], |
| capabilities1 != null ? Arrays.asList(capabilities1) : Collections.<Symbol>emptyList()); |
| target.setCapabilities(exchangeDestination.getCapabilities()); |
| destination = exchangeDestination; |
| } |
| else |
| { |
| endpoint.setTarget(null); |
| destination = null; |
| } |
| } |
| else |
| { |
| MessageDestination messageDestination = |
| getAddressSpace().getAttainedMessageDestination(addr); |
| if (messageDestination != null) |
| { |
| destination = |
| new NodeReceivingDestination(messageDestination, |
| target.getDurable(), |
| target.getExpiryPolicy(), |
| addr, |
| target.getCapabilities(), |
| _connection.getEventLogger()); |
| target.setCapabilities(destination.getCapabilities()); |
| } |
| else |
| { |
| Queue<?> queue = getQueue(addr); |
| if (queue != null) |
| { |
| |
| destination = new QueueDestination(queue, addr); |
| } |
| else |
| { |
| endpoint.setTarget(null); |
| destination = null; |
| } |
| } |
| } |
| } |
| else |
| { |
| destination = null; |
| } |
| if (destination != null) |
| { |
| final ReceivingDestination receivingDestination = (ReceivingDestination) destination; |
| MessageDestination messageDestination = receivingDestination.getMessageDestination(); |
| if(!(messageDestination instanceof Queue) || ((Queue<?>)messageDestination).isHoldOnPublishEnabled()) |
| { |
| capabilities.add(DELAYED_DELIVERY); |
| } |
| final ReceivingLinkEndpoint receivingLinkEndpoint = (ReceivingLinkEndpoint) endpoint; |
| final StandardReceivingLink_1_0 receivingLink = |
| new StandardReceivingLink_1_0(new ReceivingLinkAttachment(this, receivingLinkEndpoint), |
| getAddressSpace(), |
| receivingDestination); |
| |
| receivingLinkEndpoint.setLink(receivingLink); |
| link = receivingLink; |
| if (TerminusDurability.UNSETTLED_STATE.equals(target.getDurable()) |
| || TerminusDurability.CONFIGURATION.equals(target.getDurable())) |
| { |
| linkRegistry.registerReceivingLink(endpoint.getName(), receivingLink); |
| } |
| } |
| } |
| else |
| { |
| ReceivingLinkEndpoint receivingLinkEndpoint = (ReceivingLinkEndpoint) endpoint; |
| previousLink.setLinkAttachment(new ReceivingLinkAttachment(this, receivingLinkEndpoint)); |
| receivingLinkEndpoint.setLink(previousLink); |
| link = previousLink; |
| endpoint.setLocalUnsettled(previousLink.getUnsettledOutcomeMap()); |
| } |
| return link; |
| } |
| |
| private TxnCoordinatorReceivingLink_1_0 createCoordinatorLink(final LinkEndpoint endpoint) throws AmqpErrorException |
| { |
| Coordinator coordinator = (Coordinator) endpoint.getTarget(); |
| TxnCapability[] coordinatorCapabilities = coordinator.getCapabilities(); |
| boolean localTxn = false; |
| boolean multiplePerSession = false; |
| if (coordinatorCapabilities != null) |
| { |
| for (TxnCapability capability : coordinatorCapabilities) |
| { |
| if (capability.equals(TxnCapability.LOCAL_TXN)) |
| { |
| localTxn = true; |
| } |
| else if (capability.equals(TxnCapability.MULTI_TXNS_PER_SSN)) |
| { |
| multiplePerSession = true; |
| } |
| else |
| { |
| Error error = new Error(); |
| error.setCondition(AmqpError.NOT_IMPLEMENTED); |
| error.setDescription("Unsupported capability: " + capability); |
| throw new AmqpErrorException(error); |
| } |
| } |
| } |
| |
| final ReceivingLinkEndpoint receivingLinkEndpoint = (ReceivingLinkEndpoint) endpoint; |
| final TxnCoordinatorReceivingLink_1_0 coordinatorLink = |
| new TxnCoordinatorReceivingLink_1_0(getAddressSpace(), |
| this, |
| receivingLinkEndpoint, |
| _openTransactions); |
| receivingLinkEndpoint.setLink(coordinatorLink); |
| return coordinatorLink; |
| } |
| |
| private SendingLink_1_0 createSendingLink(final LinkEndpoint endpoint, Attach attach) throws AmqpErrorException |
| { |
| final SendingLinkEndpoint sendingLinkEndpoint = (SendingLinkEndpoint) endpoint; |
| SendingLink_1_0 link = null; |
| final LinkRegistry linkRegistry = getAddressSpace().getLinkRegistry(getConnection().getRemoteContainerId()); |
| final SendingLink_1_0 previousLink = (SendingLink_1_0) linkRegistry.getDurableSendingLink(endpoint.getName()); |
| |
| if (previousLink == null) |
| { |
| Source source = (Source) sendingLinkEndpoint.getSource(); |
| SendingDestination destination = null; |
| if (source != null) |
| { |
| destination = getSendingDestination(sendingLinkEndpoint.getName(), source); |
| if (destination == null) |
| { |
| sendingLinkEndpoint.setSource(null); |
| } |
| else |
| { |
| source.setCapabilities(destination.getCapabilities()); |
| } |
| } |
| else |
| { |
| final Symbol[] linkCapabilities = attach.getDesiredCapabilities(); |
| boolean isGlobal = hasCapability(linkCapabilities, ExchangeDestination.GLOBAL_CAPABILITY); |
| final String queueName = getMangledSubscriptionName(endpoint.getName(), true, true, isGlobal); |
| final MessageSource messageSource = getAddressSpace().getAttainedMessageSource(queueName); |
| // TODO START The Source should be persisted on the LinkEndpoint |
| if (messageSource instanceof Queue) |
| { |
| Queue<?> queue = (Queue<?>) messageSource; |
| source = new Source(); |
| List<Symbol> capabilities = new ArrayList<>(); |
| if (queue.getExclusive() == ExclusivityPolicy.SHARED_SUBSCRIPTION) |
| { |
| capabilities.add(ExchangeDestination.SHARED_CAPABILITY); |
| } |
| if (isGlobal) |
| { |
| capabilities.add(ExchangeDestination.GLOBAL_CAPABILITY); |
| } |
| capabilities.add(ExchangeDestination.TOPIC_CAPABILITY); |
| source.setCapabilities(capabilities.toArray(new Symbol[capabilities.size()])); |
| |
| final Collection<Exchange> exchanges = queue.getVirtualHost().getChildren(Exchange.class); |
| String bindingKey = null; |
| Exchange<?> foundExchange = null; |
| for (Exchange<?> exchange : exchanges) |
| { |
| for (Binding binding : exchange.getPublishingLinks(queue)) |
| { |
| String exchangeName = exchange.getName(); |
| bindingKey = binding.getName(); |
| final Map<String, Object> bindingArguments = binding.getArguments(); |
| Map<Symbol, Filter> filter = new HashMap<>(); |
| if (bindingArguments.containsKey(AMQPFilterTypes.JMS_SELECTOR)) |
| { |
| filter.put(Symbol.getSymbol("jms-selector"), new JMSSelectorFilter((String) bindingArguments.get(AMQPFilterTypes.JMS_SELECTOR))); |
| } |
| if (bindingArguments.containsKey(AMQPFilterTypes.NO_LOCAL)) |
| { |
| filter.put(Symbol.getSymbol("no-local"), NoLocalFilter.INSTANCE); |
| } |
| foundExchange = exchange; |
| source.setAddress(exchangeName + "/" + bindingKey); |
| source.setFilter(filter); |
| break; |
| } |
| if (foundExchange != null) |
| { |
| break; |
| } |
| } |
| if (foundExchange != null) |
| { |
| source.setDurable(TerminusDurability.CONFIGURATION); |
| TerminusExpiryPolicy terminusExpiryPolicy; |
| switch (queue.getLifetimePolicy()) |
| { |
| case PERMANENT: |
| terminusExpiryPolicy = TerminusExpiryPolicy.NEVER; |
| break; |
| case DELETE_ON_NO_LINKS: |
| case DELETE_ON_NO_OUTBOUND_LINKS: |
| terminusExpiryPolicy = TerminusExpiryPolicy.LINK_DETACH; |
| break; |
| case DELETE_ON_CONNECTION_CLOSE: |
| terminusExpiryPolicy = TerminusExpiryPolicy.CONNECTION_CLOSE; |
| break; |
| case DELETE_ON_SESSION_END: |
| terminusExpiryPolicy = TerminusExpiryPolicy.SESSION_END; |
| break; |
| default: |
| throw new AmqpErrorException(new Error(AmqpError.NOT_IMPLEMENTED, "unexpected liftetime policy " + queue.getLifetimePolicy())); |
| } |
| sendingLinkEndpoint.setSource(source); |
| destination = new ExchangeDestination(foundExchange, |
| queue, |
| TerminusDurability.CONFIGURATION, |
| terminusExpiryPolicy, |
| foundExchange.getName(), |
| bindingKey, |
| capabilities); |
| } |
| |
| } |
| // TODO END |
| } |
| |
| if (destination != null) |
| { |
| final SendingLink_1_0 sendingLink = |
| new SendingLink_1_0(new SendingLinkAttachment(this, sendingLinkEndpoint), |
| getAddressSpace(), |
| destination); |
| sendingLink.createConsumerTarget(); |
| |
| sendingLinkEndpoint.setLink(sendingLink); |
| registerConsumer(sendingLink); |
| |
| if (destination instanceof ExchangeDestination) |
| { |
| ExchangeDestination exchangeDestination = (ExchangeDestination) destination; |
| exchangeDestination.getQueue().setAttributes(Collections.<String, Object>singletonMap(Queue.DESIRED_STATE, State.ACTIVE)); |
| } |
| |
| link = sendingLink; |
| if (TerminusDurability.UNSETTLED_STATE.equals(source.getDurable()) |
| || TerminusDurability.CONFIGURATION.equals(source.getDurable())) |
| { |
| linkRegistry.registerSendingLink(endpoint.getName(), sendingLink); |
| } |
| |
| } |
| } |
| else |
| { |
| Source newSource = (Source) attach.getSource(); |
| Source oldSource = (Source) previousLink.getEndpoint().getSource(); |
| |
| if (previousLink.getDestination() instanceof ExchangeDestination && newSource != null && !Boolean.TRUE.equals(newSource.getDynamic())) |
| { |
| final SendingDestination newDestination = getSendingDestination(previousLink.getEndpoint().getName(), newSource); |
| if (updateSourceForSubscription(previousLink, newSource, newDestination)) |
| { |
| previousLink.setDestination(newDestination); |
| } |
| } |
| |
| sendingLinkEndpoint.setSource(oldSource); |
| previousLink.setLinkAttachment(new SendingLinkAttachment(this, sendingLinkEndpoint)); |
| sendingLinkEndpoint.setLink(previousLink); |
| link = previousLink; |
| sendingLinkEndpoint.setLocalUnsettled(previousLink.getUnsettledOutcomeMap()); |
| registerConsumer(previousLink); |
| |
| } |
| return link; |
| } |
| |
| private boolean updateSourceForSubscription(final SendingLink_1_0 previousLink, |
| final Source newSource, |
| final SendingDestination newDestination) |
| { |
| SendingDestination oldDestination = previousLink.getDestination(); |
| if (oldDestination instanceof ExchangeDestination) |
| { |
| ExchangeDestination oldExchangeDestination = (ExchangeDestination) oldDestination; |
| String newAddress = newSource.getAddress(); |
| if (newDestination instanceof ExchangeDestination) |
| { |
| ExchangeDestination newExchangeDestination = (ExchangeDestination) newDestination; |
| if (oldExchangeDestination.getQueue() != newExchangeDestination.getQueue()) |
| { |
| Source oldSource = (Source) previousLink.getEndpoint().getSource(); |
| oldSource.setAddress(newAddress); |
| oldSource.setFilter(newSource.getFilter()); |
| return true; |
| } |
| } |
| } |
| return false; |
| } |
| |
| private SendingDestination getSendingDestination(final String linkName, final Source source) throws AmqpErrorException |
| { |
| SendingDestination destination = null; |
| |
| if (Boolean.TRUE.equals(source.getDynamic())) |
| { |
| MessageSource tempQueue = createDynamicSource(source.getDynamicNodeProperties()); |
| source.setAddress(tempQueue.getName()); // todo : temporary topic |
| } |
| |
| String address = source.getAddress(); |
| if (address != null) |
| { |
| if (!address.startsWith("/") && address.contains("/")) |
| { |
| destination = createExchangeDestination(address, linkName, source); |
| } |
| else |
| { |
| MessageSource queue = getAddressSpace().getAttainedMessageSource(address); |
| if (queue != null) |
| { |
| destination = new MessageSourceDestination(queue); |
| } |
| else |
| { |
| destination = createExchangeDestination(address, null, linkName, source); |
| } |
| } |
| } |
| |
| return destination; |
| } |
| |
| private ExchangeDestination createExchangeDestination(String address, final String linkName, final Source source) |
| throws AmqpErrorException |
| { |
| String[] parts = address.split("/", 2); |
| String exchangeName = parts[0]; |
| String bindingKey = parts[1]; |
| return createExchangeDestination(exchangeName, bindingKey, linkName, source); |
| } |
| |
| private ExchangeDestination createExchangeDestination(final String exchangeName, |
| final String bindingKey, |
| final String linkName, |
| final Source source) throws AmqpErrorException |
| { |
| ExchangeDestination exchangeDestination = null; |
| Exchange<?> exchange = getExchange(exchangeName); |
| List<Symbol> sourceCapabilities = new ArrayList<>(); |
| if (exchange != null) |
| { |
| Queue queue = null; |
| if (!Boolean.TRUE.equals(source.getDynamic())) |
| { |
| final Map<String, Object> attributes = new HashMap<>(); |
| boolean isDurable = source.getDurable() != TerminusDurability.NONE; |
| boolean isShared = hasCapability(source.getCapabilities(), SHARED_CAPABILITY); |
| boolean isGlobal = hasCapability(source.getCapabilities(), ExchangeDestination.GLOBAL_CAPABILITY); |
| |
| final String name = getMangledSubscriptionName(linkName, isDurable, isShared, isGlobal); |
| |
| if (isGlobal) |
| { |
| sourceCapabilities.add(ExchangeDestination.GLOBAL_CAPABILITY); |
| } |
| |
| ExclusivityPolicy exclusivityPolicy; |
| if (isShared) |
| { |
| exclusivityPolicy = ExclusivityPolicy.SHARED_SUBSCRIPTION; |
| sourceCapabilities.add(SHARED_CAPABILITY); |
| } |
| else |
| { |
| exclusivityPolicy = ExclusivityPolicy.LINK; |
| } |
| |
| org.apache.qpid.server.model.LifetimePolicy lifetimePolicy = getLifetimePolicy(source.getExpiryPolicy()); |
| |
| attributes.put(Queue.ID, UUID.randomUUID()); |
| attributes.put(Queue.NAME, name); |
| attributes.put(Queue.LIFETIME_POLICY, lifetimePolicy); |
| attributes.put(Queue.EXCLUSIVE, exclusivityPolicy); |
| attributes.put(Queue.DURABLE, isDurable); |
| |
| BindingInfo bindingInfo = new BindingInfo(exchange, name, |
| bindingKey, source.getFilter()); |
| Map<String, Map<String, Object>> bindings = bindingInfo.getBindings(); |
| try |
| { |
| if (getAddressSpace() instanceof QueueManagingVirtualHost) |
| { |
| try |
| { |
| queue = ((QueueManagingVirtualHost) getAddressSpace()).getSubscriptionQueue(exchangeName, attributes, bindings); |
| } |
| catch (NotFoundException e) |
| { |
| throw new AmqpErrorException(new Error(AmqpError.NOT_FOUND, e.getMessage())); |
| } |
| } |
| else |
| { |
| throw new AmqpErrorException(new Error(AmqpError.INTERNAL_ERROR, |
| "Address space of unexpected type")); |
| } |
| } |
| catch(IllegalStateException e) |
| { |
| throw new AmqpErrorException(new Error(AmqpError.RESOURCE_LOCKED, |
| "Subscription is already in use")); |
| } |
| source.setFilter(bindingInfo.getActualFilters().isEmpty() ? null : bindingInfo.getActualFilters()); |
| source.setDistributionMode(StdDistMode.COPY); |
| exchangeDestination = new ExchangeDestination(exchange, |
| queue, |
| source.getDurable(), |
| source.getExpiryPolicy(), |
| exchangeName, |
| bindingKey, |
| sourceCapabilities); |
| } |
| else |
| { |
| // TODO |
| throw new AmqpErrorException(new Error(AmqpError.NOT_IMPLEMENTED, "Temporary subscription is not implemented")); |
| } |
| } |
| return exchangeDestination; |
| } |
| |
| private org.apache.qpid.server.model.LifetimePolicy getLifetimePolicy(final TerminusExpiryPolicy expiryPolicy) throws AmqpErrorException |
| { |
| org.apache.qpid.server.model.LifetimePolicy lifetimePolicy; |
| if (expiryPolicy == null || expiryPolicy == TerminusExpiryPolicy.SESSION_END) |
| { |
| lifetimePolicy = org.apache.qpid.server.model.LifetimePolicy.DELETE_ON_SESSION_END; |
| } |
| else if (expiryPolicy == TerminusExpiryPolicy.LINK_DETACH) |
| { |
| lifetimePolicy = org.apache.qpid.server.model.LifetimePolicy.DELETE_ON_NO_OUTBOUND_LINKS; |
| } |
| else if (expiryPolicy == TerminusExpiryPolicy.CONNECTION_CLOSE) |
| { |
| lifetimePolicy = org.apache.qpid.server.model.LifetimePolicy.DELETE_ON_CONNECTION_CLOSE; |
| } |
| else if (expiryPolicy == TerminusExpiryPolicy.NEVER) |
| { |
| lifetimePolicy = org.apache.qpid.server.model.LifetimePolicy.PERMANENT; |
| } |
| else |
| { |
| Error error = new Error(AmqpError.NOT_IMPLEMENTED, |
| String.format("unknown ExpiryPolicy '%s'", expiryPolicy.getValue())); |
| throw new AmqpErrorException(error); |
| } |
| return lifetimePolicy; |
| } |
| |
| private String getMangledSubscriptionName(final String linkName, |
| final boolean isDurable, |
| final boolean isShared, |
| final boolean isGlobal) |
| { |
| String remoteContainerId = getConnection().getRemoteContainerId(); |
| if (isGlobal) |
| { |
| remoteContainerId = "_global_"; |
| } |
| else |
| { |
| remoteContainerId = sanitizeName(remoteContainerId); |
| } |
| |
| String subscriptionName; |
| if (!isDurable && !isShared) |
| { |
| subscriptionName = UUID.randomUUID().toString(); |
| } |
| else |
| { |
| subscriptionName = linkName; |
| if (isShared) |
| { |
| int separator = subscriptionName.indexOf("|"); |
| if (separator > 0) |
| { |
| subscriptionName = subscriptionName.substring(0, separator); |
| } |
| } |
| subscriptionName = sanitizeName(subscriptionName); |
| } |
| return "qpidsub_/" + remoteContainerId + "_/" + subscriptionName + "_/" + (isDurable |
| ? "durable" |
| : "nondurable"); |
| } |
| |
| private String sanitizeName(String name) |
| { |
| return name.replace("_", "__") |
| .replace(".", "_:") |
| .replace("(", "_O") |
| .replace(")", "_C") |
| .replace("<", "_L") |
| .replace(">", "_R"); |
| } |
| |
| private boolean hasCapability(final Symbol[] capabilities, |
| final Symbol expectedCapability) |
| { |
| if (capabilities != null) |
| { |
| for (Symbol capability : capabilities) |
| { |
| if (expectedCapability.equals(capability)) |
| { |
| return true; |
| } |
| } |
| } |
| return false; |
| } |
| |
| |
| private void registerConsumer(final SendingLink_1_0 link) |
| { |
| MessageInstanceConsumer consumer = link.getConsumer(); |
| if(consumer instanceof Consumer<?,?>) |
| { |
| Consumer<?,ConsumerTarget_1_0> modelConsumer = (Consumer<?,ConsumerTarget_1_0>) consumer; |
| _consumers.add(modelConsumer); |
| } |
| } |
| |
| |
| private MessageSource createDynamicSource(Map properties) |
| { |
| final String queueName = _primaryDomain + "TempQueue" + UUID.randomUUID().toString(); |
| MessageSource queue = null; |
| try |
| { |
| Map<String, Object> attributes = convertDynamicNodePropertiesToAttributes(properties, queueName); |
| |
| |
| |
| queue = getAddressSpace().createMessageSource(MessageSource.class, attributes); |
| } |
| catch (AccessControlException e) |
| { |
| Error error = new Error(); |
| error.setCondition(AmqpError.UNAUTHORIZED_ACCESS); |
| error.setDescription(e.getMessage()); |
| |
| _connection.close(error); |
| } |
| catch (AbstractConfiguredObject.DuplicateNameException e) |
| { |
| _logger.error("A temporary queue was created with a name which collided with an existing queue name"); |
| throw new ConnectionScopedRuntimeException(e); |
| } |
| |
| return queue; |
| } |
| |
| |
| private MessageDestination createDynamicDestination(Map properties) |
| { |
| final String queueName = _primaryDomain + "TempQueue" + UUID.randomUUID().toString(); |
| MessageDestination queue = null; |
| try |
| { |
| Map<String, Object> attributes = convertDynamicNodePropertiesToAttributes(properties, queueName); |
| |
| |
| |
| queue = getAddressSpace().createMessageDestination(MessageDestination.class, attributes); |
| } |
| catch (AccessControlException e) |
| { |
| Error error = new Error(); |
| error.setCondition(AmqpError.UNAUTHORIZED_ACCESS); |
| error.setDescription(e.getMessage()); |
| |
| _connection.close(error); |
| } |
| catch (AbstractConfiguredObject.DuplicateNameException e) |
| { |
| _logger.error("A temporary queue was created with a name which collided with an existing queue name"); |
| throw new ConnectionScopedRuntimeException(e); |
| } |
| |
| return queue; |
| } |
| |
| private Map<String, Object> convertDynamicNodePropertiesToAttributes(final Map properties, final String queueName) |
| { |
| // TODO convert AMQP 1-0 node properties to queue attributes |
| LifetimePolicy lifetimePolicy = properties == null |
| ? null |
| : (LifetimePolicy) properties.get(LIFETIME_POLICY); |
| Map<String,Object> attributes = new HashMap<>(); |
| attributes.put(Queue.ID, UUID.randomUUID()); |
| attributes.put(Queue.NAME, queueName); |
| attributes.put(Queue.DURABLE, false); |
| |
| if(lifetimePolicy instanceof DeleteOnNoLinks) |
| { |
| attributes.put(Queue.LIFETIME_POLICY, |
| org.apache.qpid.server.model.LifetimePolicy.DELETE_ON_NO_LINKS); |
| } |
| else if(lifetimePolicy instanceof DeleteOnNoLinksOrMessages) |
| { |
| attributes.put(Queue.LIFETIME_POLICY, |
| org.apache.qpid.server.model.LifetimePolicy.IN_USE); |
| } |
| else if(lifetimePolicy instanceof DeleteOnClose) |
| { |
| attributes.put(Queue.LIFETIME_POLICY, |
| org.apache.qpid.server.model.LifetimePolicy.DELETE_ON_CONNECTION_CLOSE); |
| } |
| else if(lifetimePolicy instanceof DeleteOnNoMessages) |
| { |
| attributes.put(Queue.LIFETIME_POLICY, |
| org.apache.qpid.server.model.LifetimePolicy.IN_USE); |
| } |
| else |
| { |
| attributes.put(Queue.LIFETIME_POLICY, |
| org.apache.qpid.server.model.LifetimePolicy.DELETE_ON_CONNECTION_CLOSE); |
| } |
| return attributes; |
| } |
| |
| ServerTransaction getTransaction(Binary transactionId) |
| { |
| |
| ServerTransaction transaction = _openTransactions.get(binaryToInteger(transactionId)); |
| if(transactionId == null) |
| { |
| if(_transaction == null) |
| { |
| _transaction = new AutoCommitTransaction(_connection.getAddressSpace().getMessageStore()); |
| } |
| transaction = _transaction; |
| } |
| return transaction; |
| } |
| |
| void remoteEnd(End end) |
| { |
| // TODO - if the end has a non empty error we should log it |
| Iterator<Map.Entry<Integer, ServerTransaction>> iter = _openTransactions.entrySet().iterator(); |
| |
| while(iter.hasNext()) |
| { |
| Map.Entry<Integer, ServerTransaction> entry = iter.next(); |
| entry.getValue().rollback(); |
| iter.remove(); |
| } |
| |
| for(LinkEndpoint linkEndpoint : getLocalLinkEndpoints()) |
| { |
| linkEndpoint.remoteDetached(new Detach()); |
| } |
| |
| _connection.sessionEnded(this); |
| performCloseTasks(); |
| if(_modelObject != null) |
| { |
| _modelObject.delete(); |
| } |
| |
| } |
| |
| Integer binaryToInteger(final Binary txnId) |
| { |
| if(txnId == null) |
| { |
| return null; |
| } |
| |
| byte[] data = txnId.getArray(); |
| if(data.length > 4) |
| { |
| throw new IllegalArgumentException(); |
| } |
| |
| int id = 0; |
| for(int i = 0; i < data.length; i++) |
| { |
| id <<= 8; |
| id |= ((int)data[i] & 0xff); |
| } |
| |
| return id; |
| |
| } |
| |
| Binary integerToBinary(final int txnId) |
| { |
| byte[] data = new byte[4]; |
| data[3] = (byte) (txnId & 0xff); |
| data[2] = (byte) ((txnId & 0xff00) >> 8); |
| data[1] = (byte) ((txnId & 0xff0000) >> 16); |
| data[0] = (byte) ((txnId & 0xff000000) >> 24); |
| return new Binary(data); |
| } |
| |
| @Override |
| public void close() |
| { |
| performCloseTasks(); |
| end(); |
| if(_modelObject != null) |
| { |
| _modelObject.delete(); |
| } |
| } |
| |
| private void performCloseTasks() |
| { |
| |
| if(_closed.compareAndSet(false, true)) |
| { |
| List<Action<? super Session_1_0>> taskList = new ArrayList<Action<? super Session_1_0>>(_taskList); |
| _taskList.clear(); |
| for(Action<? super Session_1_0> task : taskList) |
| { |
| task.performAction(this); |
| } |
| getAMQPConnection().getEventLogger().message(_logSubject,ChannelMessages.CLOSE()); |
| } |
| } |
| |
| |
| public void close(ErrorCondition condition, String message) |
| { |
| performCloseTasks(); |
| final End end = new End(); |
| final Error theError = new Error(); |
| theError.setDescription(message); |
| theError.setCondition(condition); |
| end.setError(theError); |
| end(end); |
| } |
| |
| @Override |
| public void transportStateChanged() |
| { |
| for(SendingLinkEndpoint endpoint : _sendingLinkMap.values()) |
| { |
| Link_1_0 link = endpoint.getLink(); |
| ConsumerTarget_1_0 target = ((SendingLink_1_0)link).getConsumerTarget(); |
| target.flowStateChanged(); |
| } |
| |
| if (!_consumersWithPendingWork.isEmpty() && !getAMQPConnection().isTransportBlockedForWriting()) |
| { |
| getAMQPConnection().notifyWork(this); |
| } |
| |
| } |
| |
| @Override |
| public void block(final Queue<?> queue) |
| { |
| getAMQPConnection().doOnIOThreadAsync( |
| new Runnable() |
| { |
| @Override |
| public void run() |
| { |
| doBlock(queue); |
| } |
| }); |
| } |
| |
| private void doBlock(final Queue<?> queue) |
| { |
| if(_blockingEntities.add(queue)) |
| { |
| messageWithSubject(ChannelMessages.FLOW_ENFORCED(queue.getName())); |
| |
| for (ReceivingLinkEndpoint endpoint : _receivingLinkMap.values()) |
| { |
| StandardReceivingLink_1_0 link = (StandardReceivingLink_1_0) endpoint.getLink(); |
| |
| if (isQueueDestinationForLink(queue, link.getDestination())) |
| { |
| endpoint.setStopped(true); |
| } |
| } |
| |
| } |
| } |
| |
| private boolean isQueueDestinationForLink(final Queue<?> queue, final ReceivingDestination recvDest) |
| { |
| return (recvDest instanceof NodeReceivingDestination |
| && queue == ((NodeReceivingDestination) recvDest).getDestination()) |
| || recvDest instanceof QueueDestination && queue == ((QueueDestination) recvDest).getQueue(); |
| } |
| |
| @Override |
| public void unblock(final Queue<?> queue) |
| { |
| getAMQPConnection().doOnIOThreadAsync( |
| new Runnable() |
| { |
| @Override |
| public void run() |
| { |
| doUnblock(queue); |
| } |
| }); |
| } |
| |
| private void doUnblock(final Queue<?> queue) |
| { |
| if(_blockingEntities.remove(queue) && !_blockingEntities.contains(this)) |
| { |
| if(_blockingEntities.isEmpty()) |
| { |
| messageWithSubject(ChannelMessages.FLOW_REMOVED()); |
| } |
| for (ReceivingLinkEndpoint endpoint : _receivingLinkMap.values()) |
| { |
| StandardReceivingLink_1_0 link = (StandardReceivingLink_1_0) endpoint.getLink(); |
| if (isQueueDestinationForLink(queue, link.getDestination())) |
| { |
| endpoint.setStopped(false); |
| } |
| } |
| } |
| } |
| |
| @Override |
| public void block() |
| { |
| getAMQPConnection().doOnIOThreadAsync( |
| new Runnable() |
| { |
| @Override |
| public void run() |
| { |
| doBlock(); |
| } |
| }); |
| } |
| |
| private void doBlock() |
| { |
| if(_blockingEntities.add(this)) |
| { |
| messageWithSubject(ChannelMessages.FLOW_ENFORCED("** All Queues **")); |
| |
| for(LinkEndpoint endpoint : _receivingLinkMap.values()) |
| { |
| endpoint.setStopped(true); |
| } |
| } |
| } |
| |
| |
| |
| @Override |
| public void unblock() |
| { |
| getAMQPConnection().doOnIOThreadAsync( |
| new Runnable() |
| { |
| @Override |
| public void run() |
| { |
| doUnblock(); |
| } |
| }); |
| } |
| |
| private void doUnblock() |
| { |
| if(_blockingEntities.remove(this)) |
| { |
| if(_blockingEntities.isEmpty()) |
| { |
| messageWithSubject(ChannelMessages.FLOW_REMOVED()); |
| } |
| for(ReceivingLinkEndpoint endpoint : _receivingLinkMap.values()) |
| { |
| StandardReceivingLink_1_0 link = (StandardReceivingLink_1_0) endpoint.getLink(); |
| if(!_blockingEntities.contains(link.getDestination())) |
| { |
| endpoint.setStopped(false); |
| } |
| } |
| } |
| } |
| |
| @Override |
| public boolean getBlocking() |
| { |
| return !_blockingEntities.isEmpty(); |
| } |
| |
| private void messageWithSubject(final LogMessage operationalLogMessage) |
| { |
| getEventLogger().message(_logSubject, operationalLogMessage); |
| } |
| |
| @Override |
| public Object getConnectionReference() |
| { |
| return getConnection().getReference(); |
| } |
| |
| @Override |
| public int getUnacknowledgedMessageCount() |
| { |
| return _unacknowledgedMessages; |
| } |
| |
| @Override |
| public long getTxnStart() |
| { |
| return _startedTransactions; |
| } |
| |
| @Override |
| public long getTxnCommits() |
| { |
| return _committedTransactions; |
| } |
| |
| @Override |
| public long getTxnRejects() |
| { |
| return _rolledBackTransactions; |
| } |
| |
| @Override |
| public long getConsumerCount() |
| { |
| return getConsumers().size(); |
| } |
| |
| @Override |
| public String toLogString() |
| { |
| final AMQPConnection<?> amqpConnection = getAMQPConnection(); |
| long connectionId = amqpConnection.getConnectionId(); |
| |
| String remoteAddress = amqpConnection.getRemoteAddressString(); |
| final String authorizedPrincipal = amqpConnection.getAuthorizedPrincipal() == null ? "?" : amqpConnection.getAuthorizedPrincipal().getName(); |
| return "[" + |
| MessageFormat.format(CHANNEL_FORMAT, |
| connectionId, |
| authorizedPrincipal, |
| remoteAddress, |
| getAddressSpace().getName(), |
| _sendingChannel) + "] "; |
| } |
| |
| public AMQPConnection_1_0 getConnection() |
| { |
| return _connection; |
| } |
| |
| @Override |
| public void addDeleteTask(final Action<? super Session_1_0> task) |
| { |
| // TODO is the closed guard important? |
| if(!_closed.get()) |
| { |
| super.addDeleteTask(task); |
| } |
| } |
| |
| public Subject getSubject() |
| { |
| return _subject; |
| } |
| |
| private NamedAddressSpace getAddressSpace() |
| { |
| return _connection.getAddressSpace(); |
| } |
| |
| public SecurityToken getSecurityToken() |
| { |
| return _token; |
| } |
| |
| @Override |
| public Collection<Consumer<?, ConsumerTarget_1_0>> getConsumers() |
| { |
| return Collections.unmodifiableCollection(_consumers); |
| } |
| |
| @Override |
| public long getTransactionStartTimeLong() |
| { |
| return 0L; |
| } |
| |
| @Override |
| public long getTransactionUpdateTimeLong() |
| { |
| return 0L; |
| } |
| |
| @Override |
| protected void updateBlockedStateIfNecessary() |
| { |
| |
| } |
| |
| @Override |
| public boolean isClosing() |
| { |
| return END_STATES.contains(getSessionState()); |
| } |
| |
| @Override |
| public void doTimeoutAction(final String reason) |
| { |
| getAMQPConnection().closeSessionAsync(this, AMQPConnection.CloseReason.TRANSACTION_TIMEOUT, reason); |
| } |
| |
| void incrementStartedTransactions() |
| { |
| _startedTransactions++; |
| } |
| |
| void incrementCommittedTransactions() |
| { |
| _committedTransactions++; |
| } |
| |
| void incrementRolledBackTransactions() |
| { |
| _rolledBackTransactions++; |
| } |
| |
| void incrementUnacknowledged() |
| { |
| _unacknowledgedMessages++; |
| } |
| |
| void decrementUnacknowledged() |
| { |
| _unacknowledgedMessages--; |
| } |
| |
| public CapacityCheckAction getCapacityCheckAction() |
| { |
| return _capacityCheckAction; |
| } |
| |
| @Override |
| public String toString() |
| { |
| return "Session_1_0[" + _connection + ": " + _sendingChannel + ']'; |
| } |
| |
| |
| private void detach(UnsignedInteger handle, Detach detach) |
| { |
| if(_remoteLinkEndpoints.containsKey(handle)) |
| { |
| LinkEndpoint endpoint = _remoteLinkEndpoints.remove(handle); |
| |
| endpoint.remoteDetached(detach); |
| |
| _localLinkEndpoints.remove(endpoint); |
| |
| if (Boolean.TRUE.equals(detach.getClosed())) |
| { |
| Map<String, ? extends LinkEndpoint> linkMap = endpoint.getRole() == Role.SENDER ? _sendingLinkMap : _receivingLinkMap; |
| linkMap.remove(endpoint.getName()); |
| } |
| } |
| else |
| { |
| // TODO |
| } |
| } |
| |
| private void detachLinks() |
| { |
| Collection<UnsignedInteger> handles = new ArrayList<UnsignedInteger>(_remoteLinkEndpoints.keySet()); |
| for(UnsignedInteger handle : handles) |
| { |
| Detach detach = new Detach(); |
| detach.setClosed(false); |
| detach.setHandle(handle); |
| detach.setError(_sessionEndedLinkError); |
| detach(handle, detach); |
| } |
| |
| final LinkRegistry linkRegistry = getAddressSpace().getLinkRegistry(getConnection().getRemoteContainerId()); |
| |
| for(LinkEndpoint<?> linkEndpoint : _sendingLinkMap.values()) |
| { |
| final SendingLink_1_0 link = (SendingLink_1_0) linkRegistry.getDurableSendingLink(linkEndpoint.getName()); |
| |
| if (link != null) |
| { |
| synchronized (link) |
| { |
| if (link.getEndpoint() == linkEndpoint) |
| { |
| try |
| { |
| link.setLinkAttachment(new SendingLinkAttachment(null, (SendingLinkEndpoint) linkEndpoint)); |
| } |
| catch (AmqpErrorException e) |
| { |
| throw new ConnectionScopedRuntimeException(e); |
| } |
| } |
| } |
| } |
| } |
| |
| for(LinkEndpoint<?> linkEndpoint : _receivingLinkMap.values()) |
| { |
| final StandardReceivingLink_1_0 |
| link = (StandardReceivingLink_1_0) linkRegistry.getDurableReceivingLink(linkEndpoint.getName()); |
| |
| if (link != null) |
| { |
| synchronized (link) |
| { |
| if (link.getEndpoint() == linkEndpoint) |
| { |
| link.setLinkAttachment(new ReceivingLinkAttachment(null, (ReceivingLinkEndpoint) linkEndpoint)); |
| } |
| } |
| } |
| } |
| } |
| |
| |
| private UnsignedInteger findNextAvailableHandle() |
| { |
| int i = 0; |
| do |
| { |
| if(!_localLinkEndpoints.containsValue(UnsignedInteger.valueOf(i))) |
| { |
| return UnsignedInteger.valueOf(i); |
| } |
| } |
| while(++i != 0); |
| |
| // TODO |
| throw new RuntimeException(); |
| } |
| |
| |
| private Exchange<?> getExchange(String name) |
| { |
| MessageDestination destination = getAddressSpace().getAttainedMessageDestination(name); |
| return destination instanceof Exchange ? (Exchange<?>) destination : null; |
| } |
| |
| private Queue<?> getQueue(String name) |
| { |
| MessageSource source = getAddressSpace().getAttainedMessageSource(name); |
| return source instanceof Queue ? (Queue<?>) source : null; |
| } |
| |
| private String getPrimaryDomain() |
| { |
| String primaryDomain = ""; |
| final List<String> globalAddressDomains = getAddressSpace().getGlobalAddressDomains(); |
| if (globalAddressDomains != null && !globalAddressDomains.isEmpty()) |
| { |
| primaryDomain = globalAddressDomains.get(0); |
| if(primaryDomain != null) |
| { |
| primaryDomain = primaryDomain.trim(); |
| if(!primaryDomain.endsWith("/")) |
| { |
| primaryDomain += "/"; |
| } |
| } |
| } |
| return primaryDomain; |
| } |
| |
| private final class CapacityCheckAction implements Action<MessageInstance> |
| { |
| @Override |
| public void performAction(final MessageInstance entry) |
| { |
| TransactionLogResource queue = entry.getOwningResource(); |
| if(queue instanceof CapacityChecker) |
| { |
| ((CapacityChecker)queue).checkCapacity(Session_1_0.this); |
| } |
| } |
| } |
| |
| private final class BindingInfo |
| { |
| private final Map<Symbol, Filter> _actualFilters = new HashMap<>(); |
| private final Map<String, Map<String, Object>> _bindings = new HashMap<>(); |
| |
| private BindingInfo(Exchange<?> exchange, |
| final String queueName, |
| String bindingKey, |
| Map<Symbol, Filter> filters) throws AmqpErrorException |
| { |
| String binding = null; |
| final Map<String, Object> arguments = new HashMap<>(); |
| if (filters != null && !filters.isEmpty()) |
| { |
| boolean hasBindingFilter = false; |
| boolean hasMessageFilter = false; |
| for(Map.Entry<Symbol,Filter> entry : filters.entrySet()) |
| { |
| if(!hasBindingFilter |
| && entry.getValue() instanceof ExactSubjectFilter |
| && exchange.getType().equals(ExchangeDefaults.DIRECT_EXCHANGE_CLASS)) |
| { |
| ExactSubjectFilter filter = (ExactSubjectFilter) entry.getValue(); |
| binding = filter.getValue(); |
| _actualFilters.put(entry.getKey(), filter); |
| hasBindingFilter = true; |
| } |
| else if(!hasBindingFilter |
| && entry.getValue() instanceof MatchingSubjectFilter |
| && exchange.getType().equals(ExchangeDefaults.TOPIC_EXCHANGE_CLASS)) |
| { |
| MatchingSubjectFilter filter = (MatchingSubjectFilter) entry.getValue(); |
| binding = filter.getValue(); |
| _actualFilters.put(entry.getKey(), filter); |
| hasBindingFilter = true; |
| } |
| else if(entry.getValue() instanceof NoLocalFilter) |
| { |
| _actualFilters.put(entry.getKey(), entry.getValue()); |
| arguments.put(AMQPFilterTypes.NO_LOCAL.toString(), true); |
| } |
| else if (!hasMessageFilter |
| && 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(); |
| |
| // TODO: QPID-7642 - due to inconsistent handling of invalid filters |
| // by different exchange implementations |
| // we need to validate filter before creation of binding |
| try |
| { |
| new org.apache.qpid.server.filter.JMSSelectorFilter(selectorFilter.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); |
| } |
| |
| arguments.put(AMQPFilterTypes.JMS_SELECTOR.toString(), selectorFilter.getValue()); |
| _actualFilters.put(entry.getKey(), selectorFilter); |
| hasMessageFilter = true; |
| } |
| } |
| } |
| |
| if(binding != null) |
| { |
| _bindings.put(binding, arguments); |
| } |
| if(bindingKey != null) |
| { |
| _bindings.put(bindingKey, arguments); |
| } |
| if(binding == null |
| && bindingKey == null |
| && exchange.getType().equals(ExchangeDefaults.FANOUT_EXCHANGE_CLASS)) |
| { |
| _bindings.put(queueName, arguments); |
| } |
| else if(binding == null |
| && bindingKey == null |
| && exchange.getType().equals(ExchangeDefaults.TOPIC_EXCHANGE_CLASS)) |
| { |
| _bindings.put("#", arguments); |
| } |
| } |
| |
| private Map<Symbol, Filter> getActualFilters() |
| { |
| return _actualFilters; |
| } |
| |
| private Map<String, Map<String, Object>> getBindings() |
| { |
| return _bindings; |
| } |
| |
| |
| @Override |
| public boolean equals(final Object o) |
| { |
| if (this == o) |
| { |
| return true; |
| } |
| if (o == null || getClass() != o.getClass()) |
| { |
| return false; |
| } |
| |
| final BindingInfo that = (BindingInfo) o; |
| |
| return _actualFilters.equals(that._actualFilters) && _bindings.equals(that._bindings); |
| } |
| |
| @Override |
| public int hashCode() |
| { |
| int result = _actualFilters.hashCode(); |
| result = 31 * result + _bindings.hashCode(); |
| return result; |
| } |
| } |
| } |
