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 2015/03/01 22:33:37 UTC
svn commit: r1663170 - in /qpid/trunk/qpid/java:
bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/
broker-core/src/main/java/org/apache/qpid/server/filter/
broker-core/src/main/java/org/apache/qpid/server/model/
broker-core/src/main...
Author: rgodfrey
Date: Sun Mar 1 21:33:36 2015
New Revision: 1663170
URL: http://svn.apache.org/r1663170
Log:
QPID-6424 : Implement Connection.Redirect in 0-8/9/9-1
Added:
qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhostnode/RedirectingVirtualHost.java (with props)
qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhostnode/RedirectingVirtualHostImpl.java (with props)
qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhostnode/RedirectingVirtualHostNode.java (with props)
qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhostnode/RedirectingVirtualHostNodeImpl.java (with props)
qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/failover/ConnectionRedirectException.java (with props)
Modified:
qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBHAReplicaVirtualHostImpl.java
qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/filter/ArrivalTimeFilterFactory.java
qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/filter/JMSSelectorFilterFactory.java
qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/AttributeValueConverter.java
qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/Queue.java
qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/VirtualHost.java
qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/plugin/MessageFilterFactory.java
qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java
qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java
qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnectionDelegate.java
qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java
qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQBrokerDetails.java
qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java
qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java
qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/failover/FailoverHandler.java
qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionRedirectMethodHandler.java
qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java
qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java
qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/state/AMQStateManager.java
qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jms/BrokerDetails.java
qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/ClientDelegate.java
qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java
qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/ConnectionListener.java
qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/sasl/SASLEncryptor.java
qpid/trunk/qpid/java/common/src/test/java/org/apache/qpid/transport/ConnectionTest.java
Modified: qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBHAReplicaVirtualHostImpl.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBHAReplicaVirtualHostImpl.java?rev=1663170&r1=1663169&r2=1663170&view=diff
==============================================================================
--- qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBHAReplicaVirtualHostImpl.java (original)
+++ qpid/trunk/qpid/java/bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBHAReplicaVirtualHostImpl.java Sun Mar 1 21:33:36 2015
@@ -43,6 +43,7 @@ import org.apache.qpid.server.model.Mana
import org.apache.qpid.server.model.State;
import org.apache.qpid.server.model.VirtualHostAlias;
import org.apache.qpid.server.model.VirtualHostNode;
+import org.apache.qpid.server.model.port.AmqpPort;
import org.apache.qpid.server.protocol.AMQConnectionModel;
import org.apache.qpid.server.protocol.LinkRegistry;
import org.apache.qpid.server.queue.AMQQueue;
@@ -171,6 +172,12 @@ public class BDBHAReplicaVirtualHostImpl
}
@Override
+ public String getRedirectHost(final AmqpPort<?> port)
+ {
+ return null;
+ }
+
+ @Override
public boolean isQueue_deadLetterQueueEnabled()
{
return false;
Modified: qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/filter/ArrivalTimeFilterFactory.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/filter/ArrivalTimeFilterFactory.java?rev=1663170&r1=1663169&r2=1663170&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/filter/ArrivalTimeFilterFactory.java (original)
+++ qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/filter/ArrivalTimeFilterFactory.java Sun Mar 1 21:33:36 2015
@@ -31,22 +31,15 @@ public final class ArrivalTimeFilterFact
{
@Override
- public MessageFilter newInstance(final List<Object> arguments)
+ public MessageFilter newInstance(final List<String> arguments)
{
if(arguments == null || arguments.size() != 1)
{
throw new IllegalArgumentException("Cannot create a filter from these arguments: " + arguments);
}
- Object arg = arguments.get(0);
- long startingFrom;
- if(arg instanceof Number)
- {
- startingFrom = ((Number)arg).longValue();
- }
- else
- {
- startingFrom = Long.parseLong(String.valueOf(arg));
- }
+ String arg = arguments.get(0);
+ long startingFrom= Long.parseLong(arg);
+
return new ArrivalTimeFilter(startingFrom);
}
Modified: qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/filter/JMSSelectorFilterFactory.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/filter/JMSSelectorFilterFactory.java?rev=1663170&r1=1663169&r2=1663170&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/filter/JMSSelectorFilterFactory.java (original)
+++ qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/filter/JMSSelectorFilterFactory.java Sun Mar 1 21:33:36 2015
@@ -38,16 +38,16 @@ public final class JMSSelectorFilterFact
}
@Override
- public MessageFilter newInstance(final List<Object> arguments)
+ public MessageFilter newInstance(final List<String> arguments)
{
if(arguments == null || arguments.size() != 1)
{
throw new IllegalArgumentException("Cannot create a filter from these arguments: " + arguments);
}
- Object arg = arguments.get(0);
+ String arg = arguments.get(0);
try
{
- return new JMSSelectorFilter(String.valueOf(arg));
+ return new JMSSelectorFilter(arg);
}
catch (ParseException | TokenMgrError | SelectorParsingException e)
{
Modified: qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/AttributeValueConverter.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/AttributeValueConverter.java?rev=1663170&r1=1663169&r2=1663170&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/AttributeValueConverter.java (original)
+++ qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/AttributeValueConverter.java Sun Mar 1 21:33:36 2015
@@ -50,6 +50,25 @@ abstract class AttributeValueConverter<T
}
};
+ static final AttributeValueConverter<Object> OBJECT_CONVERTER = new AttributeValueConverter<Object>()
+ {
+ @Override
+ public Object convert(final Object value, final ConfiguredObject object)
+ {
+ if(value instanceof String)
+ {
+ return AbstractConfiguredObject.interpolate(object, (String) value);
+ }
+ else if(value == null)
+ {
+ return null;
+ }
+ else
+ {
+ return value;
+ }
+ }
+ };
static final AttributeValueConverter<UUID> UUID_CONVERTER = new AttributeValueConverter<UUID>()
{
@Override
@@ -398,7 +417,17 @@ abstract class AttributeValueConverter<T
}
else if(Map.class.isAssignableFrom(type))
{
- return (AttributeValueConverter<X>) MAP_CONVERTER;
+ if(returnType instanceof ParameterizedType)
+ {
+ Type keyType = ((ParameterizedType) returnType).getActualTypeArguments()[0];
+ Type valueType = ((ParameterizedType) returnType).getActualTypeArguments()[1];
+
+ return (AttributeValueConverter<X>) new GenericMapConverter(keyType,valueType);
+ }
+ else
+ {
+ return (AttributeValueConverter<X>) MAP_CONVERTER;
+ }
}
else if(Collection.class.isAssignableFrom(type))
{
@@ -416,6 +445,10 @@ abstract class AttributeValueConverter<T
{
return (AttributeValueConverter<X>) new ConfiguredObjectConverter(type);
}
+ else if(Object.class == type)
+ {
+ return (AttributeValueConverter<X>) OBJECT_CONVERTER;
+ }
throw new IllegalArgumentException("Cannot create attribute converter of type " + type.getName());
}
@@ -575,6 +608,62 @@ abstract class AttributeValueConverter<T
}
}
+ public static class GenericMapConverter extends AttributeValueConverter<Map>
+ {
+
+ private final AttributeValueConverter<?> _keyConverter;
+ private final AttributeValueConverter<?> _valueConverter;
+
+
+ public GenericMapConverter(final Type keyType, final Type valueType)
+ {
+ _keyConverter = getConverter(getRawType(keyType), keyType);
+
+ _valueConverter = getConverter(getRawType(valueType), valueType);
+ }
+
+
+ @Override
+ public Map convert(final Object value, final ConfiguredObject object)
+ {
+ if(value instanceof Map)
+ {
+ Map<?,?> original = (Map<?,?>)value;
+ Map converted = new LinkedHashMap(original.size());
+ for(Map.Entry<?,?> entry : original.entrySet())
+ {
+ converted.put(_keyConverter.convert(entry.getKey(),object),
+ _valueConverter.convert(entry.getValue(), object));
+ }
+ return Collections.unmodifiableMap(converted);
+ }
+ else if(value == null)
+ {
+ return null;
+ }
+ else
+ {
+ if(value instanceof String)
+ {
+ String interpolated = AbstractConfiguredObject.interpolate(object, (String) value);
+ ObjectMapper objectMapper = new ObjectMapper();
+ try
+ {
+ return convert(objectMapper.readValue(interpolated, Map.class), object);
+ }
+ catch (IOException e)
+ {
+ // fall through to the non-JSON single object case
+ }
+ }
+
+ throw new IllegalArgumentException("Cannot convert type " + value.getClass() + " to a Map");
+ }
+
+ }
+ }
+
+
static final class EnumConverter<X extends Enum<X>> extends AttributeValueConverter<X>
{
private final Class<X> _klazz;
Modified: qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/Queue.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/Queue.java?rev=1663170&r1=1663169&r2=1663170&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/Queue.java (original)
+++ qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/Queue.java Sun Mar 1 21:33:36 2015
@@ -163,7 +163,7 @@ public interface Queue<X extends Queue<X
long getMaximumMessageTtl();
@ManagedAttribute
- Map<String, Map<String,List<Object>>> getDefaultFilters();
+ Map<String, Map<String,List<String>>> getDefaultFilters();
//children
Collection<? extends Binding> getBindings();
Modified: qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/VirtualHost.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/VirtualHost.java?rev=1663170&r1=1663169&r2=1663170&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/VirtualHost.java (original)
+++ qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/VirtualHost.java Sun Mar 1 21:33:36 2015
@@ -28,6 +28,7 @@ import java.util.UUID;
import org.apache.qpid.server.configuration.updater.TaskExecutor;
import org.apache.qpid.server.message.MessageInstance;
+import org.apache.qpid.server.model.port.AmqpPort;
import org.apache.qpid.server.store.MessageStore;
@ManagedObject( defaultType = "ProvidedStore")
@@ -144,6 +145,8 @@ public interface VirtualHost<X extends V
void delete();
+ String getRedirectHost(AmqpPort<?> port);
+
public static interface Transaction
{
void dequeue(MessageInstance entry);
Modified: qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/plugin/MessageFilterFactory.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/plugin/MessageFilterFactory.java?rev=1663170&r1=1663169&r2=1663170&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/plugin/MessageFilterFactory.java (original)
+++ qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/plugin/MessageFilterFactory.java Sun Mar 1 21:33:36 2015
@@ -26,5 +26,5 @@ import org.apache.qpid.server.filter.Mes
public interface MessageFilterFactory extends Pluggable
{
- MessageFilter newInstance(List<Object> arguments);
+ MessageFilter newInstance(List<String> arguments);
}
Modified: qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java?rev=1663170&r1=1663169&r2=1663170&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java (original)
+++ qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java Sun Mar 1 21:33:36 2015
@@ -191,7 +191,7 @@ public abstract class AbstractQueue<X ex
private MessageDurability _messageDurability;
@ManagedAttributeField
- private Map<String, Map<String,List<Object>>> _defaultFilters;
+ private Map<String, Map<String,List<String>>> _defaultFilters;
private Object _exclusiveOwner; // could be connection, session, Principal or a String for the container name
@@ -467,17 +467,17 @@ public abstract class AbstractQueue<X ex
final Map<String, MessageFilterFactory> messageFilterFactories =
qpidServiceLoader.getInstancesByType(MessageFilterFactory.class);
- for (Map.Entry<String,Map<String,List<Object>>> entry : _defaultFilters.entrySet())
+ for (Map.Entry<String,Map<String,List<String>>> entry : _defaultFilters.entrySet())
{
String name = String.valueOf(entry.getKey());
- Map<String, List<Object>> filterValue = entry.getValue();
+ Map<String, List<String>> filterValue = entry.getValue();
if(filterValue.size() == 1)
{
String filterTypeName = String.valueOf(filterValue.keySet().iterator().next());
MessageFilterFactory filterFactory = messageFilterFactories.get(filterTypeName);
if(filterFactory != null)
{
- List<Object> filterArguments = filterValue.values().iterator().next();
+ List<String> filterArguments = filterValue.values().iterator().next();
_defaultFiltersMap.put(name, filterFactory.newInstance(filterArguments));
}
else
@@ -599,7 +599,7 @@ public abstract class AbstractQueue<X ex
}
@Override
- public Map<String, Map<String, List<Object>>> getDefaultFilters()
+ public Map<String, Map<String, List<String>>> getDefaultFilters()
{
return _defaultFilters;
}
Modified: qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java?rev=1663170&r1=1663169&r2=1663170&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java (original)
+++ qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java Sun Mar 1 21:33:36 2015
@@ -63,6 +63,7 @@ import org.apache.qpid.server.message.Me
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.model.*;
import org.apache.qpid.server.model.adapter.ConnectionAdapter;
+import org.apache.qpid.server.model.port.AmqpPort;
import org.apache.qpid.server.plugin.ConnectionValidator;
import org.apache.qpid.server.plugin.QpidServiceLoader;
import org.apache.qpid.server.plugin.SystemNodeCreator;
@@ -934,6 +935,12 @@ public abstract class AbstractVirtualHos
}
}
+ @Override
+ public String getRedirectHost(final AmqpPort<?> port)
+ {
+ return null;
+ }
+
private class VirtualHostHouseKeepingTask extends HouseKeepingTask
{
public VirtualHostHouseKeepingTask()
Added: qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhostnode/RedirectingVirtualHost.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhostnode/RedirectingVirtualHost.java?rev=1663170&view=auto
==============================================================================
--- qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhostnode/RedirectingVirtualHost.java (added)
+++ qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhostnode/RedirectingVirtualHost.java Sun Mar 1 21:33:36 2015
@@ -0,0 +1,32 @@
+/*
+ *
+ * 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.virtualhostnode;
+
+import org.apache.qpid.server.exchange.ExchangeImpl;
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.virtualhost.NonStandardVirtualHost;
+import org.apache.qpid.server.virtualhost.VirtualHostImpl;
+
+public interface RedirectingVirtualHost<X extends RedirectingVirtualHost<X>>
+ extends VirtualHostImpl<X, AMQQueue<?>, ExchangeImpl<?>>,
+ NonStandardVirtualHost<X,AMQQueue<?>,ExchangeImpl<?>>
+{
+}
Propchange: qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhostnode/RedirectingVirtualHost.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhostnode/RedirectingVirtualHostImpl.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhostnode/RedirectingVirtualHostImpl.java?rev=1663170&view=auto
==============================================================================
--- qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhostnode/RedirectingVirtualHostImpl.java (added)
+++ qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhostnode/RedirectingVirtualHostImpl.java Sun Mar 1 21:33:36 2015
@@ -0,0 +1,492 @@
+/*
+ *
+ * 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.virtualhostnode;
+
+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.ScheduledFuture;
+
+import org.apache.qpid.server.connection.IConnectionRegistry;
+import org.apache.qpid.server.exchange.ExchangeImpl;
+import org.apache.qpid.server.logging.EventLogger;
+import org.apache.qpid.server.message.MessageDestination;
+import org.apache.qpid.server.message.MessageSource;
+import org.apache.qpid.server.model.AbstractConfiguredObject;
+import org.apache.qpid.server.model.BrokerModel;
+import org.apache.qpid.server.model.ConfiguredObject;
+import org.apache.qpid.server.model.Connection;
+import org.apache.qpid.server.model.Exchange;
+import org.apache.qpid.server.model.ManagedAttributeField;
+import org.apache.qpid.server.model.ManagedObject;
+import org.apache.qpid.server.model.ManagedObjectFactoryConstructor;
+import org.apache.qpid.server.model.State;
+import org.apache.qpid.server.model.VirtualHostAlias;
+import org.apache.qpid.server.model.VirtualHostNode;
+import org.apache.qpid.server.model.port.AmqpPort;
+import org.apache.qpid.server.protocol.AMQConnectionModel;
+import org.apache.qpid.server.protocol.LinkRegistry;
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.stats.StatisticsCounter;
+import org.apache.qpid.server.store.DurableConfigurationStore;
+import org.apache.qpid.server.store.MessageStore;
+import org.apache.qpid.server.txn.DtxRegistry;
+import org.apache.qpid.server.virtualhost.ExchangeIsAlternateException;
+import org.apache.qpid.server.virtualhost.HouseKeepingTask;
+import org.apache.qpid.server.virtualhost.RequiredExchangeException;
+
+@ManagedObject( category = false, type = RedirectingVirtualHostImpl.TYPE, register = false )
+class RedirectingVirtualHostImpl
+ extends AbstractConfiguredObject<RedirectingVirtualHostImpl>
+ implements RedirectingVirtualHost<RedirectingVirtualHostImpl>
+{
+ public static final String TYPE = "REDIRECTOR";
+ private final StatisticsCounter _messagesDelivered, _dataDelivered, _messagesReceived, _dataReceived;
+
+ @ManagedAttributeField
+ private boolean _queue_deadLetterQueueEnabled;
+
+ @ManagedAttributeField
+ private long _housekeepingCheckPeriod;
+
+ @ManagedAttributeField
+ private long _storeTransactionIdleTimeoutClose;
+
+ @ManagedAttributeField
+ private long _storeTransactionIdleTimeoutWarn;
+
+ @ManagedAttributeField
+ private long _storeTransactionOpenTimeoutClose;
+
+ @ManagedAttributeField
+ private long _storeTransactionOpenTimeoutWarn;
+ @ManagedAttributeField
+ private int _housekeepingThreadCount;
+
+ @ManagedAttributeField
+ private List<String> _enabledConnectionValidators;
+
+ @ManagedAttributeField
+ private List<String> _disabledConnectionValidators;
+
+
+ @ManagedObjectFactoryConstructor
+ public RedirectingVirtualHostImpl(final Map<String, Object> attributes, VirtualHostNode<?> virtualHostNode)
+ {
+ super(parentsMap(virtualHostNode), attributes);
+
+ _messagesDelivered = new StatisticsCounter("messages-delivered-" + getName());
+ _dataDelivered = new StatisticsCounter("bytes-delivered-" + getName());
+ _messagesReceived = new StatisticsCounter("messages-received-" + getName());
+ _dataReceived = new StatisticsCounter("bytes-received-" + getName());
+ setState(State.UNAVAILABLE);
+ }
+
+ @Override
+ protected void validateChange(final ConfiguredObject<?> proxyForValidation, final Set<String> changedAttributes)
+ {
+ super.validateChange(proxyForValidation, changedAttributes);
+
+ throwUnsupportedForRedirector();
+ }
+
+ @Override
+ public String getModelVersion()
+ {
+ return BrokerModel.MODEL_VERSION;
+ }
+
+ @Override
+ protected <C extends ConfiguredObject> C addChild(final Class<C> childClass,
+ final Map<String, Object> attributes,
+ final ConfiguredObject... otherParents)
+ {
+ throwUnsupportedForRedirector();
+ return null;
+ }
+
+ @Override
+ public ExchangeImpl createExchange(final Map<String, Object> attributes)
+ {
+ throwUnsupportedForRedirector();
+ return null;
+ }
+
+ @Override
+ public void removeExchange(final ExchangeImpl<?> exchange, final boolean force)
+ throws ExchangeIsAlternateException, RequiredExchangeException
+ {
+ throwUnsupportedForRedirector();
+ }
+
+ @Override
+ public MessageDestination getMessageDestination(final String name)
+ {
+ return null;
+ }
+
+ @Override
+ public ExchangeImpl<?> getExchange(final String name)
+ {
+ return null;
+ }
+
+ @Override
+ public AMQQueue<?> createQueue(final Map<String, Object> attributes)
+ {
+ throwUnsupportedForRedirector();
+ return null;
+ }
+
+ @Override
+ public void executeTransaction(final TransactionalOperation op)
+ {
+ throwUnsupportedForRedirector();
+ }
+
+ @Override
+ public Collection<String> getExchangeTypeNames()
+ {
+ return getObjectFactory().getSupportedTypes(Exchange.class);
+ }
+
+ @Override
+ public String getRedirectHost(final AmqpPort<?> port)
+ {
+ return ((RedirectingVirtualHostNode<?>)(getParent(VirtualHostNode.class))).getRedirects().get(port);
+ }
+
+ @Override
+ public boolean isQueue_deadLetterQueueEnabled()
+ {
+ return false;
+ }
+
+ @Override
+ public long getHousekeepingCheckPeriod()
+ {
+ return 0;
+ }
+
+ @Override
+ public long getStoreTransactionIdleTimeoutClose()
+ {
+ return 0;
+ }
+
+ @Override
+ public long getStoreTransactionIdleTimeoutWarn()
+ {
+ return 0;
+ }
+
+ @Override
+ public long getStoreTransactionOpenTimeoutClose()
+ {
+ return 0;
+ }
+
+ @Override
+ public long getStoreTransactionOpenTimeoutWarn()
+ {
+ return 0;
+ }
+
+ @Override
+ public int getHousekeepingThreadCount()
+ {
+ return 0;
+ }
+
+ @Override
+ public long getQueueCount()
+ {
+ return 0;
+ }
+
+ @Override
+ public long getExchangeCount()
+ {
+ return 0;
+ }
+
+ @Override
+ public long getConnectionCount()
+ {
+ return 0;
+ }
+
+ @Override
+ public long getBytesIn()
+ {
+ return 0;
+ }
+
+ @Override
+ public long getBytesOut()
+ {
+ return 0;
+ }
+
+ @Override
+ public long getMessagesIn()
+ {
+ return 0;
+ }
+
+ @Override
+ public long getMessagesOut()
+ {
+ return 0;
+ }
+
+ @Override
+ public Collection<VirtualHostAlias> getAliases()
+ {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public Collection<Connection> getConnections()
+ {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public IConnectionRegistry getConnectionRegistry()
+ {
+ return null;
+ }
+
+ @Override
+ public AMQQueue<?> getQueue(final String name)
+ {
+ return null;
+ }
+
+ @Override
+ public MessageSource getMessageSource(final String name)
+ {
+ return null;
+ }
+
+ @Override
+ public AMQQueue<?> getQueue(final UUID id)
+ {
+ return null;
+ }
+
+ @Override
+ public Collection<AMQQueue<?>> getQueues()
+ {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public int removeQueue(final AMQQueue<?> queue)
+ {
+ throwUnsupportedForRedirector();
+ return 0;
+ }
+
+ @Override
+ public Collection<ExchangeImpl<?>> getExchanges()
+ {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public DurableConfigurationStore getDurableConfigurationStore()
+ {
+ return null;
+ }
+
+ @Override
+ public ExchangeImpl<?> getExchange(final UUID id)
+ {
+ return null;
+ }
+
+ @Override
+ public MessageDestination getDefaultDestination()
+ {
+ return null;
+ }
+
+ @Override
+ public MessageStore getMessageStore()
+ {
+ return null;
+ }
+
+ @Override
+ public void setTargetSize(final long targetSize)
+ {
+
+ }
+
+ @Override
+ public long getTotalQueueDepthBytes()
+ {
+ return 0l;
+ }
+
+ @Override
+ public org.apache.qpid.server.security.SecurityManager getSecurityManager()
+ {
+ return null;
+ }
+
+ @Override
+ public void scheduleHouseKeepingTask(final long period, final HouseKeepingTask task)
+ {
+ }
+
+ @Override
+ public long getHouseKeepingTaskCount()
+ {
+ return 0;
+ }
+
+ @Override
+ public long getHouseKeepingCompletedTaskCount()
+ {
+ return 0;
+ }
+
+ @Override
+ public int getHouseKeepingPoolSize()
+ {
+ return 0;
+ }
+
+ @Override
+ public void setHouseKeepingPoolSize(final int newSize)
+ {
+ }
+
+ @Override
+ public int getHouseKeepingActiveCount()
+ {
+ return 0;
+ }
+
+ @Override
+ public DtxRegistry getDtxRegistry()
+ {
+ return null;
+ }
+
+ @Override
+ public LinkRegistry getLinkRegistry(final String remoteContainerId)
+ {
+ return null;
+ }
+
+ @Override
+ public ScheduledFuture<?> scheduleTask(final long delay, final Runnable timeoutTask)
+ {
+ throwUnsupportedForRedirector();
+ return null;
+ }
+
+ @Override
+ public boolean getDefaultDeadLetterQueueEnabled()
+ {
+ return false;
+ }
+
+ @Override
+ public EventLogger getEventLogger()
+ {
+ return null;
+ }
+
+ @Override
+ public void registerMessageReceived(final long messageSize, final long timestamp)
+ {
+ throwUnsupportedForRedirector();
+ }
+
+ @Override
+ public void registerMessageDelivered(final long messageSize)
+ {
+ throwUnsupportedForRedirector();
+ }
+
+ @Override
+ public StatisticsCounter getMessageDeliveryStatistics()
+ {
+ return _messagesDelivered;
+ }
+
+ @Override
+ public StatisticsCounter getMessageReceiptStatistics()
+ {
+ return _messagesReceived;
+ }
+
+ @Override
+ public StatisticsCounter getDataDeliveryStatistics()
+ {
+ return _dataDelivered;
+ }
+
+ @Override
+ public StatisticsCounter getDataReceiptStatistics()
+ {
+ return _dataReceived;
+ }
+
+ @Override
+ public void resetStatistics()
+ {
+ }
+
+ @Override
+ public boolean authoriseCreateConnection(final AMQConnectionModel<?, ?> connection)
+ {
+ return false;
+ }
+
+ @Override
+ public List<String> getEnabledConnectionValidators()
+ {
+ return _enabledConnectionValidators;
+ }
+
+ @Override
+ public List<String> getDisabledConnectionValidators()
+ {
+ return _disabledConnectionValidators;
+ }
+
+ private void throwUnsupportedForRedirector()
+ {
+ throw new IllegalStateException("The virtual host state of " + getState()
+ + " does not permit this operation.");
+ }
+
+
+}
Propchange: qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhostnode/RedirectingVirtualHostImpl.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhostnode/RedirectingVirtualHostNode.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhostnode/RedirectingVirtualHostNode.java?rev=1663170&view=auto
==============================================================================
--- qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhostnode/RedirectingVirtualHostNode.java (added)
+++ qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhostnode/RedirectingVirtualHostNode.java Sun Mar 1 21:33:36 2015
@@ -0,0 +1,36 @@
+/*
+ *
+ * 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.virtualhostnode;
+
+import java.util.Map;
+
+import org.apache.qpid.server.model.ManagedAttribute;
+import org.apache.qpid.server.model.ManagedObject;
+import org.apache.qpid.server.model.Port;
+import org.apache.qpid.server.model.VirtualHostNode;
+
+@ManagedObject(type= RedirectingVirtualHostNodeImpl.VIRTUAL_HOST_NODE_TYPE, category=false, validChildTypes = "org.apache.qpid.server.virtualhostnode.RedirectingVirtualHostNodeImpl#getSupportedChildTypes()")
+public interface RedirectingVirtualHostNode<X extends RedirectingVirtualHostNode<X>> extends VirtualHostNode<X>
+{
+
+ @ManagedAttribute( defaultValue = "{}")
+ Map<Port<?>, String> getRedirects();
+}
Propchange: qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhostnode/RedirectingVirtualHostNode.java
------------------------------------------------------------------------------
svn:eol-style = native
Added: qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhostnode/RedirectingVirtualHostNodeImpl.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhostnode/RedirectingVirtualHostNodeImpl.java?rev=1663170&view=auto
==============================================================================
--- qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhostnode/RedirectingVirtualHostNodeImpl.java (added)
+++ qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhostnode/RedirectingVirtualHostNodeImpl.java Sun Mar 1 21:33:36 2015
@@ -0,0 +1,124 @@
+/*
+ *
+ * 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.virtualhostnode;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Map;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.qpid.server.model.AbstractConfiguredObject;
+import org.apache.qpid.server.model.Broker;
+import org.apache.qpid.server.model.ConfiguredObject;
+import org.apache.qpid.server.model.ManagedAttributeField;
+import org.apache.qpid.server.model.ManagedObjectFactoryConstructor;
+import org.apache.qpid.server.model.Port;
+import org.apache.qpid.server.model.RemoteReplicationNode;
+import org.apache.qpid.server.model.State;
+import org.apache.qpid.server.model.StateTransition;
+import org.apache.qpid.server.model.VirtualHost;
+import org.apache.qpid.server.store.DurableConfigurationStore;
+
+
+public class RedirectingVirtualHostNodeImpl
+ extends AbstractConfiguredObject<RedirectingVirtualHostNodeImpl> implements RedirectingVirtualHostNode<RedirectingVirtualHostNodeImpl>
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(RedirectingVirtualHostImpl.class);
+ public static final String VIRTUAL_HOST_NODE_TYPE = "Redirector";
+
+
+ @ManagedAttributeField
+ private String _virtualHostInitialConfiguration;
+
+ @ManagedAttributeField
+ private Map<Port<?>,String> _redirects;
+
+ private RedirectingVirtualHostImpl _virtualHost;
+
+ @ManagedObjectFactoryConstructor
+ public RedirectingVirtualHostNodeImpl(Map<String, Object> attributes, Broker<?> parent)
+ {
+ super(Collections.<Class<? extends ConfiguredObject>,ConfiguredObject<?>>singletonMap(Broker.class, parent),
+ attributes);
+ }
+
+ @StateTransition( currentState = {State.UNINITIALIZED, State.STOPPED, State.ERRORED }, desiredState = State.ACTIVE )
+ protected void doActivate()
+ {
+ try
+ {
+ _virtualHost = new RedirectingVirtualHostImpl(Collections.<String,Object>singletonMap(ConfiguredObject.NAME,getName()), this);
+ _virtualHost.create();
+ setState(State.ACTIVE);
+ }
+ catch(RuntimeException e)
+ {
+ setState(State.ERRORED);
+ if (getParent(Broker.class).isManagementMode())
+ {
+ LOGGER.warn("Failed to make " + this + " active.", e);
+ }
+ else
+ {
+ throw e;
+ }
+ }
+ }
+
+ @Override
+ public String getVirtualHostInitialConfiguration()
+ {
+ return _virtualHostInitialConfiguration;
+ }
+
+ @Override
+ public VirtualHost<?, ?, ?> getVirtualHost()
+ {
+ return _virtualHost;
+ }
+
+ @Override
+ public DurableConfigurationStore getConfigurationStore()
+ {
+ return null;
+ }
+
+ @Override
+ public Collection<? extends RemoteReplicationNode> getRemoteReplicationNodes()
+ {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public Map<Port<?>, String> getRedirects()
+ {
+ return _redirects;
+ }
+
+ public static Map<String, Collection<String>> getSupportedChildTypes()
+ {
+ Collection<String> validVhostTypes = Collections.singleton(RedirectingVirtualHostImpl.TYPE);
+ return Collections.singletonMap(VirtualHost.class.getSimpleName(), validVhostTypes);
+ }
+
+}
Propchange: qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/virtualhostnode/RedirectingVirtualHostNodeImpl.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified: qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnectionDelegate.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnectionDelegate.java?rev=1663170&r1=1663169&r2=1663170&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnectionDelegate.java (original)
+++ qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnectionDelegate.java Sun Mar 1 21:33:36 2015
@@ -184,7 +184,8 @@ public class ServerConnectionDelegate ex
vhostName = "";
}
- vhost = ((AmqpPort)sconn.getPort()).getVirtualHost(vhostName);
+ AmqpPort port = (AmqpPort) sconn.getPort();
+ vhost = port.getVirtualHost(vhostName);
@@ -193,7 +194,16 @@ public class ServerConnectionDelegate ex
if (vhost.getState() != State.ACTIVE)
{
sconn.setState(Connection.State.CLOSING);
- sconn.invoke(new ConnectionClose(ConnectionCloseCode.CONNECTION_FORCED, "Virtual host '"+vhostName+"' is not active"));
+ final String redirectHost = vhost.getRedirectHost(port);
+ if(redirectHost == null)
+ {
+ sconn.invoke(new ConnectionClose(ConnectionCloseCode.CONNECTION_FORCED,
+ "Virtual host '" + vhostName + "' is not active"));
+ }
+ else
+ {
+ sconn.invoke(new ConnectionRedirect(redirectHost, new ArrayList<Object>()));
+ }
return;
}
Modified: qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java?rev=1663170&r1=1663169&r2=1663170&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java (original)
+++ qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java Sun Mar 1 21:33:36 2015
@@ -1544,7 +1544,7 @@ public class AMQProtocolEngine implement
virtualHostStr = virtualHostName == null ? null : virtualHostName.toString();
}
- VirtualHostImpl virtualHost = ((AmqpPort)getPort()).getVirtualHost(virtualHostStr);
+ VirtualHostImpl<?,?,?> virtualHost = ((AmqpPort)getPort()).getVirtualHost(virtualHostStr);
if (virtualHost == null)
{
@@ -1557,8 +1557,16 @@ public class AMQProtocolEngine implement
// Check virtualhost access
if (virtualHost.getState() != State.ACTIVE)
{
- closeConnection(AMQConstant.CONNECTION_FORCED,
- "Virtual host '" + virtualHost.getName() + "' is not active",0);
+ String redirectHost = virtualHost.getRedirectHost(getPort());
+ if(redirectHost != null)
+ {
+ closeConnection(0, new AMQFrame(0,new ConnectionRedirectBody(getProtocolVersion(),AMQShortString.valueOf(redirectHost), null)));
+ }
+ else
+ {
+ closeConnection(AMQConstant.CONNECTION_FORCED,
+ "Virtual host '" + virtualHost.getName() + "' is not active", 0);
+ }
}
else
@@ -1754,7 +1762,7 @@ public class AMQProtocolEngine implement
_logger.info("Locale selected: " + locale);
SubjectCreator subjectCreator = getSubjectCreator();
- SaslServer ss = null;
+ SaslServer ss;
try
{
ss = subjectCreator.createSaslServer(String.valueOf(mechanism),
Modified: qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQBrokerDetails.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQBrokerDetails.java?rev=1663170&r1=1663169&r2=1663170&view=diff
==============================================================================
--- qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQBrokerDetails.java (original)
+++ qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQBrokerDetails.java Sun Mar 1 21:33:36 2015
@@ -20,18 +20,18 @@
*/
package org.apache.qpid.client;
-import org.apache.qpid.configuration.ClientProperties;
-import org.apache.qpid.jms.BrokerDetails;
-import org.apache.qpid.transport.ConnectionSettings;
-import org.apache.qpid.url.URLHelper;
-import org.apache.qpid.url.URLSyntaxException;
-
import java.io.Serializable;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.HashMap;
import java.util.Map;
+import org.apache.qpid.configuration.ClientProperties;
+import org.apache.qpid.jms.BrokerDetails;
+import org.apache.qpid.transport.ConnectionSettings;
+import org.apache.qpid.url.URLHelper;
+import org.apache.qpid.url.URLSyntaxException;
+
public class AMQBrokerDetails implements BrokerDetails, Serializable
{
private static final long serialVersionUID = 8450786374975932890L;
@@ -42,6 +42,14 @@ public class AMQBrokerDetails implements
private Map<String, String> _options = new HashMap<String, String>();
+ public AMQBrokerDetails(BrokerDetails details)
+ {
+ _host = details.getHost();
+ _port = details.getPort();
+ _transport = details.getTransport();
+ _options = new HashMap<>(details.getProperties());
+ }
+
public AMQBrokerDetails(){}
public AMQBrokerDetails(String url) throws URLSyntaxException
Modified: qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java?rev=1663170&r1=1663169&r2=1663170&view=diff
==============================================================================
--- qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java (original)
+++ qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java Sun Mar 1 21:33:36 2015
@@ -62,6 +62,7 @@ import org.apache.qpid.AMQDisconnectedEx
import org.apache.qpid.AMQException;
import org.apache.qpid.AMQProtocolException;
import org.apache.qpid.AMQUnresolvedAddressException;
+import org.apache.qpid.client.failover.ConnectionRedirectException;
import org.apache.qpid.client.failover.FailoverException;
import org.apache.qpid.client.failover.FailoverProtectedOperation;
import org.apache.qpid.client.protocol.AMQProtocolHandler;
@@ -462,9 +463,22 @@ public class AMQConnection extends Close
}
else if (!isConnected())
{
- retryAllowed = _failoverPolicy.failoverAllowed();
- brokerDetails = _failoverPolicy.getNextBrokerDetails();
- _protocolHandler.setStateManager(new AMQStateManager(_protocolHandler.getProtocolSession()));
+ if(connectionException instanceof ConnectionRedirectException)
+ {
+ ConnectionRedirectException redirect = (ConnectionRedirectException) connectionException;
+ retryAllowed = true;
+ brokerDetails = new AMQBrokerDetails(brokerDetails);
+ brokerDetails.setHost(redirect.getHost());
+ brokerDetails.setPort(redirect.getPort());
+ _protocolHandler.setStateManager(new AMQStateManager(_protocolHandler.getProtocolSession()));
+
+ }
+ else
+ {
+ retryAllowed = _failoverPolicy.failoverAllowed();
+ brokerDetails = _failoverPolicy.getNextBrokerDetails();
+ _protocolHandler.setStateManager(new AMQStateManager(_protocolHandler.getProtocolSession()));
+ }
}
}
@@ -599,9 +613,11 @@ public class AMQConnection extends Close
_virtualHost = virtualHost;
}
- public boolean attemptReconnection(String host, int port)
+ public boolean attemptReconnection(String host, int port, final boolean useFailoverConfigOnFailure)
{
- BrokerDetails bd = new AMQBrokerDetails(host, port);
+ BrokerDetails bd = new AMQBrokerDetails(_failoverPolicy.getCurrentBrokerDetails());
+ bd.setHost(host);
+ bd.setPort(port);
_failoverPolicy.setBroker(bd);
@@ -618,10 +634,9 @@ public class AMQConnection extends Close
_logger.info("Unable to connect to broker at " + bd);
}
- attemptReconnection();
+ return useFailoverConfigOnFailure && attemptReconnection();
}
- return false;
}
public boolean attemptReconnection()
@@ -629,32 +644,41 @@ public class AMQConnection extends Close
BrokerDetails broker = null;
while (_failoverPolicy.failoverAllowed() && (broker = _failoverPolicy.getNextBrokerDetails()) != null)
{
- try
+ if (attemptConnection(broker))
{
- makeBrokerConnection(broker);
return true;
}
- catch (Exception e)
+ }
+
+ // connection unsuccessful
+ return false;
+ }
+
+ private boolean attemptConnection(final BrokerDetails broker)
+ {
+ try
+ {
+ makeBrokerConnection(broker);
+ return true;
+ }
+ catch (Exception e)
+ {
+ if (!(e instanceof AMQException))
{
- if (!(e instanceof AMQException))
+ if (_logger.isInfoEnabled())
{
- if (_logger.isInfoEnabled())
- {
- _logger.info("Unable to connect to broker at " + _failoverPolicy.getCurrentBrokerDetails(), e);
- }
+ _logger.info("Unable to connect to broker at " + _failoverPolicy.getCurrentBrokerDetails(), e);
}
- else
+ }
+ else
+ {
+ if (_logger.isInfoEnabled())
{
- if (_logger.isInfoEnabled())
- {
- _logger.info(e.getMessage() + ":Unable to connect to broker at "
- + _failoverPolicy.getCurrentBrokerDetails());
- }
+ _logger.info(e.getMessage() + ":Unable to connect to broker at "
+ + _failoverPolicy.getCurrentBrokerDetails());
}
}
}
-
- // connection unsuccessful
return false;
}
Modified: qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java?rev=1663170&r1=1663169&r2=1663170&view=diff
==============================================================================
--- qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java (original)
+++ qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java Sun Mar 1 21:33:36 2015
@@ -61,6 +61,8 @@ import org.apache.qpid.transport.Transpo
public class AMQConnectionDelegate_0_10 implements AMQConnectionDelegate, ConnectionListener
{
+ private static final int DEFAULT_PORT = 5672;
+
/**
* This class logger.
*/
@@ -238,7 +240,7 @@ public class AMQConnectionDelegate_0_10
{
code = AMQConstant.getConstant(ce.getClose().getReplyCode().getValue());
}
- String msg = "Cannot connect to broker: " + ce.getMessage();
+ String msg = "Cannot connect to broker ("+brokerDetail+"): " + ce.getMessage();
throw new AMQException(code, msg, ce);
}
@@ -314,25 +316,39 @@ public class AMQConnectionDelegate_0_10
@Override
public void run()
{
- try
- {
- if (_conn.firePreFailover(false) && _conn.attemptReconnection())
+ try
{
- failoverPrep();
- _conn.resubscribeSessions();
- _conn.fireFailoverComplete();
- failoverDone.set(true);
+ boolean preFailover = _conn.firePreFailover(false);
+ if (preFailover)
+ {
+ boolean reconnected;
+ if(exc instanceof RedirectConnectionException)
+ {
+ RedirectConnectionException redirect = (RedirectConnectionException)exc;
+ reconnected = attemptRedirection(redirect.getHost(), redirect.getKnownHosts());
+ }
+ else
+ {
+ reconnected = _conn.attemptReconnection();
+ }
+ if(reconnected)
+ {
+ failoverPrep();
+ _conn.resubscribeSessions();
+ _conn.fireFailoverComplete();
+ failoverDone.set(true);
+ }
+ }
+ }
+ catch (Exception e)
+ {
+ _logger.error("error during failover", e);
+ }
+ finally
+ {
+ _conn.getProtocolHandler().getFailoverLatch().countDown();
+ _conn.getProtocolHandler().setFailoverLatch(null);
}
- }
- catch (Exception e)
- {
- _logger.error("error during failover", e);
- }
- finally
- {
- _conn.getProtocolHandler().getFailoverLatch().countDown();
- _conn.getProtocolHandler().setFailoverLatch(null);
- }
}
});
@@ -376,6 +392,58 @@ public class AMQConnectionDelegate_0_10
}
}
+ @Override
+ public boolean redirect(final String host, final List<Object> knownHosts)
+ {
+ exception = new RedirectConnectionException(host,knownHosts);
+
+ return false;
+ }
+
+ private boolean attemptRedirection(String host, List<Object> knownHosts)
+ {
+
+ boolean redirected = host != null && attemptRedirection(host);
+ if(knownHosts != null)
+ {
+ for(Object knownHost : knownHosts)
+ {
+ redirected = attemptRedirection(String.valueOf(knownHost));
+ if(redirected)
+ {
+ break;
+ }
+ }
+ }
+ return redirected;
+ }
+
+ private boolean attemptRedirection(String host)
+ {
+ int portIndex = host.indexOf(':');
+
+ int port;
+ if (portIndex == -1)
+ {
+ port = DEFAULT_PORT;
+ }
+ else
+ {
+ try
+ {
+ port = Integer.parseInt(host.substring(portIndex + 1));
+ }
+ catch(NumberFormatException e)
+ {
+ _logger.info("Unable to redirect to " + host + " - does not look like a valid address");
+ return false;
+ }
+ host = host.substring(0, portIndex);
+
+ }
+ return _conn.attemptReconnection(host,port,false);
+ }
+
public <T, E extends Exception> T executeRetrySupport(FailoverProtectedOperation<T,E> operation) throws E
{
if (_conn.isFailingOver())
@@ -538,4 +606,28 @@ public class AMQConnectionDelegate_0_10
{
return _qpidConnection.isMessageCompressionSupported();
}
+
+ private class RedirectConnectionException extends ConnectionException
+ {
+ private final String _host;
+ private final List<Object> _knownHosts;
+
+ public RedirectConnectionException(final String host,
+ final List<Object> knownHosts)
+ {
+ super("Connection redirected to " + host + " alternates " + knownHosts);
+ _host = host;
+ _knownHosts = knownHosts;
+ }
+
+ public String getHost()
+ {
+ return _host;
+ }
+
+ public List<Object> getKnownHosts()
+ {
+ return _knownHosts;
+ }
+ }
}
Added: qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/failover/ConnectionRedirectException.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/failover/ConnectionRedirectException.java?rev=1663170&view=auto
==============================================================================
--- qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/failover/ConnectionRedirectException.java (added)
+++ qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/failover/ConnectionRedirectException.java Sun Mar 1 21:33:36 2015
@@ -0,0 +1,46 @@
+/*
+ *
+ * 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.client.failover;
+
+import org.apache.qpid.AMQException;
+
+public class ConnectionRedirectException extends AMQException
+{
+ private final String _host;
+ private final int _port;
+
+ public ConnectionRedirectException(final String host, final int port)
+ {
+ super("Redirecting to " + host + ":" + port);
+ _host = host;
+ _port = port;
+ }
+
+ public String getHost()
+ {
+ return _host;
+ }
+
+ public int getPort()
+ {
+ return _port;
+ }
+}
Propchange: qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/failover/ConnectionRedirectException.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified: qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/failover/FailoverHandler.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/failover/FailoverHandler.java?rev=1663170&r1=1663169&r2=1663170&view=diff
==============================================================================
--- qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/failover/FailoverHandler.java (original)
+++ qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/failover/FailoverHandler.java Sun Mar 1 21:33:36 2015
@@ -20,6 +20,8 @@
*/
package org.apache.qpid.client.failover;
+import java.util.concurrent.CountDownLatch;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -28,8 +30,6 @@ import org.apache.qpid.client.protocol.A
import org.apache.qpid.client.state.AMQState;
import org.apache.qpid.client.state.AMQStateManager;
-import java.util.concurrent.CountDownLatch;
-
/**
* FailoverHandler is a continuation that performs the failover procedure on a protocol session. As described in the
* class level comment for {@link AMQProtocolHandler}, a protocol connection can span many physical transport
@@ -168,7 +168,7 @@ public class FailoverHandler implements
// if _host has value then we are performing a redirect.
if (_host != null)
{
- failoverSucceeded = _amqProtocolHandler.getConnection().attemptReconnection(_host, _port);
+ failoverSucceeded = _amqProtocolHandler.getConnection().attemptReconnection(_host, _port, true);
}
else
{
Modified: qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionRedirectMethodHandler.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionRedirectMethodHandler.java?rev=1663170&r1=1663169&r2=1663170&view=diff
==============================================================================
--- qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionRedirectMethodHandler.java (original)
+++ qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ConnectionRedirectMethodHandler.java Sun Mar 1 21:33:36 2015
@@ -20,13 +20,18 @@
*/
package org.apache.qpid.client.handler;
+import java.nio.ByteBuffer;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.qpid.AMQException;
+import org.apache.qpid.client.failover.ConnectionRedirectException;
import org.apache.qpid.client.protocol.AMQProtocolSession;
import org.apache.qpid.client.state.StateAwareMethodListener;
import org.apache.qpid.framing.ConnectionRedirectBody;
+import org.apache.qpid.transport.Sender;
+import org.apache.qpid.transport.TransportException;
public class ConnectionRedirectMethodHandler implements StateAwareMethodListener<ConnectionRedirectBody>
{
@@ -65,7 +70,21 @@ public class ConnectionRedirectMethodHan
}
- session.failover(host, port);
+ session.notifyError(new ConnectionRedirectException(host,port));
+
+ Sender<ByteBuffer> sender = session.getSender();
+
+ // Close the open TCP connection
+ try
+ {
+ sender.close();
+ }
+ catch(TransportException e)
+ {
+ //Ignore, they are already logged by the Sender and this
+ //is a connection-close being processed by the IoReceiver
+ //which will as it closes initiate failover if necessary.
+ }
}
}
Modified: qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java?rev=1663170&r1=1663169&r2=1663170&view=diff
==============================================================================
--- qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java (original)
+++ qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java Sun Mar 1 21:33:36 2015
@@ -794,14 +794,6 @@ public class AMQProtocolHandler implemen
return _writtenBytes;
}
- public void failover(String host, int port)
- {
- _failoverHandler.setHost(host);
- _failoverHandler.setPort(port);
- // see javadoc for FailoverHandler to see rationale for separate thread
- startFailoverThread();
- }
-
public void blockUntilNotFailingOver() throws InterruptedException
{
synchronized(_failoverLatchChange)
Modified: qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java?rev=1663170&r1=1663169&r2=1663170&view=diff
==============================================================================
--- qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java (original)
+++ qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java Sun Mar 1 21:33:36 2015
@@ -387,11 +387,6 @@ public class AMQProtocolSession implemen
return _protocolHandler.getSender();
}
- public void failover(String host, int port)
- {
- _protocolHandler.failover(host, port);
- }
-
protected AMQShortString generateQueueName()
{
int id;
Modified: qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/state/AMQStateManager.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/state/AMQStateManager.java?rev=1663170&r1=1663169&r2=1663170&view=diff
==============================================================================
--- qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/state/AMQStateManager.java (original)
+++ qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/state/AMQStateManager.java Sun Mar 1 21:33:36 2015
@@ -20,6 +20,10 @@
*/
package org.apache.qpid.client.state;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CopyOnWriteArrayList;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -29,10 +33,6 @@ import org.apache.qpid.framing.AMQMethod
import org.apache.qpid.protocol.AMQMethodEvent;
import org.apache.qpid.protocol.AMQMethodListener;
-import java.util.List;
-import java.util.Set;
-import java.util.concurrent.CopyOnWriteArrayList;
-
/**
* The state manager is responsible for managing the state of the protocol session.
* <p>
Modified: qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jms/BrokerDetails.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jms/BrokerDetails.java?rev=1663170&r1=1663169&r2=1663170&view=diff
==============================================================================
--- qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jms/BrokerDetails.java (original)
+++ qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/jms/BrokerDetails.java Sun Mar 1 21:33:36 2015
@@ -98,7 +98,7 @@ public interface BrokerDetails
/**
* Sets the properties associated with this connection
*
- * @param props the new p[roperties.
+ * @param props the new properties.
*/
public void setProperties(Map<String,String> props);
Modified: qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/ClientDelegate.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/ClientDelegate.java?rev=1663170&r1=1663169&r2=1663170&view=diff
==============================================================================
--- qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/ClientDelegate.java (original)
+++ qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/ClientDelegate.java Sun Mar 1 21:33:36 2015
@@ -183,7 +183,15 @@ public class ClientDelegate extends Conn
@Override
public void connectionRedirect(Connection conn, ConnectionRedirect redir)
{
- throw new UnsupportedOperationException();
+ conn.setRedirecting(true);
+ conn.getSender().close();
+ for(ConnectionListener listener : conn.getListeners())
+ {
+ if(listener.redirect(redir.getHost(), redir.getKnownHosts()))
+ {
+ break;
+ }
+ }
}
@Override
Modified: qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java?rev=1663170&r1=1663169&r2=1663170&view=diff
==============================================================================
--- qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java (original)
+++ qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java Sun Mar 1 21:33:36 2015
@@ -81,6 +81,7 @@ public class Connection extends Connecti
private NetworkConnection _networkConnection;
private FrameSizeObserver _frameSizeObserver;
private boolean _messageCompressionSupported;
+ private final AtomicBoolean _redirecting = new AtomicBoolean();
public enum State { NEW, CLOSED, OPENING, OPEN, CLOSING, CLOSE_RCVD, RESUMING }
@@ -92,6 +93,12 @@ public class Connection extends Connecti
log.error(exception, "connection exception");
}
public void closed(Connection conn) {}
+
+ @Override
+ public boolean redirect(final String host, final List<Object> knownHosts)
+ {
+ return false;
+ }
}
public static interface SessionFactory
@@ -151,6 +158,11 @@ public class Connection extends Connecti
listeners.add(listener);
}
+ public List<ConnectionListener> getListeners()
+ {
+ return Collections.unmodifiableList(listeners);
+ }
+
public Sender<ProtocolEvent> getSender()
{
return sender;
@@ -226,6 +238,7 @@ public class Connection extends Connecti
synchronized (lock)
{
conSettings = settings;
+ _redirecting.set(false);
state = OPENING;
userID = settings.getUsername();
connectionLost.set(false);
@@ -259,7 +272,7 @@ public class Connection extends Connecti
send(new ProtocolHeader(1, 0, 10));
Waiter w = new Waiter(lock, timeout);
- while (w.hasTime() && state == OPENING && error == null)
+ while (w.hasTime() && ((state == OPENING && error == null) || isRedirecting()))
{
w.await();
}
@@ -863,4 +876,15 @@ public class Connection extends Connecti
{
return _messageCompressionSupported;
}
+
+ public boolean isRedirecting()
+ {
+ return _redirecting.get();
+ }
+
+ public void setRedirecting(final boolean redirecting)
+ {
+ _redirecting.set(redirecting);
+ }
+
}
Modified: qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/ConnectionListener.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/ConnectionListener.java?rev=1663170&r1=1663169&r2=1663170&view=diff
==============================================================================
--- qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/ConnectionListener.java (original)
+++ qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/ConnectionListener.java Sun Mar 1 21:33:36 2015
@@ -21,6 +21,8 @@
package org.apache.qpid.transport;
+import java.util.List;
+
/**
* ConnectionListener
*
@@ -35,4 +37,5 @@ public interface ConnectionListener
void closed(Connection connection);
+ boolean redirect(String host, List<Object> knownHosts);
}
Modified: qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/sasl/SASLEncryptor.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/sasl/SASLEncryptor.java?rev=1663170&r1=1663169&r2=1663170&view=diff
==============================================================================
--- qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/sasl/SASLEncryptor.java (original)
+++ qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/sasl/SASLEncryptor.java Sun Mar 1 21:33:36 2015
@@ -21,6 +21,8 @@
package org.apache.qpid.transport.network.security.sasl;
+import java.util.List;
+
import javax.security.sasl.Sasl;
import javax.security.sasl.SaslClient;
@@ -56,7 +58,13 @@ public abstract class SASLEncryptor impl
}
}
}
-
+
+ @Override
+ public boolean redirect(final String host, final List<Object> knownHosts)
+ {
+ return false;
+ }
+
public void exception(Connection conn, ConnectionException exception){}
public void closed(Connection conn) {}
Modified: qpid/trunk/qpid/java/common/src/test/java/org/apache/qpid/transport/ConnectionTest.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/test/java/org/apache/qpid/transport/ConnectionTest.java?rev=1663170&r1=1663169&r2=1663170&view=diff
==============================================================================
--- qpid/trunk/qpid/java/common/src/test/java/org/apache/qpid/transport/ConnectionTest.java (original)
+++ qpid/trunk/qpid/java/common/src/test/java/org/apache/qpid/transport/ConnectionTest.java Sun Mar 1 21:33:36 2015
@@ -20,12 +20,6 @@
*/
package org.apache.qpid.transport;
-import org.apache.log4j.Logger;
-import org.apache.qpid.test.utils.QpidTestCase;
-import org.apache.qpid.transport.network.ConnectionBinding;
-import org.apache.qpid.transport.network.io.IoAcceptor;
-import org.apache.qpid.transport.util.Waiter;
-
import static org.apache.qpid.transport.Option.EXPECTED;
import static org.apache.qpid.transport.Option.NONE;
import static org.apache.qpid.transport.Option.SYNC;
@@ -37,6 +31,13 @@ import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
+import org.apache.log4j.Logger;
+
+import org.apache.qpid.test.utils.QpidTestCase;
+import org.apache.qpid.transport.network.ConnectionBinding;
+import org.apache.qpid.transport.network.io.IoAcceptor;
+import org.apache.qpid.transport.util.Waiter;
+
/**
* ConnectionTest
*/
@@ -171,6 +172,12 @@ public class ConnectionTest extends Qpid
closed.countDown();
}
}
+
+ @Override
+ public boolean redirect(final String host, final List<Object> knownHosts)
+ {
+ return false;
+ }
});
conn.connect("localhost", port, null, "guest", "guest", false, null);
return conn;
@@ -437,6 +444,12 @@ public class ConnectionTest extends Qpid
conn.connect("localhost", port, null, "guest", "guest", false, null);
conn.resume();
}
+
+ @Override
+ public boolean redirect(final String host, final List<Object> knownHosts)
+ {
+ return false;
+ }
}
class TestSessionListener implements SessionListener
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org