Author: rgodfrey Date: Wed Dec 14 15:18:33 2016 New Revision: 1774263 URL: http://svn.apache.org/viewvc?rev=1774263&view=rev Log: QPID-7589 : Allow the creation of temporary reply addresses in the $management synthetic virtual host
Added: qpid/java/trunk/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ProxyMessageSource.java (with props) Modified: qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java qpid/java/trunk/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementAddressSpace.java qpid/java/trunk/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNode.java qpid/java/trunk/systests/src/test/java/org/apache/qpid/systest/management/amqp/AmqpManagementTest.java qpid/java/trunk/test-profiles/Java10BrokenTestsExcludes Modified: qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java?rev=1774263&r1=1774262&r2=1774263&view=diff ============================================================================== --- qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java (original) +++ qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java Wed Dec 14 15:18:33 2016 @@ -2695,6 +2695,10 @@ public abstract class AbstractVirtualHos { return (T) createChild((Class<? extends Queue>)clazz, attributes); } + else if(clazz.isAssignableFrom(Queue.class)) + { + return (T) createChild(Queue.class, attributes); + } else { throw new IllegalArgumentException("Cannot create message source children of class " + clazz.getSimpleName()); @@ -2713,6 +2717,10 @@ public abstract class AbstractVirtualHos { return (T) createChild((Class<? extends Queue>)clazz, attributes); } + else if(clazz.isAssignableFrom(Queue.class)) + { + return (T) createChild(Queue.class, attributes); + } else { throw new IllegalArgumentException("Cannot create message destination children of class " + clazz.getSimpleName()); Modified: qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java?rev=1774263&r1=1774262&r2=1774263&view=diff ============================================================================== --- qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java (original) +++ qpid/java/trunk/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java Wed Dec 14 15:18:33 2016 @@ -799,7 +799,7 @@ public class Session_1_0 implements AMQS { if (Boolean.TRUE.equals(source.getDynamic())) { - Queue<?> tempQueue = createTemporaryQueue(source.getDynamicNodeProperties()); + MessageSource tempQueue = createDynamicSource(source.getDynamicNodeProperties()); source.setAddress(tempQueue.getName()); } String addr = source.getAddress(); @@ -976,7 +976,7 @@ public class Session_1_0 implements AMQS if (Boolean.TRUE.equals(target.getDynamic())) { - Queue<?> tempQueue = createTemporaryQueue(target.getDynamicNodeProperties()); + MessageDestination tempQueue = createDynamicDestination(target.getDynamicNodeProperties()); target.setAddress(tempQueue.getName()); } @@ -1114,50 +1114,47 @@ public class Session_1_0 implements AMQS } - private Queue<?> createTemporaryQueue(Map properties) + private MessageSource createDynamicSource(Map properties) { final String queueName = _primaryDomain + "TempQueue" + UUID.randomUUID().toString(); - Queue<?> queue = null; + MessageSource queue = null; try { - LifetimePolicy lifetimePolicy = properties == null - ? null - : (LifetimePolicy) properties.get(LIFETIME_POLICY); - Map<String,Object> attributes = new HashMap<String,Object>(); - attributes.put(org.apache.qpid.server.model.Queue.ID, UUID.randomUUID()); - attributes.put(org.apache.qpid.server.model.Queue.NAME, queueName); - attributes.put(org.apache.qpid.server.model.Queue.DURABLE, false); + Map<String, Object> attributes = convertDynamicNodePropertiesToAttributes(properties, queueName); - if(lifetimePolicy instanceof DeleteOnNoLinks) - { - attributes.put(org.apache.qpid.server.model.Queue.LIFETIME_POLICY, - org.apache.qpid.server.model.LifetimePolicy.DELETE_ON_NO_LINKS); - } - else if(lifetimePolicy instanceof DeleteOnNoLinksOrMessages) - { - attributes.put(org.apache.qpid.server.model.Queue.LIFETIME_POLICY, - org.apache.qpid.server.model.LifetimePolicy.IN_USE); - } - else if(lifetimePolicy instanceof DeleteOnClose) - { - attributes.put(org.apache.qpid.server.model.Queue.LIFETIME_POLICY, - org.apache.qpid.server.model.LifetimePolicy.DELETE_ON_CONNECTION_CLOSE); - } - else if(lifetimePolicy instanceof DeleteOnNoMessages) - { - attributes.put(org.apache.qpid.server.model.Queue.LIFETIME_POLICY, - org.apache.qpid.server.model.LifetimePolicy.IN_USE); - } - else - { - attributes.put(org.apache.qpid.server.model.Queue.LIFETIME_POLICY, - org.apache.qpid.server.model.LifetimePolicy.DELETE_ON_CONNECTION_CLOSE); - } - // TODO convert AMQP 1-0 node properties to queue attributes + queue = getAddressSpace().createMessageSource(MessageSource.class, attributes); + } + catch (AccessControlException e) + { + Error error = new Error(); + error.setCondition(AmqpError.UNAUTHORIZED_ACCESS); + error.setDescription(e.getMessage()); - queue = getAddressSpace().createMessageSource(Queue.class, attributes); + _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) { @@ -1176,6 +1173,45 @@ public class Session_1_0 implements AMQS 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<String,Object>(); + 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) { Modified: qpid/java/trunk/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementAddressSpace.java URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementAddressSpace.java?rev=1774263&r1=1774262&r2=1774263&view=diff ============================================================================== --- qpid/java/trunk/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementAddressSpace.java (original) +++ qpid/java/trunk/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementAddressSpace.java Wed Dec 14 15:18:33 2016 @@ -22,15 +22,25 @@ package org.apache.qpid.server.managemen import java.nio.charset.StandardCharsets; import java.security.AccessControlException; +import java.security.AccessController; import java.security.Principal; import java.util.Collection; import java.util.Collections; 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.ConcurrentMap; import java.util.concurrent.CopyOnWriteArrayList; +import javax.security.auth.Subject; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import org.apache.qpid.exchange.ExchangeDefaults; +import org.apache.qpid.server.connection.SessionPrincipal; import org.apache.qpid.server.message.InstanceProperties; import org.apache.qpid.server.message.MessageDestination; import org.apache.qpid.server.message.MessageInstance; @@ -41,6 +51,7 @@ import org.apache.qpid.server.model.Conn import org.apache.qpid.server.model.NamedAddressSpace; import org.apache.qpid.server.model.port.AmqpPort; import org.apache.qpid.server.plugin.SystemAddressSpaceCreator; +import org.apache.qpid.server.protocol.AMQSessionModel; import org.apache.qpid.server.protocol.LinkModel; import org.apache.qpid.server.protocol.LinkRegistry; import org.apache.qpid.server.security.SecurityToken; @@ -62,6 +73,8 @@ public class ManagementAddressSpace impl public static final String MANAGEMENT_ADDRESS_SPACE_NAME = "$management"; private static final String MANAGEMENT_NODE_NAME = "$management"; + private static final Logger LOGGER = LoggerFactory.getLogger(ManagementAddressSpace.class); + private final String _name; private final SystemAddressSpaceCreator.AddressSpaceRegistry _addressSpaceRegistry; private final ManagementNode _managementNode; @@ -73,6 +86,7 @@ public class ManagementAddressSpace impl private final Broker<?> _broker; private final Principal _principal; private final UUID _id; + private final ConcurrentMap<Object, ConcurrentMap<String, ProxyMessageSource>> _connectionSpecificDestinations = new ConcurrentHashMap<>(); public ManagementAddressSpace(final SystemAddressSpaceCreator.AddressSpaceRegistry addressSpaceRegistry) { @@ -110,7 +124,10 @@ public class ManagementAddressSpace impl { return _propertiesNode; } - return null; + else + { + return getProxyNode(name); + } } @Override @@ -120,9 +137,40 @@ public class ManagementAddressSpace impl { return _managementNode; } + else + { + MessageDestination connectionSpecificDestinations = getProxyNode(name); + if (connectionSpecificDestinations != null) return connectionSpecificDestinations; + } + + return null; + } + + ProxyMessageSource getProxyNode(final String name) + { + LOGGER.debug("RG: looking for proxy source {}", name); + Subject currentSubject = Subject.getSubject(AccessController.getContext()); + Set<SessionPrincipal> sessionPrincipals = currentSubject.getPrincipals(SessionPrincipal.class); + if (!sessionPrincipals.isEmpty()) + { + Object connectionReference = sessionPrincipals.iterator().next().getSession().getConnectionReference(); + Map<String, ProxyMessageSource> + connectionSpecificDestinations = _connectionSpecificDestinations.get(connectionReference); + if(connectionSpecificDestinations != null) + { + LOGGER.debug("RG: ", connectionSpecificDestinations); + + return connectionSpecificDestinations.get(name); + } + } return null; } + ManagementNode getManagementNode() + { + return _managementNode; + } + @Override public void registerConnection(final AMQPConnection<?> connection) { @@ -187,14 +235,78 @@ public class ManagementAddressSpace impl @Override public <T extends MessageSource> T createMessageSource(final Class<T> clazz, final Map<String, Object> attributes) { - return null; + if(clazz == MessageSource.class) + { + return (T) createProxyNode(attributes); + } + else + { + return null; + } } + private ProxyMessageSource createProxyNode(final Map<String, Object> attributes) + { + LOGGER.debug("RG: in create proxy node"); + Subject currentSubject = Subject.getSubject(AccessController.getContext()); + Set<SessionPrincipal> sessionPrincipals = currentSubject.getPrincipals(SessionPrincipal.class); + if (!sessionPrincipals.isEmpty()) + { + LOGGER.debug("RG: session principal present"); + final ProxyMessageSource proxyMessageSource = new ProxyMessageSource(this, attributes); + final AMQSessionModel session = sessionPrincipals.iterator().next().getSession(); + final Object connectionReference = session.getConnectionReference(); + ConcurrentMap<String, ProxyMessageSource> connectionSpecificDestinations = + _connectionSpecificDestinations.get(connectionReference); + if (connectionSpecificDestinations == null) + { + connectionSpecificDestinations = new ConcurrentHashMap<>(); + if(_connectionSpecificDestinations.putIfAbsent(connectionReference, connectionSpecificDestinations) == null) + { + session.getAMQPConnection().addDeleteTask(new Action() + { + @Override + public void performAction(final Object object) + { + _connectionSpecificDestinations.remove(connectionReference); + } + }); + } + } + connectionSpecificDestinations.put(proxyMessageSource.getName(), proxyMessageSource); + return proxyMessageSource; + } + else + { + return null; + } + } + + void removeProxyMessageSource(final Object connectionReference, final String name) + { + ConcurrentMap<String, ProxyMessageSource> connectionSpecificDestinations = + _connectionSpecificDestinations.get(connectionReference); + if(connectionSpecificDestinations != null) + { + connectionSpecificDestinations.remove(name); + } + } + + @Override public <T extends MessageDestination> T createMessageDestination(final Class<T> clazz, final Map<String, Object> attributes) { - return null; + + LOGGER.debug("RG : requesting destination creation"); + if(clazz == MessageDestination.class) + { + return (T) createProxyNode(attributes); + } + else + { + return null; + } } @Override Modified: qpid/java/trunk/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNode.java URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNode.java?rev=1774263&r1=1774262&r2=1774263&view=diff ============================================================================== --- qpid/java/trunk/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNode.java (original) +++ qpid/java/trunk/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNode.java Wed Dec 14 15:18:33 2016 @@ -48,6 +48,9 @@ import java.util.regex.Pattern; import javax.security.auth.Subject; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import org.apache.qpid.server.configuration.IllegalConfigurationException; import org.apache.qpid.server.connection.SessionPrincipal; import org.apache.qpid.server.consumer.ConsumerOption; @@ -89,6 +92,7 @@ import org.apache.qpid.server.util.State class ManagementNode implements MessageSource, MessageDestination { + private static final Logger LOGGER = LoggerFactory.getLogger(ManagementNode.class); public static final String IDENTITY_ATTRIBUTE = "identity"; public static final String INDEX_ATTRIBUTE = "index"; @@ -978,14 +982,13 @@ class ManagementNode implements MessageS AMQSessionModel publishingSession = sessionPrincipals.iterator().next().getSession(); for (ManagementNodeConsumer candidate : _consumers) { - if (candidate.getTarget().getTargetAddress().equals(replyTo) && candidate.getSessionModel() == publishingSession) + if (candidate.getTarget().getTargetAddress().equals(replyTo) && candidate.getSessionModel().getConnectionReference() == publishingSession.getConnectionReference()) { consumer = candidate; break; } } } - return consumer == null ? _addressSpace.getDefaultDestination() : consumer; } Added: qpid/java/trunk/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ProxyMessageSource.java URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ProxyMessageSource.java?rev=1774263&view=auto ============================================================================== --- qpid/java/trunk/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ProxyMessageSource.java (added) +++ qpid/java/trunk/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ProxyMessageSource.java Wed Dec 14 15:18:33 2016 @@ -0,0 +1,384 @@ +/* + * + * 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.management.amqp; + +import java.security.AccessControlException; +import java.security.AccessController; +import java.util.Collection; +import java.util.Collections; +import java.util.EnumSet; +import java.util.Map; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.atomic.AtomicBoolean; + +import javax.security.auth.Subject; + +import com.google.common.util.concurrent.ListenableFuture; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.qpid.server.connection.SessionPrincipal; +import org.apache.qpid.server.consumer.ConsumerOption; +import org.apache.qpid.server.consumer.ConsumerTarget; +import org.apache.qpid.server.filter.FilterManager; +import org.apache.qpid.server.message.InstanceProperties; +import org.apache.qpid.server.message.MessageContainer; +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.message.ServerMessage; +import org.apache.qpid.server.model.ConfiguredObject; +import org.apache.qpid.server.model.NamedAddressSpace; +import org.apache.qpid.server.protocol.AMQSessionModel; +import org.apache.qpid.server.security.SecurityToken; +import org.apache.qpid.server.store.MessageDurability; +import org.apache.qpid.server.store.StorableMessageMetaData; +import org.apache.qpid.server.txn.ServerTransaction; +import org.apache.qpid.server.util.Action; + +public class ProxyMessageSource implements MessageSource, MessageDestination +{ + private static final Logger LOGGER = LoggerFactory.getLogger(ProxyMessageSource.class); + private final String _name; + private final UUID _id = UUID.randomUUID(); + private final ManagementAddressSpace _managementAddressSpace; + private volatile MessageInstanceConsumer<?> _consumer; + private final AtomicBoolean _consumerSet = new AtomicBoolean(false); + private Object _connectionReference; + + public ProxyMessageSource(final ManagementAddressSpace managementAddressSpace, final Map<String, Object> attributes) + { + _name = String.valueOf(attributes.get(ConfiguredObject.NAME)); + _managementAddressSpace = managementAddressSpace; + } + + @Override + public String getName() + { + return _name; + } + + @Override + public NamedAddressSpace getAddressSpace() + { + return _managementAddressSpace; + } + + @Override + public void authorisePublish(final SecurityToken token, final Map<String, Object> arguments) + throws AccessControlException + { + throw new AccessControlException("Sending messages to temporary addresses in a management addres spaceis not supported"); + } + + @Override + public <M extends ServerMessage<? extends StorableMessageMetaData>> int send(final M message, + final String routingAddress, + final InstanceProperties instanceProperties, + final ServerTransaction txn, + final Action<? super MessageInstance> postEnqueueAction) + { + return 0; + } + + @Override + public UUID getId() + { + return _id; + } + + @Override + public MessageDurability getMessageDurability() + { + return MessageDurability.NEVER; + } + + @Override + public <T extends ConsumerTarget<T>> MessageInstanceConsumer<T> addConsumer(final T target, + final FilterManager filters, + final Class<? extends ServerMessage> messageClass, + final String consumerName, + final EnumSet<ConsumerOption> options, + final Integer priority) + throws ExistingExclusiveConsumer, ExistingConsumerPreventsExclusive, + ConsumerAccessRefused, QueueDeleted + { + if(_consumerSet.compareAndSet(false,true)) + { + Subject currentSubject = Subject.getSubject(AccessController.getContext()); + Set<SessionPrincipal> sessionPrincipals = currentSubject.getPrincipals(SessionPrincipal.class); + if (!sessionPrincipals.isEmpty()) + { + _connectionReference = sessionPrincipals.iterator().next().getSession().getConnectionReference(); + + WrappingTarget<T> wrapper = new WrappingTarget<>(target, _name); + _managementAddressSpace.getManagementNode().addConsumer(wrapper, filters, messageClass, _name, options, priority); + final MessageInstanceConsumer<T> consumer = wrapper.getConsumer(); + _consumer = consumer; + return consumer; + } + else + { + return null; + } + } + else + { + throw new ExistingExclusiveConsumer(); + } + } + + @Override + public Collection<? extends MessageInstanceConsumer> getConsumers() + { + return _consumer == null ? Collections.<MessageInstanceConsumer>emptySet() : Collections.singleton(_consumer); + } + + @Override + public boolean verifySessionAccess(final AMQSessionModel<?, ?> session) + { + return session.getConnectionReference() == _connectionReference; + } + + private class WrappingTarget<T extends ConsumerTarget<T>> implements ConsumerTarget<WrappingTarget<T>> + { + private final T _underlying; + private final String _address; + private MessageInstanceConsumer<T> _consumer; + + public WrappingTarget(final T target, String address) + { + _underlying = target; + _address = address; + } + + public T getUnderlying() + { + return _underlying; + } + + public MessageInstanceConsumer<T> getConsumer() + { + return _consumer; + } + + @Override + public void acquisitionRemoved(final MessageInstance node) + { + _underlying.acquisitionRemoved(node); + } + + @Override + public boolean processPending() + { + return _underlying.processPending(); + } + + @Override + public String getTargetAddress() + { + return _address; + } + + @Override + public boolean isMultiQueue() + { + return false; + } + + @Override + public void notifyWork() + { + LOGGER.debug("RG: notifyWork called"); + _underlying.notifyWork(); + } + + @Override + public void updateNotifyWorkDesired() + { + _underlying.updateNotifyWorkDesired(); + } + + @Override + public boolean isNotifyWorkDesired() + { + return _underlying.isNotifyWorkDesired(); + } + + @Override + public State getState() + { + return _underlying.getState(); + } + + @Override + public void consumerAdded(final MessageInstanceConsumer<WrappingTarget<T>> sub) + { + _consumer = new UnwrappingWrappingConsumer(sub, this); + _underlying.consumerAdded(_consumer); + } + + @Override + public ListenableFuture<Void> consumerRemoved(final MessageInstanceConsumer<WrappingTarget<T>> sub) + { + return _underlying.consumerRemoved(_consumer); + } + + @Override + public long getUnacknowledgedBytes() + { + return _underlying.getUnacknowledgedBytes(); + } + + @Override + public long getUnacknowledgedMessages() + { + return _underlying.getUnacknowledgedMessages(); + } + + @Override + public AMQSessionModel getSessionModel() + { + return _underlying.getSessionModel(); + } + + @Override + public long send(final MessageInstanceConsumer consumer, + final MessageInstance entry, + final boolean batch) + { + return _underlying.send(_consumer, entry, batch); + } + + @Override + public boolean sendNextMessage() + { + return _underlying.sendNextMessage(); + } + + @Override + public void flushBatched() + { + _underlying.flushBatched(); + } + + @Override + public void queueEmpty() + { + _underlying.queueEmpty(); + } + + @Override + public boolean allocateCredit(final ServerMessage msg) + { + return _underlying.allocateCredit(msg); + } + + @Override + public void restoreCredit(final ServerMessage queueEntry) + { + _underlying.restoreCredit(queueEntry); + } + + @Override + public boolean isSuspended() + { + return _underlying.isSuspended(); + } + + @Override + public boolean close() + { + _managementAddressSpace.removeProxyMessageSource(_connectionReference, _name); + ProxyMessageSource.this._consumer = null; + return _underlying.close(); + } + + + } + private static class UnwrappingWrappingConsumer<T extends ConsumerTarget<T>> implements MessageInstanceConsumer<T> + { + private final MessageInstanceConsumer<WrappingTarget<T>> _underlying; + private final WrappingTarget<T> _target; + + public UnwrappingWrappingConsumer(final MessageInstanceConsumer<WrappingTarget<T>> sub, WrappingTarget<T> wrappedTarget) + { + _underlying = sub; + _target = wrappedTarget; + } + + @Override + public boolean isClosed() + { + return _underlying.isClosed(); + } + + @Override + public boolean acquires() + { + return _underlying.acquires(); + } + + @Override + public String getName() + { + return _underlying.getName(); + } + + @Override + public void close() + { + _underlying.close(); + } + + @Override + public void externalStateChange() + { + _underlying.externalStateChange(); + } + + @Override + public Object getIdentifier() + { + return _underlying.getIdentifier(); + } + + @Override + public MessageContainer pullMessage() + { + return _underlying.pullMessage(); + } + + @Override + public T getTarget() + { + return _target.getUnderlying(); + } + + @Override + public void setNotifyWorkDesired(final boolean desired) + { + _underlying.setNotifyWorkDesired(desired); + } + } +} Propchange: qpid/java/trunk/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ProxyMessageSource.java ------------------------------------------------------------------------------ svn:eol-style = native Modified: qpid/java/trunk/systests/src/test/java/org/apache/qpid/systest/management/amqp/AmqpManagementTest.java URL: http://svn.apache.org/viewvc/qpid/java/trunk/systests/src/test/java/org/apache/qpid/systest/management/amqp/AmqpManagementTest.java?rev=1774263&r1=1774262&r2=1774263&view=diff ============================================================================== --- qpid/java/trunk/systests/src/test/java/org/apache/qpid/systest/management/amqp/AmqpManagementTest.java (original) +++ qpid/java/trunk/systests/src/test/java/org/apache/qpid/systest/management/amqp/AmqpManagementTest.java Wed Dec 14 15:18:33 2016 @@ -168,7 +168,7 @@ public class AmqpManagementTest extends message.setStringProperty("identity", "self"); message.setStringProperty("type", "org.amqp.management"); message.setStringProperty("operation", "QUERY"); - message.setObject("attributeNames", new ArrayList<>()); + message.setObject("attributeNames", "[]"); message.setJMSReplyTo(_replyAddress); _producer.send(message); @@ -217,7 +217,7 @@ public class AmqpManagementTest extends assertTrue("The attribute names are not a list", attributeNames instanceof Collection); attributeNamesCollection = (Collection)attributeNames; assertEquals("The attributeNames are no as expected", Arrays.asList("name", "identity", "type"), attributeNamesCollection); - Object resultsObject = ((MapMessage) responseMessage).getObject("results"); + Object resultsObject = getValueFromMapResponse(responseMessage, "results"); assertTrue("results is not a collection", resultsObject instanceof Collection); Collection results = (Collection)resultsObject; @@ -295,11 +295,11 @@ public class AmqpManagementTest extends assertTrue("The response message does not have a status code", Collections.list(responseMessage.getPropertyNames()).contains("statusCode")); assertEquals("The response code did not indicate success", 201, responseMessage.getIntProperty("statusCode")); - assertTrue("The response was not a MapMessage", responseMessage instanceof MapMessage); - assertEquals("The created queue was not a standard queue", "org.apache.qpid.StandardQueue", ((MapMessage)responseMessage).getString("type")); - assertEquals("The created queue was not a standard queue", "standard", ((MapMessage)responseMessage).getString("qpid-type")); - assertEquals("the created queue did not have the correct alerting threshold", 100L, ((MapMessage)responseMessage).getLong(ALERT_THRESHOLD_QUEUE_DEPTH_MESSAGES)); - Object identity = ((MapMessage) responseMessage).getObject("identity"); + checkResponseIsMapType(responseMessage); + assertEquals("The created queue was not a standard queue", "org.apache.qpid.StandardQueue", getValueFromMapResponse(responseMessage, "type")); + assertEquals("The created queue was not a standard queue", "standard", getValueFromMapResponse(responseMessage, "qpid-type")); + assertEquals("the created queue did not have the correct alerting threshold", 100L, getValueFromMapResponse(responseMessage,ALERT_THRESHOLD_QUEUE_DEPTH_MESSAGES)); + Object identity = getValueFromMapResponse(responseMessage,"identity"); message = _session.createMapMessage(); @@ -316,8 +316,8 @@ public class AmqpManagementTest extends assertTrue("The response message does not have a status code", Collections.list(responseMessage.getPropertyNames()).contains("statusCode")); assertEquals("The response code did not indicate success", 200, responseMessage.getIntProperty("statusCode")); - assertTrue("The response was not a MapMessage", responseMessage instanceof MapMessage); - assertEquals("the created queue did not have the correct alerting threshold", 250L, ((MapMessage)responseMessage).getLong(ALERT_THRESHOLD_QUEUE_DEPTH_MESSAGES)); + checkResponseIsMapType(responseMessage); + assertEquals("the created queue did not have the correct alerting threshold", 250L, getValueFromMapResponse(responseMessage, ALERT_THRESHOLD_QUEUE_DEPTH_MESSAGES)); message = _session.createMapMessage(); @@ -530,9 +530,9 @@ public class AmqpManagementTest extends assertTrue("The response message does not have a status code", Collections.list(responseMessage.getPropertyNames()).contains("statusCode")); assertEquals("Incorrect response code", 200, responseMessage.getIntProperty("statusCode")); - assertTrue("The response was not a MapMessage", responseMessage instanceof MapMessage); - assertEquals("The name of the virtual host is not as expected", virtualHostName, ((MapMessage)responseMessage).getString("name")); - assertEquals("The type of the virtual host is not as expected", "Memory", ((MapMessage)responseMessage).getString("qpid-type")); + checkResponseIsMapType(responseMessage); + assertEquals("The name of the virtual host is not as expected", virtualHostName, getValueFromMapResponse(responseMessage, "name")); + assertEquals("The type of the virtual host is not as expected", "Memory", getValueFromMapResponse(responseMessage, "qpid-type")); } Modified: qpid/java/trunk/test-profiles/Java10BrokenTestsExcludes URL: http://svn.apache.org/viewvc/qpid/java/trunk/test-profiles/Java10BrokenTestsExcludes?rev=1774263&r1=1774262&r2=1774263&view=diff ============================================================================== --- qpid/java/trunk/test-profiles/Java10BrokenTestsExcludes (original) +++ qpid/java/trunk/test-profiles/Java10BrokenTestsExcludes Wed Dec 14 15:18:33 2016 @@ -43,16 +43,6 @@ org.apache.qpid.systest.rest.PublishMess // this test fails - likely a client bug with the modification racing the send org.apache.qpid.test.unit.basic.BytesMessageTest#testModificationAfterSend -// AmqpManagement at the broker level needs the client to support the request/response pattern using local target names -// (since the synthetic virtual host cannot create a temporary queue). This will require the client to support this -org.apache.qpid.systest.management.amqp.AmqpManagementTest#testCreateQueueOnBrokerManagement -org.apache.qpid.systest.management.amqp.AmqpManagementTest#testCreateBindingOnBrokerManagement -org.apache.qpid.systest.management.amqp.AmqpManagementTest#testCreateConnectionOnBrokerManagement -org.apache.qpid.systest.management.amqp.AmqpManagementTest#testGetTypesOnBrokerManagement -org.apache.qpid.systest.management.amqp.AmqpManagementTest#testCreateVirtualHost -org.apache.qpid.systest.management.amqp.AmqpManagementTest#testQueryBrokerManagement - - // This test fails with error The underlying correlation-id is not binary and so can't be returned // however the correlation id in the incoming message was set as byte[] so the issue is within the conversion to // the InternalMessage and back --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@qpid.apache.org For additional commands, e-mail: commits-h...@qpid.apache.org