You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rg...@apache.org on 2017/10/12 14:08:15 UTC
qpid-broker-j git commit: QPID-7965: Centralise setting of connection
props
Repository: qpid-broker-j
Updated Branches:
refs/heads/master fc952cea2 -> fce924717
QPID-7965: Centralise setting of connection props
Project: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/commit/fce92471
Tree: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/tree/fce92471
Diff: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/diff/fce92471
Branch: refs/heads/master
Commit: fce924717643445931a836bf126917095c70b639
Parents: fc952ce
Author: rgodfrey <rg...@apache.org>
Authored: Thu Oct 12 16:02:39 2017 +0200
Committer: rgodfrey <rg...@apache.org>
Committed: Thu Oct 12 16:06:37 2017 +0200
----------------------------------------------------------------------
.../StandardConnectionPropertyEnricher.java | 115 +++++++++++++++++++
.../SuppliedMapConnectionPropertyEnricher.java | 67 +++++++++++
.../apache/qpid/server/model/port/AmqpPort.java | 10 ++
.../qpid/server/model/port/AmqpPortImpl.java | 30 ++++-
.../plugin/ConnectionPropertyEnricher.java | 27 +++++
.../v0_10/ServerConnectionDelegate.java | 46 ++------
.../protocol/v0_8/AMQPConnection_0_8Impl.java | 34 ++----
.../protocol/v1_0/AMQPConnection_1_0Impl.java | 17 ++-
8 files changed, 277 insertions(+), 69 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/fce92471/broker-core/src/main/java/org/apache/qpid/server/connection/StandardConnectionPropertyEnricher.java
----------------------------------------------------------------------
diff --git a/broker-core/src/main/java/org/apache/qpid/server/connection/StandardConnectionPropertyEnricher.java b/broker-core/src/main/java/org/apache/qpid/server/connection/StandardConnectionPropertyEnricher.java
new file mode 100644
index 0000000..d7151df
--- /dev/null
+++ b/broker-core/src/main/java/org/apache/qpid/server/connection/StandardConnectionPropertyEnricher.java
@@ -0,0 +1,115 @@
+/**
+ * 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.
+ */
+
+package org.apache.qpid.server.connection;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.qpid.server.common.ServerPropertyNames;
+import org.apache.qpid.server.configuration.CommonProperties;
+import org.apache.qpid.server.model.Broker;
+import org.apache.qpid.server.plugin.ConnectionPropertyEnricher;
+import org.apache.qpid.server.plugin.PluggableService;
+import org.apache.qpid.server.properties.ConnectionStartProperties;
+import org.apache.qpid.server.transport.AMQPConnection;
+
+@PluggableService
+public class StandardConnectionPropertyEnricher implements ConnectionPropertyEnricher
+{
+ private static final Logger LOG = LoggerFactory.getLogger(StandardConnectionPropertyEnricher.class);
+
+ @Override
+ public Map<String, Object> addConnectionProperties(final AMQPConnection<?> connection,
+ final Map<String, Object> existingProperties)
+ {
+ Map<String,Object> modifiedProperties = new LinkedHashMap<>(existingProperties);
+
+ Broker<?> broker = connection.getBroker();
+
+ modifiedProperties.put(ServerPropertyNames.PRODUCT, CommonProperties.getProductName());
+ modifiedProperties.put(ServerPropertyNames.VERSION, CommonProperties.getReleaseVersion());
+ modifiedProperties.put(ServerPropertyNames.QPID_BUILD, CommonProperties.getBuildVersion());
+ modifiedProperties.put(ServerPropertyNames.QPID_INSTANCE_NAME, broker.getName());
+ modifiedProperties.put(ConnectionStartProperties.QPID_VIRTUALHOST_PROPERTIES_SUPPORTED,
+ String.valueOf(broker.isVirtualHostPropertiesNodeEnabled()));
+
+ modifiedProperties.put(ConnectionStartProperties.QPID_MESSAGE_COMPRESSION_SUPPORTED,
+ String.valueOf(broker.isMessageCompressionEnabled()));
+ modifiedProperties.put(ConnectionStartProperties.QPID_QUEUE_LIFETIME_SUPPORTED, Boolean.TRUE.toString());
+
+
+ switch (connection.getProtocol())
+ {
+ case AMQP_0_8:
+ case AMQP_0_9:
+ case AMQP_0_9_1:
+ modifiedProperties.put(ConnectionStartProperties.QPID_CONFIRMED_PUBLISH_SUPPORTED,
+ Boolean.TRUE.toString());
+ modifiedProperties.put(ConnectionStartProperties.QPID_CLOSE_WHEN_NO_ROUTE,
+ String.valueOf(connection.getPort().getCloseWhenNoRoute()));
+
+ break;
+ case AMQP_0_10:
+ // Federation tag is used by the client to identify the broker instance
+ modifiedProperties.put(ServerPropertyNames.FEDERATION_TAG, broker.getId().toString());
+ final List<String> features = getFeatures(broker);
+ if (features.size() > 0)
+ {
+ modifiedProperties.put(ServerPropertyNames.QPID_FEATURES, features);
+ }
+ break;
+ case AMQP_1_0:
+ // message compression is not supported in 1.0
+ modifiedProperties.remove(ConnectionStartProperties.QPID_MESSAGE_COMPRESSION_SUPPORTED);
+ // this property is only meaningful for queue declare operations in 0-x
+ modifiedProperties.remove(ConnectionStartProperties.QPID_QUEUE_LIFETIME_SUPPORTED);
+
+ break;
+ default:
+ LOG.info("Unexpected protocol: " + connection.getProtocol());
+ break;
+ }
+
+ return Collections.unmodifiableMap(modifiedProperties);
+ }
+
+ private static List<String> getFeatures(Broker<?> broker)
+ {
+ String brokerDisabledFeatures = System.getProperty(Broker.PROPERTY_DISABLED_FEATURES);
+ final List<String> features = new ArrayList<String>();
+ if (brokerDisabledFeatures == null || !brokerDisabledFeatures.contains(ServerPropertyNames.FEATURE_QPID_JMS_SELECTOR))
+ {
+ features.add(ServerPropertyNames.FEATURE_QPID_JMS_SELECTOR);
+ }
+
+ return Collections.unmodifiableList(features);
+ }
+
+
+ @Override
+ public String getType()
+ {
+ return "STANDARD";
+ }
+}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/fce92471/broker-core/src/main/java/org/apache/qpid/server/connection/SuppliedMapConnectionPropertyEnricher.java
----------------------------------------------------------------------
diff --git a/broker-core/src/main/java/org/apache/qpid/server/connection/SuppliedMapConnectionPropertyEnricher.java b/broker-core/src/main/java/org/apache/qpid/server/connection/SuppliedMapConnectionPropertyEnricher.java
new file mode 100644
index 0000000..a26d868
--- /dev/null
+++ b/broker-core/src/main/java/org/apache/qpid/server/connection/SuppliedMapConnectionPropertyEnricher.java
@@ -0,0 +1,67 @@
+/**
+ * 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.
+ */
+
+package org.apache.qpid.server.connection;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.qpid.server.plugin.ConnectionPropertyEnricher;
+import org.apache.qpid.server.plugin.PluggableService;
+import org.apache.qpid.server.transport.AMQPConnection;
+
+@PluggableService
+public class SuppliedMapConnectionPropertyEnricher implements ConnectionPropertyEnricher
+{
+ private static final Logger LOG = LoggerFactory.getLogger(SuppliedMapConnectionPropertyEnricher.class);
+
+ private static final String PROPERTY_MAP_CONTEXT = "qpid.connection.property_map";
+
+ @Override
+ public Map<String, Object> addConnectionProperties(final AMQPConnection<?> connection,
+ final Map<String, Object> existingProperties)
+ {
+
+ if(connection.getContextKeys(false).contains(PROPERTY_MAP_CONTEXT))
+ {
+ Map<String,Object> modifiedProperties = new LinkedHashMap<>(existingProperties);
+
+ // This will be OK as the context will come from JSON which necessitates a string-keyed map
+ @SuppressWarnings("unchecked")
+ Map<String, Object> map = connection.getContextValue(Map.class, PROPERTY_MAP_CONTEXT);
+
+ modifiedProperties.putAll(map);
+ return Collections.unmodifiableMap(modifiedProperties);
+ }
+ else
+ {
+ return existingProperties;
+ }
+
+
+ }
+
+ @Override
+ public String getType()
+ {
+ return "SUPPLIED-MAP";
+ }
+}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/fce92471/broker-core/src/main/java/org/apache/qpid/server/model/port/AmqpPort.java
----------------------------------------------------------------------
diff --git a/broker-core/src/main/java/org/apache/qpid/server/model/port/AmqpPort.java b/broker-core/src/main/java/org/apache/qpid/server/model/port/AmqpPort.java
index 331271b..d1d4e01 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/model/port/AmqpPort.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/model/port/AmqpPort.java
@@ -21,6 +21,7 @@
package org.apache.qpid.server.model.port;
import java.net.SocketAddress;
+import java.util.List;
import java.util.Set;
import javax.net.ssl.SSLContext;
@@ -35,6 +36,7 @@ import org.apache.qpid.server.model.Protocol;
import org.apache.qpid.server.model.StatisticType;
import org.apache.qpid.server.model.StatisticUnit;
import org.apache.qpid.server.model.Transport;
+import org.apache.qpid.server.plugin.ConnectionPropertyEnricher;
@ManagedObject( category = false, type = "AMQP", amqpName = "org.apache.qpid.AmqpPort")
public interface AmqpPort<X extends AmqpPort<X>> extends Port<X>
@@ -124,6 +126,12 @@ public interface AmqpPort<X extends AmqpPort<X>> extends Port<X>
@ManagedContextDefault(name = TLS_SESSION_CACHE_SIZE, description = "TLS session cache size for AMQP ports (seconds).")
int DEFAULT_TLS_SESSION_CACHE_SIZE = 100;
+ String CONNECTION_PROPERTY_ENRICHERS = "qpid.port.connection_property_enrichers";
+ @SuppressWarnings("unused")
+ @ManagedContextDefault(name = CONNECTION_PROPERTY_ENRICHERS,
+ description = "The connection property enrichers to apply to connections created on this port.")
+ String DEFAULT_CONNECTION_PROTOCOL_ENRICHERS = "[ \"STANDARD\" ] ";
+
SSLContext getSSLContext();
@ManagedAttribute(defaultValue = "*")
@@ -193,4 +201,6 @@ public interface AmqpPort<X extends AmqpPort<X>> extends Port<X>
int decrementConnectionCount();
int getNetworkBufferSize();
+
+ List<ConnectionPropertyEnricher> getConnectionPropertyEnrichers();
}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/fce92471/broker-core/src/main/java/org/apache/qpid/server/model/port/AmqpPortImpl.java
----------------------------------------------------------------------
diff --git a/broker-core/src/main/java/org/apache/qpid/server/model/port/AmqpPortImpl.java b/broker-core/src/main/java/org/apache/qpid/server/model/port/AmqpPortImpl.java
index 4faebe6..dc66d6e 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/model/port/AmqpPortImpl.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/model/port/AmqpPortImpl.java
@@ -23,10 +23,12 @@ import java.io.IOException;
import java.io.StringWriter;
import java.net.SocketAddress;
import java.security.PrivilegedAction;
+import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -60,6 +62,7 @@ import org.apache.qpid.server.model.Transport;
import org.apache.qpid.server.model.TrustStore;
import org.apache.qpid.server.model.VirtualHostAlias;
import org.apache.qpid.server.model.VirtualHostNameAlias;
+import org.apache.qpid.server.plugin.ConnectionPropertyEnricher;
import org.apache.qpid.server.plugin.ProtocolEngineCreator;
import org.apache.qpid.server.plugin.QpidServiceLoader;
import org.apache.qpid.server.plugin.TransportProviderFactory;
@@ -109,7 +112,7 @@ public class AmqpPortImpl extends AbstractPort<AmqpPortImpl> implements AmqpPort
private volatile int _heartBeatDelay;
private volatile int _tlsSessionTimeout;
private volatile int _tlsSessionCacheSize;
-
+ private volatile List<ConnectionPropertyEnricher> _connectionPropertyEnrichers;
@ManagedObjectFactoryConstructor
public AmqpPortImpl(Map<String, Object> attributes, Container<?> container)
@@ -200,6 +203,25 @@ public class AmqpPortImpl extends AbstractPort<AmqpPortImpl> implements AmqpPort
_heartBeatDelay = getContextValue(Integer.class, AmqpPort.HEART_BEAT_DELAY);
_tlsSessionTimeout = getContextValue(Integer.class, AmqpPort.TLS_SESSION_TIMEOUT);
_tlsSessionCacheSize = getContextValue(Integer.class, AmqpPort.TLS_SESSION_CACHE_SIZE);
+
+ @SuppressWarnings("unchecked")
+ List<String> configurationPropertyEnrichers = getContextValue(List.class, AmqpPort.CONNECTION_PROPERTY_ENRICHERS);
+ List<ConnectionPropertyEnricher> enrichers = new ArrayList<>(configurationPropertyEnrichers.size());
+ final Map<String, ConnectionPropertyEnricher> enrichersByType =
+ new QpidServiceLoader().getInstancesByType(ConnectionPropertyEnricher.class);
+ for(String enricherName : configurationPropertyEnrichers)
+ {
+ ConnectionPropertyEnricher enricher = enrichersByType.get(enricherName);
+ if(enricher != null)
+ {
+ enrichers.add(enricher);
+ }
+ else
+ {
+ LOGGER.warn("Ignoring unknown Connection Property Enricher type: '"+enricherName+"' on port " + this.getName());
+ }
+ }
+ _connectionPropertyEnrichers = Collections.unmodifiableList(enrichers);
}
@Override
@@ -291,6 +313,12 @@ public class AmqpPortImpl extends AbstractPort<AmqpPortImpl> implements AmqpPort
}
@Override
+ public List<ConnectionPropertyEnricher> getConnectionPropertyEnrichers()
+ {
+ return _connectionPropertyEnrichers;
+ }
+
+ @Override
public int getBoundPort()
{
return _boundPort;
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/fce92471/broker-core/src/main/java/org/apache/qpid/server/plugin/ConnectionPropertyEnricher.java
----------------------------------------------------------------------
diff --git a/broker-core/src/main/java/org/apache/qpid/server/plugin/ConnectionPropertyEnricher.java b/broker-core/src/main/java/org/apache/qpid/server/plugin/ConnectionPropertyEnricher.java
new file mode 100644
index 0000000..e2d3859
--- /dev/null
+++ b/broker-core/src/main/java/org/apache/qpid/server/plugin/ConnectionPropertyEnricher.java
@@ -0,0 +1,27 @@
+/**
+ * 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.
+ */
+
+package org.apache.qpid.server.plugin;
+
+import java.util.Map;
+
+import org.apache.qpid.server.transport.AMQPConnection;
+
+public interface ConnectionPropertyEnricher extends Pluggable
+{
+ Map<String,Object> addConnectionProperties(AMQPConnection<?> connection, Map<String,Object> existingProperties);
+}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/fce92471/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnectionDelegate.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnectionDelegate.java b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnectionDelegate.java
index 129e393..cab769f 100644
--- a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnectionDelegate.java
+++ b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnectionDelegate.java
@@ -27,7 +27,6 @@ import java.security.Principal;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
-import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
@@ -35,11 +34,10 @@ import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.apache.qpid.server.common.ServerPropertyNames;
-import org.apache.qpid.server.configuration.CommonProperties;
import org.apache.qpid.server.model.Broker;
import org.apache.qpid.server.model.NamedAddressSpace;
import org.apache.qpid.server.model.port.AmqpPort;
+import org.apache.qpid.server.plugin.ConnectionPropertyEnricher;
import org.apache.qpid.server.properties.ConnectionStartProperties;
import org.apache.qpid.server.protocol.v0_10.transport.*;
import org.apache.qpid.server.security.SubjectCreator;
@@ -59,7 +57,7 @@ public class ServerConnectionDelegate extends MethodDelegate<ServerConnection> i
private List<Object> _locales;
private List<Object> _mechanisms;
- private final Broker _broker;
+ private final Broker<?> _broker;
private int _maxNoOfChannels;
private Map<String,Object> _clientProperties;
private final SubjectCreator _subjectCreator;
@@ -86,7 +84,6 @@ public class ServerConnectionDelegate extends MethodDelegate<ServerConnection> i
{
_port = port;
_broker = (Broker<?>) port.getParent();
- _clientProperties = createConnectionProperties((Broker<?>) port.getParent());
_mechanisms = new ArrayList<>(port.getAuthenticationProvider().getAvailableMechanisms(secure));
_maxNoOfChannels = port.getSessionCountLimit();
@@ -166,42 +163,13 @@ public class ServerConnectionDelegate extends MethodDelegate<ServerConnection> i
{
assertState(serverConnection, ConnectionState.INIT);
serverConnection.send(new ProtocolHeader(1, 0, 10));
- serverConnection.sendConnectionStart(_clientProperties, _mechanisms, Collections.singletonList((Object)"en_US"));
- _state = ConnectionState.AWAIT_START_OK;
- }
-
- private static List<String> getFeatures(Broker<?> broker)
- {
- String brokerDisabledFeatures = System.getProperty(Broker.PROPERTY_DISABLED_FEATURES);
- final List<String> features = new ArrayList<String>();
- if (brokerDisabledFeatures == null || !brokerDisabledFeatures.contains(ServerPropertyNames.FEATURE_QPID_JMS_SELECTOR))
+ Map<String,Object> props = Collections.emptyMap();
+ for(ConnectionPropertyEnricher enricher : _port.getConnectionPropertyEnrichers())
{
- features.add(ServerPropertyNames.FEATURE_QPID_JMS_SELECTOR);
+ props = enricher.addConnectionProperties(serverConnection.getAmqpConnection(), props);
}
-
- return Collections.unmodifiableList(features);
- }
-
- private static Map<String, Object> createConnectionProperties(final Broker<?> broker)
- {
- final Map<String,Object> map = new HashMap<String,Object>();
- // Federation tag is used by the client to identify the broker instance
- map.put(ServerPropertyNames.FEDERATION_TAG, broker.getId().toString());
- final List<String> features = getFeatures(broker);
- if (features != null && features.size() > 0)
- {
- map.put(ServerPropertyNames.QPID_FEATURES, features);
- }
-
- map.put(ServerPropertyNames.PRODUCT, CommonProperties.getProductName());
- map.put(ServerPropertyNames.VERSION, CommonProperties.getReleaseVersion());
- map.put(ServerPropertyNames.QPID_BUILD, CommonProperties.getBuildVersion());
- map.put(ServerPropertyNames.QPID_INSTANCE_NAME, broker.getName());
- map.put(ConnectionStartProperties.QPID_MESSAGE_COMPRESSION_SUPPORTED, String.valueOf(broker.isMessageCompressionEnabled()));
- map.put(ConnectionStartProperties.QPID_VIRTUALHOST_PROPERTIES_SUPPORTED, String.valueOf(broker.isVirtualHostPropertiesNodeEnabled()));
- map.put(ConnectionStartProperties.QPID_QUEUE_LIFETIME_SUPPORTED, Boolean.TRUE.toString());
-
- return map;
+ serverConnection.sendConnectionStart(props, _mechanisms, Collections.singletonList((Object)"en_US"));
+ _state = ConnectionState.AWAIT_START_OK;
}
@Override
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/fce92471/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQPConnection_0_8Impl.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQPConnection_0_8Impl.java b/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQPConnection_0_8Impl.java
index b7f2bb5..d49e115 100644
--- a/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQPConnection_0_8Impl.java
+++ b/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQPConnection_0_8Impl.java
@@ -48,10 +48,6 @@ import org.slf4j.LoggerFactory;
import org.apache.qpid.server.QpidException;
import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
-import org.apache.qpid.server.common.ServerPropertyNames;
-import org.apache.qpid.server.configuration.CommonProperties;
-import org.apache.qpid.server.properties.ConnectionStartProperties;
-import org.apache.qpid.server.protocol.ErrorCodes;
import org.apache.qpid.server.logging.messages.ConnectionMessages;
import org.apache.qpid.server.message.InstanceProperties;
import org.apache.qpid.server.model.Broker;
@@ -60,7 +56,10 @@ import org.apache.qpid.server.model.NamedAddressSpace;
import org.apache.qpid.server.model.Protocol;
import org.apache.qpid.server.model.Transport;
import org.apache.qpid.server.model.port.AmqpPort;
+import org.apache.qpid.server.plugin.ConnectionPropertyEnricher;
+import org.apache.qpid.server.properties.ConnectionStartProperties;
import org.apache.qpid.server.protocol.ConnectionClosingTicker;
+import org.apache.qpid.server.protocol.ErrorCodes;
import org.apache.qpid.server.protocol.ProtocolVersion;
import org.apache.qpid.server.protocol.v0_8.transport.*;
import org.apache.qpid.server.security.SubjectCreator;
@@ -70,15 +69,15 @@ import org.apache.qpid.server.session.AMQPSession;
import org.apache.qpid.server.store.StoreException;
import org.apache.qpid.server.transport.AbstractAMQPConnection;
import org.apache.qpid.server.transport.AggregateTicker;
+import org.apache.qpid.server.transport.ByteBufferSender;
import org.apache.qpid.server.transport.ProtocolEngine;
import org.apache.qpid.server.transport.ServerNetworkConnection;
+import org.apache.qpid.server.transport.TransportException;
import org.apache.qpid.server.util.Action;
import org.apache.qpid.server.util.ConnectionScopedRuntimeException;
import org.apache.qpid.server.util.ServerScopedRuntimeException;
import org.apache.qpid.server.virtualhost.NoopConnectionEstablishmentPolicy;
import org.apache.qpid.server.virtualhost.VirtualHostUnavailableException;
-import org.apache.qpid.server.transport.ByteBufferSender;
-import org.apache.qpid.server.transport.TransportException;
public class AMQPConnection_0_8Impl
extends AbstractAMQPConnection<AMQPConnection_0_8Impl, AMQPConnection_0_8Impl>
@@ -332,24 +331,13 @@ public class AMQPConnection_0_8Impl
String locales = "en_US";
+ Map<String,Object> props = Collections.emptyMap();
+ for(ConnectionPropertyEnricher enricher : getPort().getConnectionPropertyEnrichers())
+ {
+ props = enricher.addConnectionProperties(this, props);
+ }
- FieldTable serverProperties = FieldTableFactory.newFieldTable();
-
- serverProperties.setString(ServerPropertyNames.PRODUCT,
- CommonProperties.getProductName());
- serverProperties.setString(ServerPropertyNames.VERSION,
- CommonProperties.getReleaseVersion());
- serverProperties.setString(ServerPropertyNames.QPID_BUILD,
- CommonProperties.getBuildVersion());
- serverProperties.setString(ServerPropertyNames.QPID_INSTANCE_NAME,
- getBroker().getName());
- serverProperties.setString(ConnectionStartProperties.QPID_CLOSE_WHEN_NO_ROUTE,
- String.valueOf(_closeWhenNoRoute));
- serverProperties.setString(ConnectionStartProperties.QPID_MESSAGE_COMPRESSION_SUPPORTED,
- String.valueOf(getBroker().isMessageCompressionEnabled()));
- serverProperties.setString(ConnectionStartProperties.QPID_CONFIRMED_PUBLISH_SUPPORTED, Boolean.TRUE.toString());
- serverProperties.setString(ConnectionStartProperties.QPID_VIRTUALHOST_PROPERTIES_SUPPORTED, String.valueOf(getBroker().isVirtualHostPropertiesNodeEnabled()));
- serverProperties.setString(ConnectionStartProperties.QPID_QUEUE_LIFETIME_SUPPORTED, Boolean.TRUE.toString());
+ FieldTable serverProperties = FieldTable.convertToFieldTable(props);
AMQMethodBody responseBody = getMethodRegistry().createConnectionStartBody((short) getProtocolMajorVersion(),
(short) pv.getActualMinorVersion(),
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/fce92471/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AMQPConnection_1_0Impl.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AMQPConnection_1_0Impl.java b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AMQPConnection_1_0Impl.java
index e239c19..8419b0b 100644
--- a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AMQPConnection_1_0Impl.java
+++ b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/AMQPConnection_1_0Impl.java
@@ -59,8 +59,6 @@ import org.slf4j.LoggerFactory;
import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
import org.apache.qpid.server.bytebuffer.QpidByteBufferUtils;
-import org.apache.qpid.server.common.ServerPropertyNames;
-import org.apache.qpid.server.configuration.CommonProperties;
import org.apache.qpid.server.logging.messages.ConnectionMessages;
import org.apache.qpid.server.model.AuthenticationProvider;
import org.apache.qpid.server.model.Broker;
@@ -69,6 +67,7 @@ import org.apache.qpid.server.model.NamedAddressSpace;
import org.apache.qpid.server.model.Protocol;
import org.apache.qpid.server.model.Transport;
import org.apache.qpid.server.model.port.AmqpPort;
+import org.apache.qpid.server.plugin.ConnectionPropertyEnricher;
import org.apache.qpid.server.protocol.ConnectionClosingTicker;
import org.apache.qpid.server.protocol.v1_0.codec.DescribedTypeConstructorRegistry;
import org.apache.qpid.server.protocol.v1_0.codec.FrameWriter;
@@ -249,10 +248,6 @@ public class AMQPConnection_1_0Impl extends AbstractAMQPConnection<AMQPConnectio
super(broker, network, port, transport, Protocol.AMQP_1_0, id, aggregateTicker);
_subjectCreator = port.getSubjectCreator(transport.isSecure(), network.getSelectedHost());
- _properties.put(Symbol.valueOf(ServerPropertyNames.PRODUCT), CommonProperties.getProductName());
- _properties.put(Symbol.valueOf(ServerPropertyNames.VERSION), CommonProperties.getReleaseVersion());
- _properties.put(Symbol.valueOf(ServerPropertyNames.QPID_BUILD), CommonProperties.getBuildVersion());
- _properties.put(Symbol.valueOf(ServerPropertyNames.QPID_INSTANCE_NAME), broker.getName());
List<Symbol> offeredCapabilities = new ArrayList<>();
offeredCapabilities.add(ANONYMOUS_RELAY);
@@ -1693,6 +1688,16 @@ public class AMQPConnection_1_0Impl extends AbstractAMQPConnection<AMQPConnectio
{
Open open = new Open();
+ Map<String,Object> props = Collections.emptyMap();
+ for(ConnectionPropertyEnricher enricher : getPort().getConnectionPropertyEnrichers())
+ {
+ props = enricher.addConnectionProperties(this, props);
+ }
+ for(Map.Entry<String,Object> entry : props.entrySet())
+ {
+ _properties.put(Symbol.valueOf(entry.getKey()), entry.getValue());
+ }
+
if (_receivingSessions == null)
{
_receivingSessions = new Session_1_0[channelMax + 1];
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org