diff options
Diffstat (limited to 'qpid/java/broker/src/main')
9 files changed, 133 insertions, 69 deletions
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/federation/Bridge.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/federation/Bridge.java index fbc5387daf..2dff45c326 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/federation/Bridge.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/federation/Bridge.java @@ -38,6 +38,7 @@ import org.apache.qpid.server.queue.BaseQueue; import org.apache.qpid.server.queue.QueueRegistry; import org.apache.qpid.server.store.MessageStore; import org.apache.qpid.server.store.StoredMessage; +import org.apache.qpid.server.subscription.SubscriptionFactoryImpl; import org.apache.qpid.server.subscription.Subscription_0_10; import org.apache.qpid.server.transport.ServerSession; import org.apache.qpid.server.txn.AutoCommitTransaction; @@ -696,7 +697,7 @@ public class Bridge implements BridgeConfig //TODO Handle the passing of non-null Filters and Arguments here - Subscription_0_10 sub = new Subscription_0_10((ServerSession)session, + Subscription_0_10 sub = SubscriptionFactoryImpl.INSTANCE.createSubscription((ServerSession)session, _destination, MessageAcceptMode.NONE, MessageAcquireMode.PRE_ACQUIRED, @@ -768,7 +769,7 @@ public class Bridge implements BridgeConfig //TODO Handle the passing of non-null Filters and Arguments here - Subscription_0_10 sub = new Subscription_0_10((ServerSession)session, + Subscription_0_10 sub = SubscriptionFactoryImpl.INSTANCE.createSubscription((ServerSession)session, _destination, MessageAcceptMode.NONE, MessageAcquireMode.PRE_ACQUIRED, diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicGetMethodHandler.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicGetMethodHandler.java index 790027f293..14ce85530e 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicGetMethodHandler.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/handler/BasicGetMethodHandler.java @@ -162,14 +162,7 @@ public class BasicGetMethodHandler implements StateAwareMethodListener<BasicGetB } else { - sub = new GetNoAckSubscription(channel, - session, - null, - null, - false, - singleMessageCredit, - getDeliveryMethod, - getRecordMethod); + sub = SubscriptionFactoryImpl.INSTANCE.createBasicGetNoAckSubscription(channel, session, null, null, false, singleMessageCredit, getDeliveryMethod, getRecordMethod); } queue.registerSubscription(sub,false); @@ -180,27 +173,5 @@ public class BasicGetMethodHandler implements StateAwareMethodListener<BasicGetB } - public static final class GetNoAckSubscription extends SubscriptionImpl.NoAckSubscription - { - public GetNoAckSubscription(AMQChannel channel, AMQProtocolSession protocolSession, - AMQShortString consumerTag, FieldTable filters, - boolean noLocal, FlowCreditManager creditManager, - ClientDeliveryMethod deliveryMethod, - RecordDeliveryMethod recordMethod) - throws AMQException - { - super(channel, protocolSession, consumerTag, filters, noLocal, creditManager, deliveryMethod, recordMethod); - } - public boolean isTransient() - { - return true; - } - - public boolean wouldSuspend(QueueEntry msg) - { - return !getCreditManager().useCreditForMessage(msg.getMessage()); - } - - } } diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionFactory.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionFactory.java index ce0362d73f..0fd7fdffe5 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionFactory.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionFactory.java @@ -20,13 +20,21 @@ */ package org.apache.qpid.server.subscription; +import java.util.Map; + import org.apache.qpid.AMQException; import org.apache.qpid.framing.AMQShortString; import org.apache.qpid.framing.FieldTable; import org.apache.qpid.server.protocol.AMQProtocolSession; +import org.apache.qpid.server.filter.FilterManager; import org.apache.qpid.server.flow.FlowCreditManager; +import org.apache.qpid.server.flow.FlowCreditManager_0_10; import org.apache.qpid.server.subscription.Subscription; +import org.apache.qpid.server.transport.ServerSession; import org.apache.qpid.server.AMQChannel; +import org.apache.qpid.transport.MessageAcceptMode; +import org.apache.qpid.transport.MessageAcquireMode; +import org.apache.qpid.transport.MessageFlowMode; /** * Allows the customisation of the creation of a subscription. This is typically done within an AMQQueue. This factory @@ -56,4 +64,23 @@ public interface SubscriptionFactory RecordDeliveryMethod recordMethod ) throws AMQException; + + + SubscriptionImpl.GetNoAckSubscription createBasicGetNoAckSubscription(AMQChannel channel, + AMQProtocolSession session, + AMQShortString consumerTag, + FieldTable filters, + boolean noLocal, + FlowCreditManager creditManager, + ClientDeliveryMethod deliveryMethod, + RecordDeliveryMethod recordMethod) throws AMQException; + + Subscription_0_10 createSubscription(final ServerSession session, + final String destination, + final MessageAcceptMode acceptMode, + final MessageAcquireMode acquireMode, + final MessageFlowMode flowMode, + final FlowCreditManager_0_10 creditManager, + final FilterManager filterManager, + final Map<String,Object> arguments); } diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionFactoryImpl.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionFactoryImpl.java index 1bba2529c6..1622d63648 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionFactoryImpl.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionFactoryImpl.java @@ -20,17 +20,28 @@ */ package org.apache.qpid.server.subscription; +import java.util.Map; +import java.util.concurrent.atomic.AtomicLong; + import org.apache.qpid.AMQException; import org.apache.qpid.common.AMQPFilterTypes; import org.apache.qpid.framing.AMQShortString; import org.apache.qpid.framing.FieldTable; import org.apache.qpid.protocol.AMQConstant; import org.apache.qpid.server.AMQChannel; +import org.apache.qpid.server.filter.FilterManager; import org.apache.qpid.server.flow.FlowCreditManager; +import org.apache.qpid.server.flow.FlowCreditManager_0_10; import org.apache.qpid.server.protocol.AMQProtocolSession; +import org.apache.qpid.server.transport.ServerSession; +import org.apache.qpid.transport.MessageAcceptMode; +import org.apache.qpid.transport.MessageAcquireMode; +import org.apache.qpid.transport.MessageFlowMode; public class SubscriptionFactoryImpl implements SubscriptionFactory { + private static final AtomicLong SUB_ID_GENERATOR = new AtomicLong(0); + public Subscription createSubscription(int channelId, AMQProtocolSession protocolSession, AMQShortString consumerTag, boolean acks, FieldTable filters, boolean noLocal, FlowCreditManager creditManager) throws AMQException @@ -78,18 +89,47 @@ public class SubscriptionFactoryImpl implements SubscriptionFactory if(isBrowser) { - return new SubscriptionImpl.BrowserSubscription(channel, protocolSession, consumerTag, filters, noLocal, creditManager, clientMethod, recordMethod); + return new SubscriptionImpl.BrowserSubscription(channel, protocolSession, consumerTag, filters, noLocal, creditManager, clientMethod, recordMethod, getNextSubscriptionId()); } else if(acks) { - return new SubscriptionImpl.AckSubscription(channel, protocolSession, consumerTag, filters, noLocal, creditManager, clientMethod, recordMethod); + return new SubscriptionImpl.AckSubscription(channel, protocolSession, consumerTag, filters, noLocal, creditManager, clientMethod, recordMethod, getNextSubscriptionId()); } else { - return new SubscriptionImpl.NoAckSubscription(channel, protocolSession, consumerTag, filters, noLocal, creditManager, clientMethod, recordMethod); + return new SubscriptionImpl.NoAckSubscription(channel, protocolSession, consumerTag, filters, noLocal, creditManager, clientMethod, recordMethod, getNextSubscriptionId()); } } + public SubscriptionImpl.GetNoAckSubscription createBasicGetNoAckSubscription(final AMQChannel channel, + final AMQProtocolSession session, + final AMQShortString consumerTag, + final FieldTable filters, + final boolean noLocal, + final FlowCreditManager creditManager, + final ClientDeliveryMethod deliveryMethod, + final RecordDeliveryMethod recordMethod) throws AMQException + { + return new SubscriptionImpl.GetNoAckSubscription(channel, session, null, null, false, creditManager, deliveryMethod, recordMethod, getNextSubscriptionId()); + } + + public Subscription_0_10 createSubscription(final ServerSession session, + final String destination, + final MessageAcceptMode acceptMode, + final MessageAcquireMode acquireMode, + final MessageFlowMode flowMode, + final FlowCreditManager_0_10 creditManager, + final FilterManager filterManager, + final Map<String,Object> arguments) + { + return new Subscription_0_10(session, destination, acceptMode, acquireMode, + flowMode, creditManager, filterManager, arguments, getNextSubscriptionId()); + } public static final SubscriptionFactoryImpl INSTANCE = new SubscriptionFactoryImpl(); + + private static long getNextSubscriptionId() + { + return SUB_ID_GENERATOR.getAndIncrement(); + } } diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionImpl.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionImpl.java index d8f44c9f7f..b791596054 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionImpl.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/SubscriptionImpl.java @@ -88,9 +88,7 @@ public abstract class SubscriptionImpl implements Subscription, FlowCreditManage private final Lock _stateChangeLock; - private static final AtomicLong idGenerator = new AtomicLong(0); - // Create a simple ID that increments for ever new Subscription - private final long _subscriptionID = idGenerator.getAndIncrement(); + private final long _subscriptionID; private LogSubject _logSubject; private LogActor _logActor; private UUID _id; @@ -104,10 +102,11 @@ public abstract class SubscriptionImpl implements Subscription, FlowCreditManage AMQShortString consumerTag, FieldTable filters, boolean noLocal, FlowCreditManager creditManager, ClientDeliveryMethod deliveryMethod, - RecordDeliveryMethod recordMethod) + RecordDeliveryMethod recordMethod, + long subscriptionID) throws AMQException { - super(channel, protocolSession, consumerTag, filters, noLocal, creditManager, deliveryMethod, recordMethod); + super(channel, protocolSession, consumerTag, filters, noLocal, creditManager, deliveryMethod, recordMethod, subscriptionID); } @@ -151,10 +150,11 @@ public abstract class SubscriptionImpl implements Subscription, FlowCreditManage AMQShortString consumerTag, FieldTable filters, boolean noLocal, FlowCreditManager creditManager, ClientDeliveryMethod deliveryMethod, - RecordDeliveryMethod recordMethod) + RecordDeliveryMethod recordMethod, + long subscriptionID) throws AMQException { - super(channel, protocolSession, consumerTag, filters, noLocal, creditManager, deliveryMethod, recordMethod); + super(channel, protocolSession, consumerTag, filters, noLocal, creditManager, deliveryMethod, recordMethod, subscriptionID); } @@ -211,16 +211,45 @@ public abstract class SubscriptionImpl implements Subscription, FlowCreditManage } + /** + * NoAck Subscription for use with BasicGet method. + */ + public static final class GetNoAckSubscription extends SubscriptionImpl.NoAckSubscription + { + public GetNoAckSubscription(AMQChannel channel, AMQProtocolSession protocolSession, + AMQShortString consumerTag, FieldTable filters, + boolean noLocal, FlowCreditManager creditManager, + ClientDeliveryMethod deliveryMethod, + RecordDeliveryMethod recordMethod, + long subscriptionID) + throws AMQException + { + super(channel, protocolSession, consumerTag, filters, noLocal, creditManager, deliveryMethod, recordMethod, subscriptionID); + } + + public boolean isTransient() + { + return true; + } + + public boolean wouldSuspend(QueueEntry msg) + { + return !getCreditManager().useCreditForMessage(msg.getMessage()); + } + + } + static final class AckSubscription extends SubscriptionImpl { public AckSubscription(AMQChannel channel, AMQProtocolSession protocolSession, AMQShortString consumerTag, FieldTable filters, boolean noLocal, FlowCreditManager creditManager, ClientDeliveryMethod deliveryMethod, - RecordDeliveryMethod recordMethod) + RecordDeliveryMethod recordMethod, + long subscriptionID) throws AMQException { - super(channel, protocolSession, consumerTag, filters, noLocal, creditManager, deliveryMethod, recordMethod); + super(channel, protocolSession, consumerTag, filters, noLocal, creditManager, deliveryMethod, recordMethod, subscriptionID); } @@ -296,10 +325,11 @@ public abstract class SubscriptionImpl implements Subscription, FlowCreditManage AMQShortString consumerTag, FieldTable arguments, boolean noLocal, FlowCreditManager creditManager, ClientDeliveryMethod deliveryMethod, - RecordDeliveryMethod recordMethod) + RecordDeliveryMethod recordMethod, + long subscriptionID) throws AMQException { - + _subscriptionID = subscriptionID; _channel = channel; _consumerTag = consumerTag; diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/Subscription_0_10.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/Subscription_0_10.java index 68e47fd86a..309879c17c 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/Subscription_0_10.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/subscription/Subscription_0_10.java @@ -40,7 +40,6 @@ import org.apache.qpid.server.logging.actors.GenericActor; import org.apache.qpid.server.logging.messages.SubscriptionMessages; import org.apache.qpid.server.logging.LogActor; import org.apache.qpid.server.logging.LogSubject; -import org.apache.qpid.server.logging.actors.SubscriptionActor; import org.apache.qpid.server.message.ServerMessage; import org.apache.qpid.server.message.MessageTransferMessage; import org.apache.qpid.server.message.AMQMessage; @@ -80,10 +79,7 @@ import java.nio.ByteBuffer; public class Subscription_0_10 implements Subscription, FlowCreditManager.FlowCreditManagerListener, SubscriptionConfig, LogSubject { - - private static final AtomicLong idGenerator = new AtomicLong(0); - // Create a simple ID that increments for ever new Subscription - private final long _subscriptionID = idGenerator.getAndIncrement(); + private final long _subscriptionID; private final QueueEntry.SubscriptionAcquiredState _owningState = new QueueEntry.SubscriptionAcquiredState(this); private final QueueEntry.SubscriptionAssignedState _assignedState = new QueueEntry.SubscriptionAssignedState(this); @@ -114,7 +110,6 @@ public class Subscription_0_10 implements Subscription, FlowCreditManager.FlowCr private MessageFlowMode _flowMode; private final ServerSession _session; private AtomicBoolean _stopped = new AtomicBoolean(true); - private ConcurrentHashMap<Integer, QueueEntry> _sentMap = new ConcurrentHashMap<Integer, QueueEntry>(); private static final Struct[] EMPTY_STRUCT_ARRAY = new Struct[0]; private LogActor _logActor; @@ -131,8 +126,9 @@ public class Subscription_0_10 implements Subscription, FlowCreditManager.FlowCr MessageAcquireMode acquireMode, MessageFlowMode flowMode, FlowCreditManager_0_10 creditManager, - FilterManager filters,Map<String, Object> arguments) + FilterManager filters,Map<String, Object> arguments, long subscriptionId) { + _subscriptionID = subscriptionId; _session = session; _destination = destination; _acceptMode = acceptMode; diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java index 19c11080b8..f65cad23e7 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerConnectionDelegate.java @@ -74,7 +74,7 @@ public class ServerConnectionDelegate extends ServerDelegate public ServerSession getSession(Connection conn, SessionAttach atc) { - SessionDelegate serverSessionDelegate = new ServerSessionDelegate(_appRegistry); + SessionDelegate serverSessionDelegate = new ServerSessionDelegate(); ServerSession ssn = new ServerSession(conn, serverSessionDelegate, new Binary(atc.getName()), 0); diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java index 0f4147ebb4..65790e2e6f 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSession.java @@ -129,16 +129,6 @@ public class ServerSession extends Session implements AuthorizationHolder, Sessi this(connection, delegate, name, expiry, ((ServerConnection)connection).getConfig()); } - protected void setState(State state) - { - super.setState(state); - - if (state == State.OPEN) - { - _actor.message(ChannelMessages.CREATE()); - } - } - public ServerSession(Connection connection, SessionDelegate delegate, Binary name, long expiry, ConnectionConfig connConfig) { super(connection, delegate, name, expiry); @@ -150,6 +140,16 @@ public class ServerSession extends Session implements AuthorizationHolder, Sessi getConfigStore().addConfiguredObject(this); } + protected void setState(State state) + { + super.setState(state); + + if (state == State.OPEN) + { + _actor.message(ChannelMessages.CREATE()); + } + } + private ConfigStore getConfigStore() { return getConnectionConfig().getConfigStore(); diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSessionDelegate.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSessionDelegate.java index f316d60c6a..f89e86e433 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSessionDelegate.java +++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSessionDelegate.java @@ -47,11 +47,11 @@ import org.apache.qpid.server.queue.AMQQueueFactory; import org.apache.qpid.server.queue.BaseQueue; import org.apache.qpid.server.queue.QueueRegistry; import org.apache.qpid.server.registry.ApplicationRegistry; -import org.apache.qpid.server.registry.IApplicationRegistry; import org.apache.qpid.server.security.SecurityManager; import org.apache.qpid.server.store.DurableConfigurationStore; import org.apache.qpid.server.store.MessageStore; import org.apache.qpid.server.store.StoredMessage; +import org.apache.qpid.server.subscription.SubscriptionFactoryImpl; import org.apache.qpid.server.subscription.Subscription_0_10; import org.apache.qpid.server.virtualhost.VirtualHost; import org.apache.qpid.transport.Acquired; @@ -98,11 +98,10 @@ import org.apache.qpid.transport.TxSelect; public class ServerSessionDelegate extends SessionDelegate { private static final Logger LOGGER = Logger.getLogger(ServerSessionDelegate.class); - private final IApplicationRegistry _appRegistry; - public ServerSessionDelegate(IApplicationRegistry appRegistry) + public ServerSessionDelegate() { - _appRegistry = appRegistry; + } @Override @@ -254,7 +253,7 @@ public class ServerSessionDelegate extends SessionDelegate return; } - Subscription_0_10 sub = new Subscription_0_10((ServerSession)session, + Subscription_0_10 sub = SubscriptionFactoryImpl.INSTANCE.createSubscription((ServerSession)session, destination, method.getAcceptMode(), method.getAcquireMode(), |