ProtonServerSenderContext extends ProtonInitializable implements ProtonDeliveryHandler { @SuppressWarnings("unchecked") @Override public void initialise() throws Exception { super.initialise(); Source source = (Source) sender.getRemoteSource(); SimpleString queue = null; String selector = null; final Map<Symbol, Object> supportedFilters = new HashMap<>(); sender.setSenderSettleMode(sender.getRemoteSenderSettleMode()); sender.setReceiverSettleMode(ReceiverSettleMode.FIRST); if (source != null) { Map.Entry<Symbol, DescribedType> filter = AmqpSupport.findFilter(source.getFilter(), AmqpSupport.JMS_SELECTOR_FILTER_IDS); if (filter != null) { selector = filter.getValue().getDescribed().toString(); try { SelectorParser.parse(selector); } catch (FilterException e) { throw new ActiveMQAMQPException(AmqpError.INVALID_FIELD, "Invalid filter", ActiveMQExceptionType.INVALID_FILTER_EXPRESSION); } supportedFilters.put(filter.getKey(), filter.getValue()); } } if (source == null) { String clientId = getClientId(); String pubId = sender.getName(); global = hasRemoteDesiredCapability(sender, GLOBAL); queue = createQueueName(connection.isUseCoreSubscriptionNaming(), clientId, pubId, true, global, false); QueueQueryResult result = sessionSPI.queueQuery(queue, RoutingType.MULTICAST, false); multicast = true; routingTypeToUse = RoutingType.MULTICAST; if (result.isExists()) { source = new org.apache.qpid.proton.amqp.messaging.Source(); source.setAddress(queue.toString()); source.setDurable(TerminusDurability.UNSETTLED_STATE); source.setExpiryPolicy(TerminusExpiryPolicy.NEVER); source.setDistributionMode(COPY); source.setCapabilities(TOPIC); SimpleString filterString = result.getFilterString(); if (filterString != null) { selector = filterString.toString(); boolean noLocal = false; String remoteContainerId = sender.getSession().getConnection().getRemoteContainer(); String noLocalFilter = MessageUtil.CONNECTION_ID_PROPERTY_NAME.toString() + "<>'" + remoteContainerId + "'"; if (selector.endsWith(noLocalFilter)) { if (selector.length() > noLocalFilter.length()) { noLocalFilter = " AND " + noLocalFilter; selector = selector.substring(0, selector.length() - noLocalFilter.length()); } else { selector = null; } noLocal = true; } if (noLocal) { supportedFilters.put(AmqpSupport.NO_LOCAL_NAME, AmqpNoLocalFilter.NO_LOCAL); } if (selector != null && !selector.trim().isEmpty()) { supportedFilters.put(AmqpSupport.JMS_SELECTOR_NAME, new AmqpJmsSelectorFilter(selector)); } } sender.setSource(source); } else { throw new ActiveMQAMQPNotFoundException("Unknown subscription link: " + sender.getName()); } } else if (source.getDynamic()) { queue = SimpleString.toSimpleString(java.util.UUID.randomUUID().toString()); tempQueueName = queue; try { sessionSPI.createTemporaryQueue(queue, RoutingType.ANYCAST); } catch (Exception e) { throw ActiveMQAMQPProtocolMessageBundle.BUNDLE.errorCreatingTemporaryQueue(e.getMessage()); } source.setAddress(queue.toString()); } else { SimpleString addressToUse; SimpleString queueNameToUse = null; shared = hasCapabilities(SHARED, source); global = hasCapabilities(GLOBAL, source); if (CompositeAddress.isFullyQualified(source.getAddress())) { addressToUse = SimpleString.toSimpleString(CompositeAddress.extractAddressName(source.getAddress())); queueNameToUse = SimpleString.toSimpleString(CompositeAddress.extractQueueName(source.getAddress())); } else { addressToUse = SimpleString.toSimpleString(source.getAddress()); } boolean clientDefined = hasCapabilities(TOPIC, source) || hasCapabilities(QUEUE, source); if (clientDefined) { multicast = hasCapabilities(TOPIC, source); AddressQueryResult addressQueryResult = null; try { addressQueryResult = sessionSPI.addressQuery(addressToUse, multicast ? RoutingType.MULTICAST : RoutingType.ANYCAST, true); } catch (ActiveMQSecurityException e) { throw ActiveMQAMQPProtocolMessageBundle.BUNDLE.securityErrorCreatingConsumer(e.getMessage()); } catch (ActiveMQAMQPException e) { throw e; } catch (Exception e) { throw new ActiveMQAMQPInternalErrorException(e.getMessage(), e); } if (!addressQueryResult.isExists()) { throw ActiveMQAMQPProtocolMessageBundle.BUNDLE.sourceAddressDoesntExist(); } Set<RoutingType> routingTypes = addressQueryResult.getRoutingTypes(); if (multicast && !routingTypes.contains(RoutingType.MULTICAST)) { throw new ActiveMQAMQPIllegalStateException("Address " + addressToUse + " is not configured for topic support"); } else if (!multicast && !routingTypes.contains(RoutingType.ANYCAST)) { if (queueNameToUse == null) { throw new ActiveMQAMQPIllegalStateException("Address " + addressToUse + " is not configured for queue support"); } } } else { AddressQueryResult addressQueryResult = null; try { addressQueryResult = sessionSPI.addressQuery(addressToUse, defaultRoutingType, true); } catch (ActiveMQSecurityException e) { throw ActiveMQAMQPProtocolMessageBundle.BUNDLE.securityErrorCreatingConsumer(e.getMessage()); } catch (ActiveMQAMQPException e) { throw e; } catch (Exception e) { throw new ActiveMQAMQPInternalErrorException(e.getMessage(), e); } if (!addressQueryResult.isExists()) { throw ActiveMQAMQPProtocolMessageBundle.BUNDLE.sourceAddressDoesntExist(); } Set<RoutingType> routingTypes = addressQueryResult.getRoutingTypes(); if (routingTypes.contains(RoutingType.MULTICAST) && routingTypes.size() == 1) { multicast = true; } else { multicast = false; } } routingTypeToUse = multicast ? RoutingType.MULTICAST : RoutingType.ANYCAST; if (multicast) { Map.Entry<Symbol, DescribedType> filter = AmqpSupport.findFilter(source.getFilter(), AmqpSupport.NO_LOCAL_FILTER_IDS); if (filter != null) { String remoteContainerId = sender.getSession().getConnection().getRemoteContainer(); String noLocalFilter = MessageUtil.CONNECTION_ID_PROPERTY_NAME.toString() + "<>'" + remoteContainerId + "'"; if (selector != null) { selector += " AND " + noLocalFilter; } else { selector = noLocalFilter; } supportedFilters.put(filter.getKey(), filter.getValue()); } queue = getMatchingQueue(queueNameToUse, addressToUse, RoutingType.MULTICAST); SimpleString simpleStringSelector = SimpleString.toSimpleString(selector); if (queue != null) { multicast = false; } else if (TerminusDurability.UNSETTLED_STATE.equals(source.getDurable()) || TerminusDurability.CONFIGURATION.equals(source.getDurable())) { String clientId = getClientId(); String pubId = sender.getName(); queue = createQueueName(connection.isUseCoreSubscriptionNaming(), clientId, pubId, shared, global, false); QueueQueryResult result = sessionSPI.queueQuery(queue, routingTypeToUse, false); if (result.isExists()) { if (!Objects.equals(result.getFilterString(), simpleStringSelector) || (sender.getSource() != null && !sender.getSource().getAddress().equals(result.getAddress().toString()))) { if (result.getConsumerCount() == 0) { sessionSPI.deleteQueue(queue); sessionSPI.createUnsharedDurableQueue(addressToUse, RoutingType.MULTICAST, queue, simpleStringSelector); } else { throw new ActiveMQAMQPIllegalStateException("Unable to recreate subscription, consumers already exist"); } } } else { if (shared) { sessionSPI.createSharedDurableQueue(addressToUse, RoutingType.MULTICAST, queue, simpleStringSelector); } else { sessionSPI.createUnsharedDurableQueue(addressToUse, RoutingType.MULTICAST, queue, simpleStringSelector); } } } else { isVolatile = true; if (shared && sender.getName() != null) { queue = createQueueName(connection.isUseCoreSubscriptionNaming(), getClientId(), sender.getName(), shared, global, isVolatile); QueueQueryResult result = sessionSPI.queueQuery(queue, routingTypeToUse, false); if (!(result.isExists() && Objects.equals(result.getAddress(), addressToUse) && Objects.equals(result.getFilterString(), simpleStringSelector))) { sessionSPI.createSharedVolatileQueue(addressToUse, RoutingType.MULTICAST, queue, simpleStringSelector); } } else { queue = SimpleString.toSimpleString(java.util.UUID.randomUUID().toString()); tempQueueName = queue; try { sessionSPI.createTemporaryQueue(addressToUse, queue, RoutingType.MULTICAST, simpleStringSelector); } catch (Exception e) { throw ActiveMQAMQPProtocolMessageBundle.BUNDLE.errorCreatingTemporaryQueue(e.getMessage()); } } } } else { if (queueNameToUse != null) { routingTypeToUse = null; SimpleString matchingAnycastQueue = getMatchingQueue(queueNameToUse, addressToUse, null); if (matchingAnycastQueue != null) { queue = matchingAnycastQueue; } else { throw ActiveMQAMQPProtocolMessageBundle.BUNDLE.sourceAddressDoesntExist(); } } else { SimpleString matchingAnycastQueue = sessionSPI.getMatchingQueue(addressToUse, RoutingType.ANYCAST); if (matchingAnycastQueue != null) { queue = matchingAnycastQueue; } else { queue = addressToUse; } } } if (queue == null) { throw ActiveMQAMQPProtocolMessageBundle.BUNDLE.sourceAddressNotSet(); } try { if (!sessionSPI.queueQuery(queue, routingTypeToUse, !multicast).isExists()) { throw ActiveMQAMQPProtocolMessageBundle.BUNDLE.sourceAddressDoesntExist(); } } catch (ActiveMQAMQPNotFoundException e) { throw e; } catch (Exception e) { throw new ActiveMQAMQPInternalErrorException(e.getMessage(), e); } } preSettle = sender.getRemoteSenderSettleMode() == SenderSettleMode.SETTLED; source.setFilter(supportedFilters.isEmpty() ? null : supportedFilters); boolean browseOnly = !multicast && source.getDistributionMode() != null && source.getDistributionMode().equals(COPY); try { brokerConsumer = (Consumer) sessionSPI.createSender(this, queue, multicast ? null : selector, browseOnly); onflowControlReady = brokerConsumer::promptDelivery; } catch (ActiveMQAMQPResourceLimitExceededException e1) { throw new ActiveMQAMQPResourceLimitExceededException(e1.getMessage()); } catch (ActiveMQSecurityException e) { throw ActiveMQAMQPProtocolMessageBundle.BUNDLE.securityErrorCreatingConsumer(e.getMessage()); } catch (Exception e) { throw ActiveMQAMQPProtocolMessageBundle.BUNDLE.errorCreatingConsumer(e.getMessage()); } } ProtonServerSenderContext(AMQPConnectionContext connection,
                                    Sender sender,
                                    AMQPSessionContext protonSession,
                                    AMQPSessionCallback server); Object getBrokerConsumer(); @Override void onFlow(int currentCredits, boolean drain); boolean hasCredits(); Sender getSender(); void start(); @SuppressWarnings("unchecked") @Override void initialise(); @Override void close(ErrorCondition condition); @Override void close(boolean remoteLinkClose); @Override void onMessage(Delivery delivery); void settle(Delivery delivery); synchronized void checkState(); int deliverMessage(final MessageReference messageReference, final ServerConsumer consumer); void reportDrained(); AMQPSessionContext getSessionContext();  }
@Test(expected = ActiveMQAMQPNotFoundException.class) public void testAcceptsNullSourceAddressWhenInitialising() throws Exception { ProtonProtocolManager mock = mock(ProtonProtocolManager.class); when(mock.getServer()).thenReturn(mock(ActiveMQServer.class)); Sender mockSender = mock(Sender.class); AMQPConnectionContext mockConnContext = mock(AMQPConnectionContext.class); when(mockConnContext.getProtocolManager()).thenReturn(mock); AMQPSessionCallback mockSessionCallback = mock(AMQPSessionCallback.class); AddressQueryResult queryResult = new AddressQueryResult(null, Collections.emptySet(), 0, false, false, false, false, 0); when(mockSessionCallback.addressQuery(any(), any(), anyBoolean())).thenReturn(queryResult); ProtonServerSenderContext sc = new ProtonServerSenderContext( mockConnContext, mockSender, null, mockSessionCallback); Source source = new Source(); source.setAddress(null); when(mockSender.getRemoteSource()).thenReturn(source); sc.initialise(); }