You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ro...@apache.org on 2012/06/28 18:46:36 UTC
svn commit: r1355072 [12/16] - in /qpid/trunk/qpid/java: ./ bdbstore/
bdbstore/jmx/ bdbstore/jmx/src/ bdbstore/jmx/src/main/
bdbstore/jmx/src/main/java/ bdbstore/jmx/src/main/java/org/
bdbstore/jmx/src/main/java/org/apache/ bdbstore/jmx/src/main/java/o...
Added: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/ExchangeAdapter.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/ExchangeAdapter.java?rev=1355072&view=auto
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/ExchangeAdapter.java (added)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/ExchangeAdapter.java Thu Jun 28 16:46:12 2012
@@ -0,0 +1,409 @@
+/*
+ *
+ * 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.model.adapter;
+
+import java.security.AccessControlException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.qpid.AMQException;
+import org.apache.qpid.AMQInternalException;
+import org.apache.qpid.AMQSecurityException;
+import org.apache.qpid.server.binding.Binding;
+import org.apache.qpid.server.model.ConfiguredObject;
+import org.apache.qpid.server.model.Exchange;
+import org.apache.qpid.server.model.LifetimePolicy;
+import org.apache.qpid.server.model.Publisher;
+import org.apache.qpid.server.model.Queue;
+import org.apache.qpid.server.model.State;
+import org.apache.qpid.server.model.Statistics;
+import org.apache.qpid.server.model.UUIDGenerator;
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.virtualhost.VirtualHost;
+
+final class ExchangeAdapter extends AbstractAdapter implements Exchange, org.apache.qpid.server.exchange.Exchange.BindingListener
+{
+
+ private final org.apache.qpid.server.exchange.Exchange _exchange;
+ private final Map<Binding, BindingAdapter> _bindingAdapters =
+ new HashMap<Binding, BindingAdapter>();
+ private VirtualHostAdapter _vhost;
+ private final ExchangeStatistics _statistics;
+
+
+ public ExchangeAdapter(final VirtualHostAdapter virtualHostAdapter,
+ final org.apache.qpid.server.exchange.Exchange exchange)
+ {
+ super(virtualHostAdapter.getName(), exchange.getName());
+ _statistics = new ExchangeStatistics();
+ _vhost = virtualHostAdapter;
+ _exchange = exchange;
+ addParent(org.apache.qpid.server.model.VirtualHost.class, virtualHostAdapter);
+
+ exchange.addBindingListener(this);
+ populateBindings();
+ }
+
+ private void populateBindings()
+ {
+ Collection<Binding> actualBindings = _exchange.getBindings();
+ synchronized (_bindingAdapters)
+ {
+ for(Binding binding : actualBindings)
+ {
+ if(!_bindingAdapters.containsKey(binding))
+ {
+ QueueAdapter queueAdapter = _vhost.getQueueAdapter(binding.getQueue());
+ BindingAdapter adapter = new BindingAdapter(binding, this, queueAdapter);
+ _bindingAdapters.put(binding, adapter);
+
+ queueAdapter.bindingRegistered(binding, adapter);
+ }
+ }
+ }
+
+ }
+
+ public String getExchangeType()
+ {
+ return _exchange.getType().getName().toString();
+ }
+
+ public Collection<org.apache.qpid.server.model.Binding> getBindings()
+ {
+ synchronized (_bindingAdapters)
+ {
+ return new ArrayList<org.apache.qpid.server.model.Binding>(_bindingAdapters.values());
+ }
+
+ }
+
+ public Collection<Publisher> getPublishers()
+ {
+ // TODO
+ return Collections.emptyList();
+ }
+
+
+ public org.apache.qpid.server.model.Binding createBinding(Queue queue,
+ Map<String, Object> attributes)
+ throws AccessControlException, IllegalStateException
+ {
+ attributes = new HashMap<String, Object>(attributes);
+ String bindingKey = getStringAttribute(org.apache.qpid.server.model.Binding.NAME, attributes, "");
+ Map<String, Object> bindingArgs = getMapAttribute(org.apache.qpid.server.model.Binding.ARGUMENTS, attributes, Collections.EMPTY_MAP);
+
+ attributes.remove(org.apache.qpid.server.model.Binding.NAME);
+ attributes.remove(org.apache.qpid.server.model.Binding.ARGUMENTS);
+
+ return createBinding(bindingKey, queue, bindingArgs, attributes);
+
+ }
+
+ public org.apache.qpid.server.model.Binding createBinding(String bindingKey, Queue queue,
+ Map<String, Object> bindingArguments,
+ Map<String, Object> attributes)
+ throws AccessControlException, IllegalStateException
+ {
+ VirtualHost virtualHost = _vhost.getVirtualHost();
+
+
+ AMQQueue amqQueue = ((QueueAdapter)queue).getAMQQueue();
+
+ try
+ {
+ if(!virtualHost.getBindingFactory().addBinding(bindingKey, amqQueue, _exchange, bindingArguments))
+ {
+ Binding oldBinding = virtualHost.getBindingFactory().getBinding(bindingKey, amqQueue, _exchange,
+ bindingArguments);
+
+ Map<String, Object> oldArgs = oldBinding.getArguments();
+ if((oldArgs == null && !bindingArguments.isEmpty()) || (oldArgs != null && !oldArgs.equals(bindingArguments)))
+ {
+ //TODO: generate deterministic UUID
+ virtualHost.getBindingFactory().replaceBinding(UUIDGenerator.generateUUID(), bindingKey, amqQueue, _exchange, bindingArguments);
+ }
+ }
+ Binding binding = virtualHost.getBindingFactory().getBinding(bindingKey, amqQueue, _exchange, bindingArguments);
+
+ synchronized (_bindingAdapters)
+ {
+ return binding == null ? null : _bindingAdapters.get(binding);
+ }
+ }
+ catch(AMQSecurityException e)
+ {
+ throw new AccessControlException(e.toString());
+ }
+ catch(AMQInternalException e)
+ {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ public void delete()
+ {
+ try
+ {
+ _vhost.getVirtualHost().getExchangeRegistry().unregisterExchange(getName(), false);
+ }
+ catch(AMQException e)
+ {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ public String getName()
+ {
+ return _exchange.getName();
+ }
+
+ public String setName(final String currentName, final String desiredName)
+ throws IllegalStateException, AccessControlException
+ {
+ return null; //TODO
+ }
+
+ public State getActualState()
+ {
+ return null; //TODO
+ }
+
+ public boolean isDurable()
+ {
+ return _exchange.isDurable();
+ }
+
+ public void setDurable(final boolean durable)
+ throws IllegalStateException, AccessControlException, IllegalArgumentException
+ {
+ //TODO
+ }
+
+ public LifetimePolicy getLifetimePolicy()
+ {
+ return _exchange.isAutoDelete() ? LifetimePolicy.AUTO_DELETE : LifetimePolicy.PERMANENT;
+ }
+
+ public LifetimePolicy setLifetimePolicy(final LifetimePolicy expected, final LifetimePolicy desired)
+ throws IllegalStateException, AccessControlException, IllegalArgumentException
+ {
+ return null; //TODO
+ }
+
+ public long getTimeToLive()
+ {
+ return 0; //TODO
+ }
+
+ public long setTimeToLive(final long expected, final long desired)
+ throws IllegalStateException, AccessControlException, IllegalArgumentException
+ {
+ return 0; //TODO
+ }
+
+ public Statistics getStatistics()
+ {
+ return _statistics;
+ }
+
+ @Override
+ public <C extends ConfiguredObject> Collection<C> getChildren(Class<C> clazz)
+ {
+ if(clazz == org.apache.qpid.server.model.Binding.class)
+ {
+ return (Collection<C>) getBindings();
+ }
+ else
+ {
+ return Collections.emptySet();
+ }
+ }
+
+ @Override
+ public <C extends ConfiguredObject> C createChild(Class<C> childClass, Map<String, Object> attributes, ConfiguredObject... otherParents)
+ {
+ if(childClass == org.apache.qpid.server.model.Binding.class)
+ {
+ if(otherParents != null && otherParents.length == 1 && otherParents[0] instanceof Queue)
+ {
+ Queue queue = (Queue) otherParents[0];
+ if(queue.getParent(org.apache.qpid.server.model.VirtualHost.class) == getParent(org.apache.qpid.server.model.VirtualHost.class))
+ {
+ return (C) createBinding(queue, attributes);
+ }
+ else
+ {
+ throw new IllegalArgumentException("Queue and Exchange parents of a binding must be on same virtual host");
+ }
+ }
+ else
+ {
+ throw new IllegalArgumentException("Other parent must be a queue");
+ }
+ }
+ else
+ {
+ throw new IllegalArgumentException();
+ }
+ }
+
+ public void bindingAdded(org.apache.qpid.server.exchange.Exchange exchange, Binding binding)
+ {
+ BindingAdapter adapter = null;
+ synchronized (_bindingAdapters)
+ {
+ if(!_bindingAdapters.containsKey(binding))
+ {
+ QueueAdapter queueAdapter = _vhost.getQueueAdapter(binding.getQueue());
+ adapter = new BindingAdapter(binding, this, queueAdapter);
+ _bindingAdapters.put(binding,adapter);
+ queueAdapter.bindingRegistered(binding,adapter);
+ }
+ }
+ if(adapter != null)
+ {
+ childAdded(adapter);
+ }
+ }
+
+ public void bindingRemoved(org.apache.qpid.server.exchange.Exchange exchange, Binding binding)
+ {
+ BindingAdapter adapter = null;
+ synchronized (_bindingAdapters)
+ {
+ adapter = _bindingAdapters.remove(binding);
+ }
+ if(adapter != null)
+ {
+ _vhost.getQueueAdapter(binding.getQueue()).bindingUnregistered(binding);
+ childRemoved(adapter);
+ }
+ }
+
+ org.apache.qpid.server.exchange.Exchange getExchange()
+ {
+ return _exchange;
+ }
+
+ @Override
+ public Object getAttribute(String name)
+ {
+ if(ID.equals(name))
+ {
+ return getId();
+ }
+ else if(NAME.equals(name))
+ {
+ return getName();
+ }
+ else if(STATE.equals(name))
+ {
+ return State.ACTIVE;
+ }
+ else if(DURABLE.equals(name))
+ {
+ return isDurable();
+ }
+ else if(LIFETIME_POLICY.equals(name))
+ {
+ return _exchange.isAutoDelete() ? LifetimePolicy.AUTO_DELETE : LifetimePolicy.PERMANENT;
+ }
+ else if(TIME_TO_LIVE.equals(name))
+ {
+
+ }
+ else if(CREATED.equals(name))
+ {
+
+ }
+ else if(UPDATED.equals(name))
+ {
+
+ }
+ else if(ALTERNATE_EXCHANGE.equals(name))
+ {
+ return _exchange.getAlternateExchange();
+ }
+ else if(TYPE.equals(name))
+ {
+ return _exchange.getType().getName().asString();
+ }
+ return super.getAttribute(name);
+ }
+
+ @Override
+ public Object setAttribute(String name, Object expected, Object desired)
+ throws IllegalStateException, AccessControlException, IllegalArgumentException
+ {
+ return super.setAttribute(name, expected, desired); //TODO - Implement
+ }
+
+ @Override
+ public Collection<String> getAttributeNames()
+ {
+ return AVAILABLE_ATTRIBUTES;
+ }
+
+ private class ExchangeStatistics implements Statistics
+ {
+
+ public Collection<String> getStatisticNames()
+ {
+ return AVAILABLE_STATISTICS;
+ }
+
+ public Object getStatistic(String name)
+ {
+ if(BINDING_COUNT.equals(name))
+ {
+ return _exchange.getBindingCount();
+ }
+ else if(BYTES_DROPPED.equals(name))
+ {
+ return _exchange.getByteDrops();
+ }
+ else if(BYTES_IN.equals(name))
+ {
+ return _exchange.getByteReceives();
+ }
+ else if(MESSAGES_DROPPED.equals(name))
+ {
+ return _exchange.getMsgDrops();
+ }
+ else if(MESSAGES_IN.equals(name))
+ {
+ return _exchange.getMsgReceives();
+ }
+ else if(PRODUCER_COUNT.equals(name))
+ {
+
+ }
+ else if(STATE_CHANGED.equals(name))
+ {
+
+ }
+ return null; // TODO - Implement
+ }
+ }
+}
Added: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/HTTPPortAdapter.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/HTTPPortAdapter.java?rev=1355072&view=auto
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/HTTPPortAdapter.java (added)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/HTTPPortAdapter.java Thu Jun 28 16:46:12 2012
@@ -0,0 +1,262 @@
+/*
+ *
+ * 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.model.adapter;
+
+import java.security.AccessControlException;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Map;
+import org.apache.qpid.server.model.ConfiguredObject;
+import org.apache.qpid.server.model.Connection;
+import org.apache.qpid.server.model.LifetimePolicy;
+import org.apache.qpid.server.model.Port;
+import org.apache.qpid.server.model.Protocol;
+import org.apache.qpid.server.model.State;
+import org.apache.qpid.server.model.Statistics;
+import org.apache.qpid.server.model.Transport;
+import org.apache.qpid.server.model.VirtualHostAlias;
+
+public class HTTPPortAdapter extends AbstractAdapter implements Port
+{
+ private final BrokerAdapter _broker;
+ private int _port;
+ public HTTPPortAdapter(BrokerAdapter brokerAdapter, int port)
+ {
+ _broker = brokerAdapter;
+ _port = port;
+
+ }
+
+ @Override
+ public String getBindingAddress()
+ {
+ return "0.0.0.0";
+ }
+
+ @Override
+ public int getPort()
+ {
+ return _port;
+ }
+
+ @Override
+ public Collection<Transport> getTransports()
+ {
+ return Collections.singleton(Transport.TCP);
+ }
+
+ @Override
+ public void addTransport(Transport transport)
+ throws IllegalStateException, AccessControlException, IllegalArgumentException
+ {
+ throw new IllegalStateException(); // TODO - Implement
+ }
+
+ @Override
+ public Transport removeTransport(Transport transport)
+ throws IllegalStateException, AccessControlException, IllegalArgumentException
+ {
+ throw new IllegalStateException(); // TODO - Implement
+ }
+
+ @Override
+ public Collection<Protocol> getProtocols()
+ {
+ return Collections.singleton(Protocol.HTTP);
+ }
+
+ @Override
+ public void addProtocol(Protocol protocol)
+ throws IllegalStateException, AccessControlException, IllegalArgumentException
+ {
+ throw new IllegalStateException(); // TODO - Implement
+ }
+
+ @Override
+ public Protocol removeProtocol(Protocol protocol)
+ throws IllegalStateException, AccessControlException, IllegalArgumentException
+ {
+ throw new IllegalStateException(); // TODO - Implement
+ }
+
+ @Override
+ public Collection<VirtualHostAlias> getVirtualHostBindings()
+ {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public Collection<Connection> getConnections()
+ {
+ return Collections.emptySet(); // TODO - Implement
+ }
+
+ @Override
+ public String getName()
+ {
+ return getBindingAddress() + ":" + getPort(); // TODO - Implement
+ }
+
+ @Override
+ public String setName(String currentName, String desiredName) throws IllegalStateException, AccessControlException
+ {
+ throw new IllegalStateException(); // TODO - Implement
+ }
+
+ @Override
+ public State getActualState()
+ {
+ return State.ACTIVE;
+ }
+
+ @Override
+ public boolean isDurable()
+ {
+ return false; // TODO - Implement
+ }
+
+ @Override
+ public void setDurable(boolean durable)
+ throws IllegalStateException, AccessControlException, IllegalArgumentException
+ {
+ throw new IllegalStateException();
+ }
+
+ @Override
+ public LifetimePolicy getLifetimePolicy()
+ {
+ return LifetimePolicy.PERMANENT;
+ }
+
+ @Override
+ public LifetimePolicy setLifetimePolicy(LifetimePolicy expected, LifetimePolicy desired)
+ throws IllegalStateException, AccessControlException, IllegalArgumentException
+ {
+ throw new IllegalStateException(); // TODO - Implement
+ }
+
+ @Override
+ public long getTimeToLive()
+ {
+ return 0; // TODO - Implement
+ }
+
+ @Override
+ public long setTimeToLive(long expected, long desired)
+ throws IllegalStateException, AccessControlException, IllegalArgumentException
+ {
+ throw new IllegalStateException(); // TODO - Implement
+ }
+
+ @Override
+ public Statistics getStatistics()
+ {
+ return NoStatistics.getInstance();
+ }
+
+ @Override
+ public <C extends ConfiguredObject> Collection<C> getChildren(Class<C> clazz)
+ {
+ if(clazz == Connection.class)
+ {
+ return (Collection<C>) getConnections();
+ }
+ else
+ {
+ return Collections.emptySet();
+ }
+ }
+
+ @Override
+ public <C extends ConfiguredObject> C createChild(Class<C> childClass, Map<String, Object> attributes, ConfiguredObject... otherParents)
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Object getAttribute(String name)
+ {
+ if(ID.equals(name))
+ {
+ return getId();
+ }
+ else if(NAME.equals(name))
+ {
+ return getName();
+ }
+ else if(STATE.equals(name))
+ {
+ return getActualState();
+ }
+ else if(DURABLE.equals(name))
+ {
+ return isDurable();
+ }
+ else if(LIFETIME_POLICY.equals(name))
+ {
+ return getLifetimePolicy();
+ }
+ else if(TIME_TO_LIVE.equals(name))
+ {
+ return getTimeToLive();
+ }
+ else if(CREATED.equals(name))
+ {
+
+ }
+ else if(UPDATED.equals(name))
+ {
+
+ }
+ else if(BINDING_ADDRESS.equals(name))
+ {
+ return getBindingAddress();
+ }
+ else if(PORT.equals(name))
+ {
+ return getPort();
+ }
+ else if(PROTOCOLS.equals(name))
+ {
+ return getProtocols();
+ }
+ else if(TRANSPORTS.equals(name))
+ {
+ return getTransports();
+ }
+
+ return super.getAttribute(name); //TODO - Implement
+ }
+
+ @Override
+ public Collection<String> getAttributeNames()
+ {
+ return AVAILABLE_ATTRIBUTES;
+ }
+
+ @Override
+ public Object setAttribute(String name, Object expected, Object desired)
+ throws IllegalStateException, AccessControlException, IllegalArgumentException
+ {
+ return super.setAttribute(name, expected, desired); //TODO - Implement
+ }
+}
Copied: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/NoStatistics.java (from r1354912, qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/Statistics.java)
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/NoStatistics.java?p2=qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/NoStatistics.java&p1=qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/Statistics.java&r1=1354912&r2=1355072&rev=1355072&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/Statistics.java (original)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/NoStatistics.java Thu Jun 28 16:46:12 2012
@@ -1,6 +1,9 @@
-package org.apache.qpid.server.model;
+package org.apache.qpid.server.model.adapter;
+
+import org.apache.qpid.server.model.Statistics;
import java.util.Collection;
+import java.util.Collections;
/**
* Licensed to the Apache Software Foundation (ASF) under one or more
@@ -18,8 +21,26 @@ import java.util.Collection;
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-public interface Statistics
+public class NoStatistics implements Statistics
{
- Collection<String> getStatisticNames();
- public Object getStatistic(String name);
+ private static final NoStatistics INSTANCE = new NoStatistics();
+
+ private NoStatistics()
+ {
+ }
+
+ public Collection<String> getStatisticNames()
+ {
+ return Collections.emptyList();
+ }
+
+ public Object getStatistic(String name)
+ {
+ return null;
+ }
+
+ public static NoStatistics getInstance()
+ {
+ return INSTANCE;
+ }
}
Added: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/PortAdapter.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/PortAdapter.java?rev=1355072&view=auto
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/PortAdapter.java (added)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/PortAdapter.java Thu Jun 28 16:46:12 2012
@@ -0,0 +1,318 @@
+/*
+ *
+ * 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.model.adapter;
+
+import java.util.Map;
+import org.apache.qpid.server.model.ConfiguredObject;
+import org.apache.qpid.server.model.Connection;
+import org.apache.qpid.server.model.LifetimePolicy;
+import org.apache.qpid.server.model.Port;
+import org.apache.qpid.server.model.Protocol;
+import org.apache.qpid.server.model.State;
+import org.apache.qpid.server.model.Statistics;
+import org.apache.qpid.server.model.Transport;
+import org.apache.qpid.server.model.VirtualHost;
+import org.apache.qpid.server.model.VirtualHostAlias;
+import org.apache.qpid.server.protocol.AmqpProtocolVersion;
+import org.apache.qpid.server.transport.QpidAcceptor;
+
+import java.net.InetSocketAddress;
+import java.security.AccessControlException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+
+public class PortAdapter extends AbstractAdapter implements Port
+{
+ private final BrokerAdapter _broker;
+ private final QpidAcceptor _acceptor;
+ private final InetSocketAddress _address;
+ private final Collection<Protocol> _protocols;
+
+ public PortAdapter(BrokerAdapter brokerAdapter, QpidAcceptor acceptor, InetSocketAddress address)
+ {
+ _broker = brokerAdapter;
+ _acceptor = acceptor;
+ _address = address;
+
+ List<Protocol> protocols = new ArrayList<Protocol>();
+
+ for(AmqpProtocolVersion pv : _acceptor.getSupported())
+ {
+ switch(pv)
+ {
+ case v0_8:
+ protocols.add(Protocol.AMQP_0_8);
+ break;
+ case v0_9:
+ protocols.add(Protocol.AMQP_0_9);
+ break;
+ case v0_9_1:
+ protocols.add(Protocol.AMQP_0_9_1);
+ break;
+ case v0_10:
+ protocols.add(Protocol.AMQP_0_10);
+ break;
+ case v1_0_0:
+ protocols.add(Protocol.AMQP_1_0);
+ break;
+ }
+ }
+
+ _protocols = Collections.unmodifiableCollection(protocols);
+
+ }
+
+ @Override
+ public String getBindingAddress()
+ {
+ return _address.getHostName();
+ }
+
+ @Override
+ public int getPort()
+ {
+ return _address.getPort();
+ }
+
+ @Override
+ public Collection<Transport> getTransports()
+ {
+ switch (_acceptor.getTransport())
+ {
+ case TCP:
+ return Collections.singleton(Transport.TCP);
+ case SSL:
+ return Collections.singleton(Transport.SSL);
+ }
+
+ return null; // TODO - Implement
+ }
+
+ @Override
+ public void addTransport(Transport transport)
+ throws IllegalStateException, AccessControlException, IllegalArgumentException
+ {
+ throw new IllegalStateException(); // TODO - Implement
+ }
+
+ @Override
+ public Transport removeTransport(Transport transport)
+ throws IllegalStateException, AccessControlException, IllegalArgumentException
+ {
+ throw new IllegalStateException(); // TODO - Implement
+ }
+
+ @Override
+ public Collection<Protocol> getProtocols()
+ {
+ return _protocols;
+ }
+
+ @Override
+ public void addProtocol(Protocol protocol)
+ throws IllegalStateException, AccessControlException, IllegalArgumentException
+ {
+ throw new IllegalStateException(); // TODO - Implement
+ }
+
+ @Override
+ public Protocol removeProtocol(Protocol protocol)
+ throws IllegalStateException, AccessControlException, IllegalArgumentException
+ {
+ throw new IllegalStateException(); // TODO - Implement
+ }
+
+ @Override
+ public Collection<VirtualHostAlias> getVirtualHostBindings()
+ {
+ List<VirtualHostAlias> aliases = new ArrayList<VirtualHostAlias>();
+ for(VirtualHost vh : _broker.getVirtualHosts())
+ {
+ for(VirtualHostAlias alias : vh.getAliases())
+ {
+ if(alias.getPort().equals(this))
+ {
+ aliases.add(alias);
+ }
+ }
+ }
+ return Collections.unmodifiableCollection(aliases);
+ }
+
+ @Override
+ public Collection<Connection> getConnections()
+ {
+ return null; // TODO - Implement
+ }
+
+ @Override
+ public String getName()
+ {
+ return getBindingAddress() + ":" + getPort(); // TODO - Implement
+ }
+
+ @Override
+ public String setName(String currentName, String desiredName) throws IllegalStateException, AccessControlException
+ {
+ throw new IllegalStateException(); // TODO - Implement
+ }
+
+ @Override
+ public State getActualState()
+ {
+ return State.ACTIVE;
+ }
+
+ @Override
+ public boolean isDurable()
+ {
+ return false; // TODO - Implement
+ }
+
+ @Override
+ public void setDurable(boolean durable)
+ throws IllegalStateException, AccessControlException, IllegalArgumentException
+ {
+ throw new IllegalStateException();
+ }
+
+ @Override
+ public LifetimePolicy getLifetimePolicy()
+ {
+ return LifetimePolicy.PERMANENT;
+ }
+
+ @Override
+ public LifetimePolicy setLifetimePolicy(LifetimePolicy expected, LifetimePolicy desired)
+ throws IllegalStateException, AccessControlException, IllegalArgumentException
+ {
+ throw new IllegalStateException(); // TODO - Implement
+ }
+
+ @Override
+ public long getTimeToLive()
+ {
+ return 0; // TODO - Implement
+ }
+
+ @Override
+ public long setTimeToLive(long expected, long desired)
+ throws IllegalStateException, AccessControlException, IllegalArgumentException
+ {
+ throw new IllegalStateException(); // TODO - Implement
+ }
+
+ @Override
+ public Statistics getStatistics()
+ {
+ return NoStatistics.getInstance();
+ }
+
+ @Override
+ public <C extends ConfiguredObject> Collection<C> getChildren(Class<C> clazz)
+ {
+ if(clazz == Connection.class)
+ {
+ return (Collection<C>) getConnections();
+ }
+ else
+ {
+ return Collections.emptySet();
+ }
+ }
+
+ @Override
+ public <C extends ConfiguredObject> C createChild(Class<C> childClass, Map<String, Object> attributes, ConfiguredObject... otherParents)
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Object getAttribute(String name)
+ {
+ if(ID.equals(name))
+ {
+ return getId();
+ }
+ else if(NAME.equals(name))
+ {
+ return getName();
+ }
+ else if(STATE.equals(name))
+ {
+ return getActualState();
+ }
+ else if(DURABLE.equals(name))
+ {
+ return isDurable();
+ }
+ else if(LIFETIME_POLICY.equals(name))
+ {
+ return getLifetimePolicy();
+ }
+ else if(TIME_TO_LIVE.equals(name))
+ {
+ return getTimeToLive();
+ }
+ else if(CREATED.equals(name))
+ {
+
+ }
+ else if(UPDATED.equals(name))
+ {
+
+ }
+ else if(BINDING_ADDRESS.equals(name))
+ {
+ return getBindingAddress();
+ }
+ else if(PORT.equals(name))
+ {
+ return getPort();
+ }
+ else if(PROTOCOLS.equals(name))
+ {
+ return getProtocols();
+ }
+ else if(TRANSPORTS.equals(name))
+ {
+ return getTransports();
+ }
+
+ return super.getAttribute(name); //TODO - Implement
+ }
+
+ @Override
+ public Collection<String> getAttributeNames()
+ {
+ return AVAILABLE_ATTRIBUTES;
+ }
+
+ @Override
+ public Object setAttribute(String name, Object expected, Object desired)
+ throws IllegalStateException, AccessControlException, IllegalArgumentException
+ {
+ return super.setAttribute(name, expected, desired); //TODO - Implement
+ }
+}
Added: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/QueueAdapter.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/QueueAdapter.java?rev=1355072&view=auto
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/QueueAdapter.java (added)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/QueueAdapter.java Thu Jun 28 16:46:12 2012
@@ -0,0 +1,702 @@
+/*
+ *
+ * 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.model.adapter;
+
+import java.security.AccessControlException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import org.apache.qpid.AMQException;
+import org.apache.qpid.AMQStoreException;
+import org.apache.qpid.server.binding.Binding;
+import org.apache.qpid.server.model.ConfiguredObject;
+import org.apache.qpid.server.model.ConfiguredObjectFinder;
+import org.apache.qpid.server.model.Consumer;
+import org.apache.qpid.server.model.Exchange;
+import org.apache.qpid.server.model.LifetimePolicy;
+import org.apache.qpid.server.model.Queue;
+import org.apache.qpid.server.model.QueueNotificationListener;
+import org.apache.qpid.server.model.State;
+import org.apache.qpid.server.model.Statistics;
+import org.apache.qpid.server.queue.*;
+import org.apache.qpid.server.subscription.Subscription;
+
+final class QueueAdapter extends AbstractAdapter implements Queue, AMQQueue.SubscriptionRegistrationListener, AMQQueue.NotificationListener
+{
+
+ static final Map<String, String> ATTRIBUTE_MAPPINGS = new HashMap<String, String>();
+ static
+ {
+ QueueAdapter.ATTRIBUTE_MAPPINGS.put(Queue.ALERT_REPEAT_GAP, "x-qpid-minimum-alert-repeat-gap");
+ QueueAdapter.ATTRIBUTE_MAPPINGS.put(Queue.ALERT_THRESHOLD_MESSAGE_AGE, "x-qpid-maximum-message-size");
+ QueueAdapter.ATTRIBUTE_MAPPINGS.put(Queue.ALERT_THRESHOLD_MESSAGE_SIZE, "x-qpid-maximum-message-age");
+ QueueAdapter.ATTRIBUTE_MAPPINGS.put(Queue.ALERT_THRESHOLD_QUEUE_DEPTH_MESSAGES, "x-qpid-maximum-message-count");
+
+ QueueAdapter.ATTRIBUTE_MAPPINGS.put(Queue.MAXIMUM_DELIVERY_ATTEMPTS, "x-qpid-maximum-delivery-count");
+
+ QueueAdapter.ATTRIBUTE_MAPPINGS.put(Queue.QUEUE_FLOW_CONTROL_SIZE_BYTES, "x-qpid-capacity");
+ QueueAdapter.ATTRIBUTE_MAPPINGS.put(Queue.QUEUE_FLOW_RESUME_SIZE_BYTES, "x-qpid-flow-resume-capacity");
+
+ QueueAdapter.ATTRIBUTE_MAPPINGS.put(Queue.SORT_KEY, "qpid.sort_key");
+ QueueAdapter.ATTRIBUTE_MAPPINGS.put(Queue.LVQ_KEY, "qpid.last_value_queue_key");
+
+ }
+
+ private final AMQQueue _queue;
+ private final Map<Binding, BindingAdapter> _bindingAdapters =
+ new HashMap<Binding, BindingAdapter>();
+ private Map<org.apache.qpid.server.subscription.Subscription, ConsumerAdapter> _consumerAdapters =
+ new HashMap<org.apache.qpid.server.subscription.Subscription, ConsumerAdapter>();
+
+
+ private final VirtualHostAdapter _vhost;
+ private QueueStatisticsAdapter _statistics;
+ private QueueNotificationListener _queueNotificationListener;
+
+ public QueueAdapter(final VirtualHostAdapter virtualHostAdapter, final AMQQueue queue)
+ {
+ super(virtualHostAdapter.getName(), queue.getName());
+ _vhost = virtualHostAdapter;
+ addParent(org.apache.qpid.server.model.VirtualHost.class, virtualHostAdapter);
+
+ _queue = queue;
+ _queue.addSubscriptionRegistrationListener(this);
+ populateConsumers();
+ _statistics = new QueueStatisticsAdapter(queue);
+ _queue.setNotificationListener(this);
+ }
+
+ private void populateConsumers()
+ {
+ Collection<org.apache.qpid.server.subscription.Subscription> actualSubscriptions = _queue.getConsumers();
+
+ synchronized (_consumerAdapters)
+ {
+ Iterator<org.apache.qpid.server.subscription.Subscription> iter = _consumerAdapters.keySet().iterator();
+ for(org.apache.qpid.server.subscription.Subscription subscription : actualSubscriptions)
+ {
+ if(!_consumerAdapters.containsKey(subscription))
+ {
+ _consumerAdapters.put(subscription, new ConsumerAdapter(this, subscription));
+ }
+ }
+ }
+ }
+
+ public Collection<org.apache.qpid.server.model.Binding> getBindings()
+ {
+ synchronized (_bindingAdapters)
+ {
+ return new ArrayList<org.apache.qpid.server.model.Binding>(_bindingAdapters.values());
+ }
+ }
+
+ public Collection<Consumer> getConsumers()
+ {
+ synchronized (_consumerAdapters)
+ {
+ return new ArrayList<Consumer>(_consumerAdapters.values());
+ }
+
+ }
+
+ public void visit(final QueueEntryVisitor visitor)
+ {
+ _queue.visit(visitor);
+ }
+
+ public void delete()
+ {
+ try
+ {
+ _queue.delete();
+ if (_queue.isDurable())
+ {
+
+ _queue.getVirtualHost().getMessageStore().removeQueue(_queue);
+ }
+ }
+ catch(AMQException e)
+ {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ public String getName()
+ {
+ return _queue.getName();
+ }
+
+ public String setName(final String currentName, final String desiredName)
+ throws IllegalStateException, AccessControlException
+ {
+ return null; //TODO
+ }
+
+ public State getActualState()
+ {
+ return null; //TODO
+ }
+
+ public boolean isDurable()
+ {
+ return _queue.isDurable();
+ }
+
+ public void setDurable(final boolean durable)
+ throws IllegalStateException, AccessControlException, IllegalArgumentException
+ {
+ //TODO
+ }
+
+ public LifetimePolicy getLifetimePolicy()
+ {
+ return _queue.isAutoDelete() ? LifetimePolicy.AUTO_DELETE : LifetimePolicy.PERMANENT;
+ }
+
+ public LifetimePolicy setLifetimePolicy(final LifetimePolicy expected, final LifetimePolicy desired)
+ throws IllegalStateException, AccessControlException, IllegalArgumentException
+ {
+ return null; //TODO
+ }
+
+ public long getTimeToLive()
+ {
+ return 0; //TODO
+ }
+
+ public long setTimeToLive(final long expected, final long desired)
+ throws IllegalStateException, AccessControlException, IllegalArgumentException
+ {
+ return 0; //TODO
+ }
+
+
+ @Override
+ public Collection<String> getAttributeNames()
+ {
+ return Queue.AVAILABLE_ATTRIBUTES;
+ }
+
+ @Override
+ public Object setAttribute(String name, Object expected, Object desired) throws IllegalStateException, AccessControlException, IllegalArgumentException
+ {
+ try
+ {
+ if(ALERT_REPEAT_GAP.equals(name))
+ {
+ _queue.setMinimumAlertRepeatGap((Long)desired);
+ return desired;
+ }
+ else if(ALERT_THRESHOLD_MESSAGE_AGE.equals(name))
+ {
+ _queue.setMaximumMessageAge((Long)desired);
+ return desired;
+ }
+ else if(ALERT_THRESHOLD_MESSAGE_SIZE.equals(name))
+ {
+ _queue.setMaximumMessageSize((Long)desired);
+ return desired;
+ }
+ else if(ALERT_THRESHOLD_QUEUE_DEPTH_BYTES.equals(name))
+ {
+ _queue.setMaximumQueueDepth((Long)desired);
+ return desired;
+ }
+ else if(ALERT_THRESHOLD_QUEUE_DEPTH_MESSAGES.equals(name))
+ {
+ _queue.setMaximumMessageCount((Long)desired);
+ return desired;
+ }
+ else if(ALTERNATE_EXCHANGE.equals(name))
+ {
+ // In future we may want to accept a UUID as an alternative way to identifying the exchange
+ ExchangeAdapter alternateExchange = (ExchangeAdapter) desired;
+ _queue.setAlternateExchange(alternateExchange == null ? null : alternateExchange.getExchange());
+ return desired;
+ }
+ else if(EXCLUSIVE.equals(name))
+ {
+ Boolean exclusiveFlag = (Boolean) desired;
+ _queue.setExclusive(exclusiveFlag);
+ return desired;
+ }
+ else if(MESSAGE_GROUP_KEY.equals(name))
+ {
+ // TODO
+ }
+ else if(MESSAGE_GROUP_DEFAULT_GROUP.equals(name))
+ {
+ // TODO
+ }
+ else if(MESSAGE_GROUP_SHARED_GROUPS.equals(name))
+ {
+ // TODO
+ }
+ else if(LVQ_KEY.equals(name))
+ {
+ // TODO
+ }
+ else if(MAXIMUM_DELIVERY_ATTEMPTS.equals(name))
+ {
+ _queue.setMaximumDeliveryCount((Integer)desired);
+ return desired;
+ }
+ else if(NO_LOCAL.equals(name))
+ {
+ // TODO
+ }
+ else if(OWNER.equals(name))
+ {
+ // TODO
+ }
+ else if(QUEUE_FLOW_CONTROL_SIZE_BYTES.equals(name))
+ {
+ _queue.setCapacity((Long)desired);
+ return desired;
+ }
+ else if(QUEUE_FLOW_RESUME_SIZE_BYTES.equals(name))
+ {
+ _queue.setFlowResumeCapacity((Long)desired);
+ return desired;
+ }
+ else if(QUEUE_FLOW_STOPPED.equals(name))
+ {
+ // TODO
+ }
+ else if(SORT_KEY.equals(name))
+ {
+ // TODO
+ }
+ else if(TYPE.equals(name))
+ {
+ // TODO
+ }
+ else if (DESCRIPTION.equals(name))
+ {
+ _queue.setDescription((String) desired);
+ return desired;
+ }
+
+ return super.setAttribute(name, expected, desired);
+ }
+ finally
+ {
+ if (_queue.isDurable())
+ {
+ try
+ {
+ _queue.getVirtualHost().getMessageStore().updateQueue(_queue);
+ }
+ catch (AMQStoreException e)
+ {
+ throw new IllegalStateException(e);
+ }
+ }
+ }
+ }
+
+ @Override
+ public Object getAttribute(String name)
+ {
+
+ if(ALERT_REPEAT_GAP.equals(name))
+ {
+ return _queue.getMinimumAlertRepeatGap();
+ }
+ else if(ALERT_THRESHOLD_MESSAGE_AGE.equals(name))
+ {
+ return _queue.getMaximumMessageAge();
+ }
+ else if(ALERT_THRESHOLD_MESSAGE_SIZE.equals(name))
+ {
+ return _queue.getMaximumMessageSize();
+ }
+ else if(ALERT_THRESHOLD_QUEUE_DEPTH_BYTES.equals(name))
+ {
+ return _queue.getMaximumQueueDepth();
+ }
+ else if(ALERT_THRESHOLD_QUEUE_DEPTH_MESSAGES.equals(name))
+ {
+ return _queue.getMaximumMessageCount();
+ }
+ else if(ALTERNATE_EXCHANGE.equals(name))
+ {
+ org.apache.qpid.server.exchange.Exchange alternateExchange = _queue.getAlternateExchange();
+ return alternateExchange == null ? null :
+ ConfiguredObjectFinder.findConfiguredObjectByName(_vhost.getExchanges(),
+ alternateExchange.getName());
+ }
+ else if(EXCLUSIVE.equals(name))
+ {
+ return _queue.isExclusive();
+ }
+ else if(MESSAGE_GROUP_KEY.equals(name))
+ {
+ // TODO
+ }
+ else if(MESSAGE_GROUP_DEFAULT_GROUP.equals(name))
+ {
+ // TODO
+ }
+ else if(MESSAGE_GROUP_SHARED_GROUPS.equals(name))
+ {
+ // TODO
+ }
+ else if(LVQ_KEY.equals(name))
+ {
+ if(_queue instanceof ConflationQueue)
+ {
+ return ((ConflationQueue)_queue).getConflationKey();
+ }
+ }
+ else if(MAXIMUM_DELIVERY_ATTEMPTS.equals(name))
+ {
+ return _queue.getMaximumDeliveryCount();
+ }
+ else if(NO_LOCAL.equals(name))
+ {
+ // TODO
+ }
+ else if(OWNER.equals(name))
+ {
+ return _queue.getOwner() == null ? null : _queue.getOwner().asString();
+ }
+ else if(QUEUE_FLOW_CONTROL_SIZE_BYTES.equals(name))
+ {
+ return _queue.getCapacity();
+ }
+ else if(QUEUE_FLOW_RESUME_SIZE_BYTES.equals(name))
+ {
+ return _queue.getFlowResumeCapacity();
+ }
+ else if(QUEUE_FLOW_STOPPED.equals(name))
+ {
+ return _queue.isOverfull();
+ }
+ else if(SORT_KEY.equals(name))
+ {
+ if(_queue instanceof SortedQueue)
+ {
+ return ((SortedQueue)_queue).getSortedPropertyName();
+ }
+ }
+ else if(TYPE.equals(name))
+ {
+ if(_queue instanceof SortedQueue)
+ {
+ return "sorted";
+ }
+ if(_queue instanceof ConflationQueue)
+ {
+ return "lvq";
+ }
+ if(_queue instanceof AMQPriorityQueue)
+ {
+ return "priority";
+ }
+ return "standard";
+ }
+ else if(CREATED.equals(name))
+ {
+ // TODO
+ }
+ else if(DURABLE.equals(name))
+ {
+ return _queue.isDurable();
+ }
+ else if(ID.equals(name))
+ {
+ return getId();
+ }
+ else if(LIFETIME_POLICY.equals(name))
+ {
+ return _queue.isAutoDelete() ? LifetimePolicy.AUTO_DELETE : LifetimePolicy.PERMANENT;
+ }
+ else if(NAME.equals(name))
+ {
+ return _queue.getName();
+ }
+ else if(STATE.equals(name))
+ {
+ return State.ACTIVE; // TODO
+ }
+ else if(TIME_TO_LIVE.equals(name))
+ {
+ // TODO
+ }
+ else if(UPDATED.equals(name))
+ {
+ // TODO
+ }
+ else if (DESCRIPTION.equals(name))
+ {
+ return _queue.getDescription();
+ }
+
+ return super.getAttribute(name);
+ }
+
+ public Statistics getStatistics()
+ {
+ return _statistics;
+ }
+
+ @Override
+ public <C extends ConfiguredObject> Collection<C> getChildren(Class<C> clazz)
+ {
+ if(clazz == Consumer.class)
+ {
+ return (Collection<C>) getConsumers();
+ }
+ else if(clazz == org.apache.qpid.server.model.Binding.class)
+ {
+ return (Collection<C>) getBindings();
+ }
+ else
+ {
+ return Collections.emptySet();
+ }
+ }
+
+ public org.apache.qpid.server.model.Binding createBinding(Exchange exchange, Map<String, Object> attributes)
+ throws AccessControlException, IllegalStateException
+ {
+ attributes = new HashMap<String, Object>(attributes);
+ String bindingKey = getStringAttribute(org.apache.qpid.server.model.Binding.NAME, attributes, "");
+ Map<String, Object> bindingArgs = getMapAttribute(org.apache.qpid.server.model.Binding.ARGUMENTS, attributes, Collections.EMPTY_MAP);
+
+ attributes.remove(org.apache.qpid.server.model.Binding.NAME);
+ attributes.remove(org.apache.qpid.server.model.Binding.ARGUMENTS);
+
+ return exchange.createBinding(bindingKey, this, bindingArgs, attributes);
+
+ }
+
+
+
+ @Override
+ public <C extends ConfiguredObject> C createChild(Class<C> childClass, Map<String, Object> attributes, ConfiguredObject... otherParents)
+ {
+ if(childClass == org.apache.qpid.server.model.Binding.class)
+ {
+ if(otherParents != null && otherParents.length == 1 && otherParents[0] instanceof Exchange)
+ {
+ Exchange exchange = (Exchange) otherParents[0];
+ if(exchange.getParent(org.apache.qpid.server.model.VirtualHost.class) == getParent(org.apache.qpid.server.model.VirtualHost.class))
+ {
+ return (C) createBinding(exchange, attributes);
+ }
+ else
+ {
+ throw new IllegalArgumentException("Queue and Exchange parents of a binding must be on same virtual host");
+ }
+ }
+ else
+ {
+ throw new IllegalArgumentException("Other parent must be an exchange");
+ }
+ }
+ else
+ {
+ throw new IllegalArgumentException();
+ }
+ }
+
+ void bindingRegistered(Binding binding, BindingAdapter adapter)
+ {
+ synchronized (_bindingAdapters)
+ {
+ _bindingAdapters.put(binding, adapter);
+ }
+ childAdded(adapter);
+ }
+
+ void bindingUnregistered(Binding binding)
+ {
+ BindingAdapter adapter = null;
+ synchronized (_bindingAdapters)
+ {
+ adapter = _bindingAdapters.remove(binding);
+ }
+ if(adapter != null)
+ {
+ childRemoved(adapter);
+ }
+ }
+
+ AMQQueue getAMQQueue()
+ {
+ return _queue;
+ }
+
+ public void subscriptionRegistered(final AMQQueue queue, final Subscription subscription)
+ {
+ ConsumerAdapter adapter = null;
+ synchronized (_consumerAdapters)
+ {
+ if(!_consumerAdapters.containsKey(subscription))
+ {
+ adapter = new ConsumerAdapter(this, subscription);
+ _consumerAdapters.put(subscription,adapter);
+ // TODO - register with session
+ }
+ }
+ if(adapter != null)
+ {
+ childAdded(adapter);
+ }
+ }
+
+ public void subscriptionUnregistered(final AMQQueue queue, final Subscription subscription)
+ {
+ ConsumerAdapter adapter = null;
+
+ synchronized (_consumerAdapters)
+ {
+ adapter = _consumerAdapters.remove(subscription);
+ // TODO - register with session
+ }
+ if(adapter != null)
+ {
+ childRemoved(adapter);
+ }
+ }
+
+ VirtualHostAdapter getVirtualHost()
+ {
+ return _vhost;
+ }
+
+
+ private static class QueueStatisticsAdapter implements Statistics
+ {
+
+ private final AMQQueue _queue;
+
+ public QueueStatisticsAdapter(AMQQueue queue)
+ {
+ _queue = queue;
+ }
+
+ public Collection<String> getStatisticNames()
+ {
+ return Queue.AVAILABLE_STATISTICS;
+ }
+
+ public Object getStatistic(String name)
+ {
+ if(BINDING_COUNT.equals(name))
+ {
+ return _queue.getBindingCount();
+ }
+ else if(CONSUMER_COUNT.equals(name))
+ {
+ return _queue.getConsumerCount();
+ }
+ else if(CONSUMER_COUNT_WITH_CREDIT.equals(name))
+ {
+ return _queue.getActiveConsumerCount();
+ }
+ else if(DISCARDS_TTL_BYTES.equals(name))
+ {
+ return null; // TODO
+ }
+ else if(DISCARDS_TTL_MESSAGES.equals(name))
+ {
+ return null; // TODO
+ }
+ else if(PERSISTENT_DEQUEUED_BYTES.equals(name))
+ {
+ return _queue.getPersistentByteDequeues();
+ }
+ else if(PERSISTENT_DEQUEUED_MESSAGES.equals(name))
+ {
+ return _queue.getPersistentMsgDequeues();
+ }
+ else if(PERSISTENT_ENQUEUED_BYTES.equals(name))
+ {
+ return _queue.getPersistentByteEnqueues();
+ }
+ else if(PERSISTENT_ENQUEUED_MESSAGES.equals(name))
+ {
+ return _queue.getPersistentMsgEnqueues();
+ }
+ else if(QUEUE_DEPTH_BYTES.equals(name))
+ {
+ return _queue.getQueueDepth();
+ }
+ else if(QUEUE_DEPTH_MESSAGES.equals(name))
+ {
+ return _queue.getMessageCount();
+ }
+ else if(STATE_CHANGED.equals(name))
+ {
+ return null; // TODO
+ }
+ else if(TOTAL_DEQUEUED_BYTES.equals(name))
+ {
+ return _queue.getTotalDequeueSize();
+ }
+ else if(TOTAL_DEQUEUED_MESSAGES.equals(name))
+ {
+ return _queue.getTotalDequeueCount();
+ }
+ else if(TOTAL_ENQUEUED_BYTES.equals(name))
+ {
+ return _queue.getTotalEnqueueSize();
+ }
+ else if(TOTAL_ENQUEUED_MESSAGES.equals(name))
+ {
+ return _queue.getTotalEnqueueCount();
+ }
+ else if(UNACKNOWLEDGED_BYTES.equals(name))
+ {
+ return _queue.getUnackedMessageBytes();
+ }
+ else if(UNACKNOWLEDGED_MESSAGES.equals(name))
+ {
+ return _queue.getUnackedMessageCount();
+ }
+
+ return null;
+ }
+ }
+
+ @Override
+ public void setNotificationListener(QueueNotificationListener listener)
+ {
+ _queueNotificationListener = listener;
+ }
+
+ @Override
+ public void notifyClients(NotificationCheck notification, AMQQueue queue, String notificationMsg)
+ {
+ QueueNotificationListener listener = _queueNotificationListener;
+ if(listener != null)
+ {
+ listener.notifyClients(notification, this, notificationMsg);
+ }
+ }
+}
Added: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/SessionAdapter.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/SessionAdapter.java?rev=1355072&view=auto
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/SessionAdapter.java (added)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/SessionAdapter.java Thu Jun 28 16:46:12 2012
@@ -0,0 +1,239 @@
+/*
+ *
+ * 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.model.adapter;
+
+import java.security.AccessControlException;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+
+import java.util.Map;
+import org.apache.qpid.server.model.ConfiguredObject;
+import org.apache.qpid.server.model.LifetimePolicy;
+import org.apache.qpid.server.model.Publisher;
+import org.apache.qpid.server.model.Session;
+import org.apache.qpid.server.model.State;
+import org.apache.qpid.server.model.Statistics;
+import org.apache.qpid.server.model.Consumer;
+import org.apache.qpid.server.protocol.AMQSessionModel;
+
+final class SessionAdapter extends AbstractAdapter implements Session
+{
+ // Attributes
+
+
+ private AMQSessionModel _session;
+ private SessionStatistics _statistics;
+
+ public SessionAdapter(final AMQSessionModel session)
+ {
+ _session = session;
+ _statistics = new SessionStatistics();
+ }
+
+ public Collection<Consumer> getSubscriptions()
+ {
+ return null; //TODO
+ }
+
+ public Collection<Publisher> getPublishers()
+ {
+ return null; //TODO
+ }
+
+ public String getName()
+ {
+ return String.valueOf(_session.getChannelId());
+ }
+
+ public String setName(final String currentName, final String desiredName)
+ throws IllegalStateException, AccessControlException
+ {
+ return null; //TODO
+ }
+
+ public State getActualState()
+ {
+ return null; //TODO
+ }
+
+ public boolean isDurable()
+ {
+ return false; //TODO
+ }
+
+ public void setDurable(final boolean durable)
+ throws IllegalStateException, AccessControlException, IllegalArgumentException
+ {
+ //TODO
+ }
+
+ public LifetimePolicy getLifetimePolicy()
+ {
+ return null; //TODO
+ }
+
+ public LifetimePolicy setLifetimePolicy(final LifetimePolicy expected, final LifetimePolicy desired)
+ throws IllegalStateException, AccessControlException, IllegalArgumentException
+ {
+ return null; //TODO
+ }
+
+ public long getTimeToLive()
+ {
+ return 0; //TODO
+ }
+
+ public long setTimeToLive(final long expected, final long desired)
+ throws IllegalStateException, AccessControlException, IllegalArgumentException
+ {
+ return 0; //TODO
+ }
+
+ @Override
+ public Collection<String> getAttributeNames()
+ {
+ Collection<String> names = new HashSet<String>(super.getAttributeNames());
+ names.addAll(AVAILABLE_ATTRIBUTES);
+
+ return Collections.unmodifiableCollection(names);
+ }
+
+ @Override
+ public Object getAttribute(String name)
+ {
+ if(name.equals(ID))
+ {
+ return getId();
+ }
+ else if (name.equals(NAME))
+ {
+ return getName();
+ }
+ else if(name.equals(CHANNEL_ID))
+ {
+ return _session.getChannelId();
+ }
+ else if(name.equals(PRODUCER_FLOW_BLOCKED))
+ {
+ return _session.getBlocking();
+ }
+ return super.getAttribute(name); //TODO - Implement
+ }
+
+ @Override
+ public Object setAttribute(String name, Object expected, Object desired)
+ throws IllegalStateException, AccessControlException, IllegalArgumentException
+ {
+ return super.setAttribute(name, expected, desired); //TODO - Implement
+ }
+
+ public Statistics getStatistics()
+ {
+ return _statistics;
+ }
+
+ @Override
+ public <C extends ConfiguredObject> Collection<C> getChildren(Class<C> clazz)
+ {
+ if(clazz == Consumer.class)
+ {
+ return (Collection<C>) getSubscriptions();
+ }
+ else if(clazz == Publisher.class)
+ {
+ return (Collection<C>) getPublishers();
+ }
+ else
+ {
+ return Collections.emptySet();
+ }
+ }
+
+ @Override
+ public <C extends ConfiguredObject> C createChild(Class<C> childClass, Map<String, Object> attributes, ConfiguredObject... otherParents)
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ private class SessionStatistics implements Statistics
+ {
+
+ public SessionStatistics()
+ {
+ }
+
+ public Collection<String> getStatisticNames()
+ {
+ return AVAILABLE_STATISTICS;
+ }
+
+ public Object getStatistic(String name)
+ {
+ if(name.equals(BYTES_IN))
+ {
+ }
+ else if(name.equals(BYTES_OUT))
+ {
+ }
+ else if(name.equals(CONSUMER_COUNT))
+ {
+ final Collection<Consumer> subscriptions = getSubscriptions();
+ return subscriptions == null ? 0 : subscriptions.size();
+ }
+ else if(name.equals(LOCAL_TRANSACTION_BEGINS))
+ {
+ return _session.getTxnStart();
+ }
+ else if(name.equals(LOCAL_TRANSACTION_OPEN))
+ {
+ long open = _session.getTxnCount() - (_session.getTxnCommits() + _session.getTxnRejects());
+ return (Boolean) (open > 0l);
+ }
+ else if(name.equals(LOCAL_TRANSACTION_ROLLBACKS))
+ {
+ return _session.getTxnCommits();
+ }
+ else if(name.equals(STATE_CHANGED))
+ {
+ }
+ else if(name.equals(UNACKNOWLEDGED_BYTES))
+ {
+ }
+ else if(name.equals(UNACKNOWLEDGED_MESSAGES))
+ {
+ return _session.getUnacknowledgedMessageCount();
+ }
+ else if(name.equals(XA_TRANSACTION_BRANCH_ENDS))
+ {
+ }
+ else if(name.equals(XA_TRANSACTION_BRANCH_STARTS))
+ {
+ }
+ else if(name.equals(XA_TRANSACTION_BRANCH_SUSPENDS))
+ {
+
+ }
+
+ return null; // TODO - Implement
+ }
+ }
+}
Added: qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/StatisticsAdapter.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/StatisticsAdapter.java?rev=1355072&view=auto
==============================================================================
--- qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/StatisticsAdapter.java (added)
+++ qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/model/adapter/StatisticsAdapter.java Thu Jun 28 16:46:12 2012
@@ -0,0 +1,67 @@
+package org.apache.qpid.server.model.adapter;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.qpid.server.model.Statistics;
+import org.apache.qpid.server.stats.StatisticsCounter;
+import org.apache.qpid.server.stats.StatisticsGatherer;
+
+/**
+* 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
+* <p/>
+* http://www.apache.org/licenses/LICENSE-2.0
+* <p/>
+* 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.
+*/
+class StatisticsAdapter implements Statistics
+{
+
+ private final Map<String, StatisticsCounter> _statistics =
+ new HashMap<String, StatisticsCounter>();
+
+
+ private static final String BYTES_IN = "bytesIn";
+ private static final String BYTES_OUT = "bytesOut";
+ private static final String MESSAGES_IN = "messagesIn";
+ private static final String MESSAGES_OUT = "messagesOut";
+
+ private static final Collection<String> STATISTIC_NAMES =
+ Collections.unmodifiableCollection(Arrays.asList(BYTES_IN, BYTES_OUT, MESSAGES_IN, MESSAGES_OUT));
+
+
+
+ public StatisticsAdapter(StatisticsGatherer statGatherer)
+ {
+ _statistics.put(BYTES_OUT, statGatherer.getDataDeliveryStatistics());
+ _statistics.put(BYTES_IN, statGatherer.getDataReceiptStatistics());
+ _statistics.put(MESSAGES_OUT, statGatherer.getMessageDeliveryStatistics());
+ _statistics.put(MESSAGES_IN, statGatherer.getMessageReceiptStatistics());
+ }
+
+
+ public Collection<String> getStatisticNames()
+ {
+ return STATISTIC_NAMES;
+ }
+
+ public Object getStatistic(String name)
+ {
+ StatisticsCounter counter = _statistics.get(name);
+ return counter == null ? null : counter.getTotal();
+
+ }
+
+
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org