You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by kp...@apache.org on 2012/05/04 17:40:13 UTC
svn commit: r1334037 [19/24] - in /qpid/branches/asyncstore: ./ bin/ cc/
cpp/ cpp/bindings/ cpp/bindings/qmf/ruby/ cpp/bindings/qmf/tests/
cpp/bindings/qpid/ cpp/bindings/qpid/dotnet/ cpp/bindings/qpid/python/
cpp/bindings/qpid/ruby/ cpp/bindings/qpid/...
Modified: qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/AMQSession.java?rev=1334037&r1=1334036&r2=1334037&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/AMQSession.java (original)
+++ qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/AMQSession.java Fri May 4 15:39:19 2012
@@ -120,18 +120,6 @@ public abstract class AMQSession<C exten
public static final long DEFAULT_FLOW_CONTROL_WAIT_FAILURE = 120000L;
/**
- * The default value for immediate flag used by producers created by this session is false. That is, a consumer does
- * not need to be attached to a queue.
- */
- private final boolean _defaultImmediateValue = Boolean.parseBoolean(System.getProperty("qpid.default_immediate", "false"));
-
- /**
- * The default value for mandatory flag used by producers created by this session is true. That is, server will not
- * silently drop messages where no queue is connected to the exchange for the message.
- */
- private final boolean _defaultMandatoryValue = Boolean.parseBoolean(System.getProperty("qpid.default_mandatory", "true"));
-
- /**
* The period to wait while flow controlled before sending a log message confirming that the session is still
* waiting on flow control being revoked
*/
@@ -542,9 +530,9 @@ public abstract class AMQSession<C exten
}
// Add creation logging to tie in with the existing close logging
- if (_logger.isInfoEnabled())
+ if (_logger.isDebugEnabled())
{
- _logger.info("Created session:" + this);
+ _logger.debug("Created session:" + this);
}
}
@@ -733,9 +721,9 @@ public abstract class AMQSession<C exten
private void close(long timeout, boolean sendClose) throws JMSException
{
- if (_logger.isInfoEnabled())
+ if (_logger.isDebugEnabled())
{
- _logger.info("Closing session: " + this);
+ _logger.debug("Closing session: " + this);
}
// Ensure we only try and close an open session.
@@ -904,11 +892,11 @@ public abstract class AMQSession<C exten
// Flush any pending messages for this consumerTag
if (_dispatcher != null)
{
- _logger.info("Dispatcher is not null");
+ _logger.debug("Dispatcher is not null");
}
else
{
- _logger.info("Dispatcher is null so created stopped dispatcher");
+ _logger.debug("Dispatcher is null so created stopped dispatcher");
startDispatcherIfNecessary(true);
}
@@ -926,9 +914,9 @@ public abstract class AMQSession<C exten
// There is a small window where the message is between the two queues in the dispatcher.
if (consumer.isClosed())
{
- if (_logger.isInfoEnabled())
+ if (_logger.isDebugEnabled())
{
- _logger.info("Closing consumer:" + consumer.debugIdentity());
+ _logger.debug("Closing consumer:" + consumer.debugIdentity());
}
deregisterConsumer(consumer);
@@ -1101,6 +1089,10 @@ public abstract class AMQSession<C exten
// possible to determine when querying the broker whether there are no arguments or just a non-matching selector
// argument, as specifying null for the arguments when querying means they should not be checked at all
args.put(AMQPFilterTypes.JMS_SELECTOR.getValue().toString(), messageSelector == null ? "" : messageSelector);
+ if(noLocal)
+ {
+ args.put(AMQPFilterTypes.NO_LOCAL.getValue().toString(), true);
+ }
// if the queue is bound to the exchange but NOT for this topic and selector, then the JMS spec
// says we must trash the subscription.
@@ -1198,12 +1190,12 @@ public abstract class AMQSession<C exten
public P createProducer(Destination destination) throws JMSException
{
- return createProducerImpl(destination, _defaultMandatoryValue, _defaultImmediateValue);
+ return createProducerImpl(destination, null, null);
}
public P createProducer(Destination destination, boolean immediate) throws JMSException
{
- return createProducerImpl(destination, _defaultMandatoryValue, immediate);
+ return createProducerImpl(destination, null, immediate);
}
public P createProducer(Destination destination, boolean mandatory, boolean immediate)
@@ -1692,7 +1684,7 @@ public abstract class AMQSession<C exten
{
AMQProtocolHandler protocolHandler = getProtocolHandler();
declareExchange(amqd, protocolHandler, false);
- AMQShortString queueName = declareQueue(amqd, protocolHandler, false);
+ AMQShortString queueName = declareQueue(amqd, false);
bindQueue(queueName, amqd.getRoutingKey(), new FieldTable(), amqd.getExchangeName(), amqd);
}
@@ -2378,9 +2370,9 @@ public abstract class AMQSession<C exten
_dispatcherThread.setDaemon(DEAMON_DISPATCHER_THREAD);
_dispatcher.setConnectionStopped(initiallyStopped);
_dispatcherThread.start();
- if (_dispatcherLogger.isInfoEnabled())
+ if (_dispatcherLogger.isDebugEnabled())
{
- _dispatcherLogger.info(_dispatcherThread.getName() + " created");
+ _dispatcherLogger.debug(_dispatcherThread.getName() + " created");
}
}
else
@@ -2613,7 +2605,7 @@ public abstract class AMQSession<C exten
public abstract void sendConsume(C consumer, AMQShortString queueName,
AMQProtocolHandler protocolHandler, boolean nowait, int tag) throws AMQException, FailoverException;
- private P createProducerImpl(final Destination destination, final boolean mandatory, final boolean immediate)
+ private P createProducerImpl(final Destination destination, final Boolean mandatory, final Boolean immediate)
throws JMSException
{
return new FailoverRetrySupport<P, JMSException>(
@@ -2642,8 +2634,8 @@ public abstract class AMQSession<C exten
}, _connection).execute();
}
- public abstract P createMessageProducer(final Destination destination, final boolean mandatory,
- final boolean immediate, final long producerId) throws JMSException;
+ public abstract P createMessageProducer(final Destination destination, final Boolean mandatory,
+ final Boolean immediate, final long producerId) throws JMSException;
private void declareExchange(AMQDestination amqd, AMQProtocolHandler protocolHandler, boolean nowait) throws AMQException
{
@@ -2726,6 +2718,12 @@ public abstract class AMQSession<C exten
public abstract void sendExchangeDeclare(final AMQShortString name, final AMQShortString type, final AMQProtocolHandler protocolHandler,
final boolean nowait) throws AMQException, FailoverException;
+
+ void declareQueuePassive(AMQDestination queue) throws AMQException
+ {
+ declareQueue(queue,false,false,true);
+ }
+
/**
* Declares a queue for a JMS destination.
*
@@ -2735,27 +2733,35 @@ public abstract class AMQSession<C exten
*
* <p/>Note that this operation automatically retries in the event of fail-over.
*
- * @param amqd The destination to declare as a queue.
- * @param protocolHandler The protocol handler to communicate through.
*
+ * @param amqd The destination to declare as a queue.
* @return The name of the decalred queue. This is useful where the broker is generating a queue name on behalf of
* the client.
*
+ *
+ *
* @throws AMQException If the queue cannot be declared for any reason.
* @todo Verify the destiation is valid or throw an exception.
* @todo Be aware of possible changes to parameter order as versions change.
*/
- protected AMQShortString declareQueue(final AMQDestination amqd, final AMQProtocolHandler protocolHandler,
+ protected AMQShortString declareQueue(final AMQDestination amqd,
final boolean noLocal) throws AMQException
{
- return declareQueue(amqd, protocolHandler, noLocal, false);
+ return declareQueue(amqd, noLocal, false);
}
- protected AMQShortString declareQueue(final AMQDestination amqd, final AMQProtocolHandler protocolHandler,
+ protected AMQShortString declareQueue(final AMQDestination amqd,
final boolean noLocal, final boolean nowait)
+ throws AMQException
+ {
+ return declareQueue(amqd, noLocal, nowait, false);
+ }
+
+ protected AMQShortString declareQueue(final AMQDestination amqd,
+ final boolean noLocal, final boolean nowait, final boolean passive)
throws AMQException
{
- /*return new FailoverRetrySupport<AMQShortString, AMQException>(*/
+ final AMQProtocolHandler protocolHandler = getProtocolHandler();
return new FailoverNoopSupport<AMQShortString, AMQException>(
new FailoverProtectedOperation<AMQShortString, AMQException>()
{
@@ -2767,7 +2773,7 @@ public abstract class AMQSession<C exten
amqd.setQueueName(protocolHandler.generateQueueName());
}
- sendQueueDeclare(amqd, protocolHandler, nowait);
+ sendQueueDeclare(amqd, protocolHandler, nowait, passive);
return amqd.getAMQQueueName();
}
@@ -2775,7 +2781,7 @@ public abstract class AMQSession<C exten
}
public abstract void sendQueueDeclare(final AMQDestination amqd, final AMQProtocolHandler protocolHandler,
- final boolean nowait) throws AMQException, FailoverException;
+ final boolean nowait, boolean passive) throws AMQException, FailoverException;
/**
* Undeclares the specified queue.
@@ -2916,7 +2922,7 @@ public abstract class AMQSession<C exten
if (_delareQueues || amqd.isNameRequired())
{
- declareQueue(amqd, protocolHandler, consumer.isNoLocal(), nowait);
+ declareQueue(amqd, consumer.isNoLocal(), nowait);
}
bindQueue(amqd.getAMQQueueName(), amqd.getRoutingKey(), consumer.getArguments(), amqd.getExchangeName(), amqd, nowait);
}
@@ -2939,7 +2945,7 @@ public abstract class AMQSession<C exten
try
{
suspendChannel(true);
- _logger.info(
+ _logger.debug(
"Prefetching delayed existing messages will not flow until requested via receive*() or setML().");
}
catch (AMQException e)
@@ -2951,7 +2957,7 @@ public abstract class AMQSession<C exten
}
else
{
- _logger.info("Immediately prefetching existing messages to new consumer.");
+ _logger.debug("Immediately prefetching existing messages to new consumer.");
}
try
@@ -2983,18 +2989,18 @@ public abstract class AMQSession<C exten
private void rejectMessagesForConsumerTag(int consumerTag, boolean requeue, boolean rejectAllConsumers)
{
Iterator messages = _queue.iterator();
- if (_logger.isInfoEnabled())
+ if (_logger.isDebugEnabled())
{
- _logger.info("Rejecting messages from _queue for Consumer tag(" + consumerTag + ") (PDispatchQ) requeue:"
+ _logger.debug("Rejecting messages from _queue for Consumer tag(" + consumerTag + ") (PDispatchQ) requeue:"
+ requeue);
if (messages.hasNext())
{
- _logger.info("Checking all messages in _queue for Consumer tag(" + consumerTag + ")");
+ _logger.debug("Checking all messages in _queue for Consumer tag(" + consumerTag + ")");
}
else
{
- _logger.info("No messages in _queue to reject");
+ _logger.debug("No messages in _queue to reject");
}
}
while (messages.hasNext())
@@ -3037,7 +3043,7 @@ public abstract class AMQSession<C exten
private void resubscribeProducers() throws AMQException
{
ArrayList producers = new ArrayList(_producers.values());
- _logger.info(MessageFormat.format("Resubscribing producers = {0} producers.size={1}", producers, producers.size())); // FIXME: removeKey
+ _logger.debug(MessageFormat.format("Resubscribing producers = {0} producers.size={1}", producers, producers.size())); // FIXME: removeKey
for (Iterator it = producers.iterator(); it.hasNext();)
{
P producer = (P) it.next();
@@ -3127,7 +3133,10 @@ public abstract class AMQSession<C exten
public void setFlowControl(final boolean active)
{
_flowControl.setFlowControl(active);
- _logger.warn("Broker enforced flow control " + (active ? "no longer in effect" : "has been enforced"));
+ if (_logger.isInfoEnabled())
+ {
+ _logger.info("Broker enforced flow control " + (active ? "no longer in effect" : "has been enforced"));
+ }
}
public void checkFlowControl() throws InterruptedException, JMSException
@@ -3141,7 +3150,10 @@ public abstract class AMQSession<C exten
{
_flowControl.wait(_flowControlWaitPeriod);
- _logger.warn("Message send delayed by " + (System.currentTimeMillis() + _flowControlWaitFailure - expiryTime)/1000 + "s due to broker enforced flow control");
+ if (_logger.isInfoEnabled())
+ {
+ _logger.info("Message send delayed by " + (System.currentTimeMillis() + _flowControlWaitFailure - expiryTime)/1000 + "s due to broker enforced flow control");
+ }
}
if(!_flowControl.getFlowControl())
{
@@ -3200,28 +3212,15 @@ public abstract class AMQSession<C exten
public void rejectPending(C consumer)
{
- synchronized (_lock)
- {
- boolean stopped = _dispatcher.connectionStopped();
+ // Reject messages on pre-receive queue
+ consumer.rollbackPendingMessages();
- if (!stopped)
- {
- _dispatcher.setConnectionStopped(true);
- }
+ // Reject messages on pre-dispatch queue
+ rejectMessagesForConsumerTag(consumer.getConsumerTag(), true, false);
- // Reject messages on pre-receive queue
- consumer.rollbackPendingMessages();
+ // closeConsumer
+ consumer.markClosed();
- // Reject messages on pre-dispatch queue
- rejectMessagesForConsumerTag(consumer.getConsumerTag(), true, false);
- //Let the dispatcher deal with this when it gets to them.
-
- // closeConsumer
- consumer.markClosed();
-
- _dispatcher.setConnectionStopped(stopped);
-
- }
}
public void rollback()
@@ -3294,9 +3293,9 @@ public abstract class AMQSession<C exten
public void run()
{
- if (_dispatcherLogger.isInfoEnabled())
+ if (_dispatcherLogger.isDebugEnabled())
{
- _dispatcherLogger.info(_dispatcherThread.getName() + " started");
+ _dispatcherLogger.debug(_dispatcherThread.getName() + " started");
}
// Allow disptacher to start stopped
@@ -3318,7 +3317,7 @@ public abstract class AMQSession<C exten
try
{
Dispatchable disp;
- while (!_closed.get() && ((disp = (Dispatchable) _queue.take()) != null))
+ while (((disp = (Dispatchable) _queue.take()) != null) && !_closed.get())
{
disp.dispatch(AMQSession.this);
}
@@ -3328,9 +3327,9 @@ public abstract class AMQSession<C exten
// ignored as run will exit immediately
}
- if (_dispatcherLogger.isInfoEnabled())
+ if (_dispatcherLogger.isDebugEnabled())
{
- _dispatcherLogger.info(_dispatcherThread.getName() + " thread terminating for channel " + _channelId + ":" + AMQSession.this);
+ _dispatcherLogger.debug(_dispatcherThread.getName() + " thread terminating for channel " + _channelId + ":" + AMQSession.this);
}
}
@@ -3413,7 +3412,7 @@ public abstract class AMQSession<C exten
{
final C consumer = _consumers.get(message.getConsumerTag());
- if ((consumer == null) || consumer.isClosed())
+ if ((consumer == null) || consumer.isClosed() || consumer.isClosing())
{
if (_dispatcherLogger.isInfoEnabled())
{
Modified: qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java?rev=1334037&r1=1334036&r2=1334037&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java (original)
+++ qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java Fri May 4 15:39:19 2012
@@ -17,9 +17,20 @@
*/
package org.apache.qpid.client;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
+import java.lang.ref.WeakReference;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Timer;
+import java.util.TimerTask;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import javax.jms.Destination;
+import javax.jms.JMSException;
import org.apache.qpid.AMQException;
import org.apache.qpid.client.AMQDestination.AddressOption;
import org.apache.qpid.client.AMQDestination.Binding;
@@ -27,7 +38,6 @@ import org.apache.qpid.client.AMQDestina
import org.apache.qpid.client.failover.FailoverException;
import org.apache.qpid.client.failover.FailoverNoopSupport;
import org.apache.qpid.client.failover.FailoverProtectedOperation;
-import org.apache.qpid.client.filter.MessageFilter;
import org.apache.qpid.client.message.AMQMessageDelegateFactory;
import org.apache.qpid.client.message.FieldTableSupport;
import org.apache.qpid.client.message.MessageFactoryRegistry;
@@ -42,28 +52,14 @@ import org.apache.qpid.framing.AMQShortS
import org.apache.qpid.framing.FieldTable;
import org.apache.qpid.protocol.AMQConstant;
import org.apache.qpid.transport.*;
-import org.apache.qpid.util.Serial;
-import org.apache.qpid.util.Strings;
-
import static org.apache.qpid.transport.Option.BATCH;
import static org.apache.qpid.transport.Option.NONE;
import static org.apache.qpid.transport.Option.SYNC;
import static org.apache.qpid.transport.Option.UNRELIABLE;
-
-import javax.jms.Destination;
-import javax.jms.JMSException;
-import java.lang.ref.WeakReference;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.Date;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.Timer;
-import java.util.TimerTask;
-import java.util.UUID;
-import java.util.concurrent.ConcurrentLinkedQueue;
+import org.apache.qpid.util.Serial;
+import org.apache.qpid.util.Strings;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* This is a 0.10 Session
@@ -654,8 +650,8 @@ public class AMQSession_0_10 extends AMQ
/**
* Create an 0_10 message producer
*/
- public BasicMessageProducer_0_10 createMessageProducer(final Destination destination, final boolean mandatory,
- final boolean immediate, final long producerId) throws JMSException
+ public BasicMessageProducer_0_10 createMessageProducer(final Destination destination, final Boolean mandatory,
+ final Boolean immediate, final long producerId) throws JMSException
{
try
{
@@ -725,7 +721,7 @@ public class AMQSession_0_10 extends AMQ
* Declare a queue with the given queueName
*/
public void sendQueueDeclare(final AMQDestination amqd, final AMQProtocolHandler protocolHandler,
- final boolean nowait)
+ final boolean nowait, boolean passive)
throws AMQException, FailoverException
{
// do nothing this is only used by 0_8
@@ -735,7 +731,7 @@ public class AMQSession_0_10 extends AMQ
* Declare a queue with the given queueName
*/
public AMQShortString send0_10QueueDeclare(final AMQDestination amqd, final AMQProtocolHandler protocolHandler,
- final boolean noLocal, final boolean nowait)
+ final boolean noLocal, final boolean nowait, boolean passive)
throws AMQException
{
AMQShortString queueName;
@@ -761,7 +757,8 @@ public class AMQSession_0_10 extends AMQ
getQpidSession().queueDeclare(queueName.toString(), "" , arguments,
amqd.isAutoDelete() ? Option.AUTO_DELETE : Option.NONE,
amqd.isDurable() ? Option.DURABLE : Option.NONE,
- amqd.isExclusive() ? Option.EXCLUSIVE : Option.NONE);
+ amqd.isExclusive() ? Option.EXCLUSIVE : Option.NONE,
+ passive ? Option.PASSIVE : Option.NONE);
}
else
{
@@ -931,11 +928,12 @@ public class AMQSession_0_10 extends AMQ
return getCurrentException();
}
- protected AMQShortString declareQueue(final AMQDestination amqd, final AMQProtocolHandler protocolHandler,
- final boolean noLocal, final boolean nowait)
+ protected AMQShortString declareQueue(final AMQDestination amqd,
+ final boolean noLocal, final boolean nowait, final boolean passive)
throws AMQException
{
- /*return new FailoverRetrySupport<AMQShortString, AMQException>(*/
+ final AMQProtocolHandler protocolHandler = getProtocolHandler();
+
return new FailoverNoopSupport<AMQShortString, AMQException>(
new FailoverProtectedOperation<AMQShortString, AMQException>()
{
@@ -952,7 +950,7 @@ public class AMQSession_0_10 extends AMQ
amqd.setQueueName(new AMQShortString( binddingKey + "@"
+ amqd.getExchangeName().toString() + "_" + UUID.randomUUID()));
}
- return send0_10QueueDeclare(amqd, protocolHandler, noLocal, nowait);
+ return send0_10QueueDeclare(amqd, protocolHandler, noLocal, nowait, passive);
}
}, getAMQConnection()).execute();
}
@@ -1209,7 +1207,7 @@ public class AMQSession_0_10 extends AMQ
else if(createNode)
{
setLegacyFiledsForQueueType(dest);
- send0_10QueueDeclare(dest,null,noLocal,noWait);
+ send0_10QueueDeclare(dest,null,noLocal,noWait, false);
sendQueueBind(dest.getAMQQueueName(), dest.getRoutingKey(),
null,dest.getExchangeName(),dest, false);
break;
@@ -1315,7 +1313,7 @@ public class AMQSession_0_10 extends AMQ
}
node.setExclusive(true);
node.setAutoDelete(!node.isDurable());
- send0_10QueueDeclare(dest,null,noLocal,true);
+ send0_10QueueDeclare(dest,null,noLocal,true, false);
getQpidSession().exchangeBind(dest.getQueueName(),
dest.getAddressName(),
dest.getSubject(),
Modified: qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_8.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_8.java?rev=1334037&r1=1334036&r2=1334037&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_8.java (original)
+++ qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_8.java Fri May 4 15:39:19 2012
@@ -38,7 +38,6 @@ import org.apache.qpid.client.protocol.A
import org.apache.qpid.client.state.AMQState;
import org.apache.qpid.client.state.AMQStateManager;
import org.apache.qpid.client.state.listener.SpecificMethodFrameListener;
-import org.apache.qpid.client.filter.MessageFilter;
import org.apache.qpid.framing.*;
import org.apache.qpid.framing.amqp_0_9.MethodRegistry_0_9;
import org.apache.qpid.framing.amqp_0_91.MethodRegistry_0_91;
@@ -401,9 +400,17 @@ public class AMQSession_0_8 extends AMQS
}
public void sendQueueDeclare(final AMQDestination amqd, final AMQProtocolHandler protocolHandler,
- final boolean nowait) throws AMQException, FailoverException
+ final boolean nowait, boolean passive) throws AMQException, FailoverException
{
- QueueDeclareBody body = getMethodRegistry().createQueueDeclareBody(getTicket(),amqd.getAMQQueueName(),false,amqd.isDurable(),amqd.isExclusive(),amqd.isAutoDelete(),false,null);
+ QueueDeclareBody body =
+ getMethodRegistry().createQueueDeclareBody(getTicket(),
+ amqd.getAMQQueueName(),
+ passive,
+ amqd.isDurable(),
+ amqd.isExclusive(),
+ amqd.isAutoDelete(),
+ false,
+ null);
AMQFrame queueDeclare = body.generateFrame(getChannelId());
@@ -441,8 +448,8 @@ public class AMQSession_0_8 extends AMQS
}
- public BasicMessageProducer_0_8 createMessageProducer(final Destination destination, final boolean mandatory,
- final boolean immediate, long producerId) throws JMSException
+ public BasicMessageProducer_0_8 createMessageProducer(final Destination destination, final Boolean mandatory,
+ final Boolean immediate, long producerId) throws JMSException
{
try
{
Modified: qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java?rev=1334037&r1=1334036&r2=1334037&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java (original)
+++ qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer.java Fri May 4 15:39:19 2012
@@ -199,6 +199,10 @@ public abstract class BasicMessageConsum
// possible to determine when querying the broker whether there are no arguments or just a non-matching selector
// argument, as specifying null for the arguments when querying means they should not be checked at all
ft.put(AMQPFilterTypes.JMS_SELECTOR.getValue(), messageSelector == null ? "" : messageSelector);
+ if(noLocal)
+ {
+ ft.put(AMQPFilterTypes.NO_LOCAL.getValue(), noLocal);
+ }
_arguments = ft;
@@ -275,7 +279,10 @@ public abstract class BasicMessageConsum
throw new javax.jms.IllegalStateException("Attempt to alter listener while session is started.");
}
- _logger.debug("Message listener set for destination " + _destination);
+ if (_logger.isDebugEnabled())
+ {
+ _logger.debug("Message listener set for destination " + _destination);
+ }
if (messageListener != null)
{
@@ -553,9 +560,9 @@ public abstract class BasicMessageConsum
public void close(boolean sendClose) throws JMSException
{
- if (_logger.isInfoEnabled())
+ if (_logger.isDebugEnabled())
{
- _logger.info("Closing consumer:" + debugIdentity());
+ _logger.debug("Closing consumer:" + debugIdentity());
}
if (!setClosed())
@@ -586,7 +593,10 @@ public abstract class BasicMessageConsum
// no point otherwise as the connection will be gone
if (!_session.isClosed() || _session.isClosing())
{
- sendCancel();
+ synchronized(_session.getMessageDeliveryLock())
+ {
+ sendCancel();
+ }
cleanupQueue();
}
}
Modified: qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java?rev=1334037&r1=1334036&r2=1334037&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java (original)
+++ qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer.java Fri May 4 15:39:19 2012
@@ -20,18 +20,8 @@
*/
package org.apache.qpid.client;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.qpid.AMQException;
-import org.apache.qpid.client.message.AbstractJMSMessage;
-import org.apache.qpid.client.message.MessageConverter;
-import org.apache.qpid.client.protocol.AMQProtocolHandler;
-import org.apache.qpid.framing.ContentBody;
-import org.apache.qpid.transport.TransportException;
-import org.apache.qpid.util.UUIDGen;
-import org.apache.qpid.util.UUIDs;
-
+import java.io.UnsupportedEncodingException;
+import java.util.UUID;
import javax.jms.BytesMessage;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
@@ -42,77 +32,22 @@ import javax.jms.Message;
import javax.jms.ObjectMessage;
import javax.jms.StreamMessage;
import javax.jms.TextMessage;
-import java.io.UnsupportedEncodingException;
-import java.util.UUID;
+import javax.jms.Topic;
+import org.apache.qpid.AMQException;
+import org.apache.qpid.client.message.AbstractJMSMessage;
+import org.apache.qpid.client.message.MessageConverter;
+import org.apache.qpid.client.protocol.AMQProtocolHandler;
+import org.apache.qpid.transport.TransportException;
+import org.apache.qpid.util.UUIDGen;
+import org.apache.qpid.util.UUIDs;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public abstract class BasicMessageProducer extends Closeable implements org.apache.qpid.jms.MessageProducer
{
- /**
- * If true, messages will not get a timestamp.
- */
- protected boolean isDisableTimestamps()
- {
- return _disableTimestamps;
- }
-
- protected void setDisableTimestamps(boolean disableTimestamps)
- {
- _disableTimestamps = disableTimestamps;
- }
-
- protected void setDestination(AMQDestination destination)
- {
- _destination = destination;
- }
-
- protected AMQProtocolHandler getProtocolHandler()
- {
- return _protocolHandler;
- }
-
- protected void setProtocolHandler(AMQProtocolHandler protocolHandler)
- {
- _protocolHandler = protocolHandler;
- }
-
- protected int getChannelId()
- {
- return _channelId;
- }
-
- protected void setChannelId(int channelId)
- {
- _channelId = channelId;
- }
-
- protected void setSession(AMQSession session)
- {
- _session = session;
- }
-
- protected String getUserID()
- {
- return _userID;
- }
-
- protected void setUserID(String userID)
- {
- _userID = userID;
- }
-
- protected PublishMode getPublishMode()
- {
- return publishMode;
- }
-
- protected void setPublishMode(PublishMode publishMode)
- {
- this.publishMode = publishMode;
- }
-
enum PublishMode { ASYNC_PUBLISH_ALL, SYNC_PUBLISH_PERSISTENT, SYNC_PUBLISH_ALL };
- private final Logger _logger = LoggerFactory.getLogger(getClass());
+ private final Logger _logger ;
private AMQConnection _connection;
@@ -166,7 +101,7 @@ public abstract class BasicMessageProduc
private final boolean _immediate;
- private final boolean _mandatory;
+ private final Boolean _mandatory;
private boolean _disableMessageId;
@@ -174,14 +109,37 @@ public abstract class BasicMessageProduc
private String _userID; // ref user id used in the connection.
- private static final ContentBody[] NO_CONTENT_BODIES = new ContentBody[0];
+
+ /**
+ * The default value for immediate flag used this producer is false. That is, a consumer does
+ * not need to be attached to a queue.
+ */
+ private final boolean _defaultImmediateValue = Boolean.parseBoolean(System.getProperty("qpid.default_immediate", "false"));
+
+ /**
+ * The default value for mandatory flag used by this producer is true. That is, server will not
+ * silently drop messages where no queue is connected to the exchange for the message.
+ */
+ private final boolean _defaultMandatoryValue = Boolean.parseBoolean(System.getProperty("qpid.default_mandatory", "true"));
+
+ /**
+ * The default value for mandatory flag used by this producer when publishing to a Topic is false. That is, server
+ * will silently drop messages where no queue is connected to the exchange for the message.
+ */
+ private final boolean _defaultMandatoryTopicValue =
+ Boolean.parseBoolean(System.getProperty("qpid.default_mandatory_topic",
+ System.getProperties().containsKey("qpid.default_mandatory")
+ ? System.getProperty("qpid.default_mandatory")
+ : "false"));
private PublishMode publishMode = PublishMode.ASYNC_PUBLISH_ALL;
- protected BasicMessageProducer(AMQConnection connection, AMQDestination destination, boolean transacted, int channelId,
- AMQSession session, AMQProtocolHandler protocolHandler, long producerId, boolean immediate, boolean mandatory) throws AMQException
+ protected BasicMessageProducer(Logger logger,AMQConnection connection, AMQDestination destination, boolean transacted, int channelId,
+ AMQSession session, AMQProtocolHandler protocolHandler, long producerId,
+ Boolean immediate, Boolean mandatory) throws AMQException
{
- _connection = connection;
+ _logger = logger;
+ _connection = connection;
_destination = destination;
_transacted = transacted;
_protocolHandler = protocolHandler;
@@ -193,8 +151,14 @@ public abstract class BasicMessageProduc
declareDestination(destination);
}
- _immediate = immediate;
- _mandatory = mandatory;
+ _immediate = immediate == null ? _defaultImmediateValue : immediate;
+ _mandatory = mandatory == null
+ ? destination == null ? null
+ : destination instanceof Topic
+ ? _defaultMandatoryTopicValue
+ : _defaultMandatoryValue
+ : mandatory;
+
_userID = connection.getUsername();
setPublishMode();
}
@@ -215,7 +179,10 @@ public abstract class BasicMessageProduc
publishMode = PublishMode.SYNC_PUBLISH_ALL;
}
- _logger.info("MessageProducer " + toString() + " using publish mode : " + publishMode);
+ if (_logger.isDebugEnabled())
+ {
+ _logger.debug("MessageProducer " + toString() + " using publish mode : " + publishMode);
+ }
}
void resubscribe() throws AMQException
@@ -381,7 +348,12 @@ public abstract class BasicMessageProduc
synchronized (_connection.getFailoverMutex())
{
validateDestination(destination);
- sendImpl((AMQDestination) destination, message, _deliveryMode, _messagePriority, _timeToLive, _mandatory,
+ sendImpl((AMQDestination) destination, message, _deliveryMode, _messagePriority, _timeToLive,
+ _mandatory == null
+ ? destination instanceof Topic
+ ? _defaultMandatoryTopicValue
+ : _defaultMandatoryValue
+ : _mandatory,
_immediate);
}
}
@@ -394,7 +366,13 @@ public abstract class BasicMessageProduc
synchronized (_connection.getFailoverMutex())
{
validateDestination(destination);
- sendImpl((AMQDestination) destination, message, deliveryMode, priority, timeToLive, _mandatory, _immediate);
+ sendImpl((AMQDestination) destination, message, deliveryMode, priority, timeToLive,
+ _mandatory == null
+ ? destination instanceof Topic
+ ? _defaultMandatoryTopicValue
+ : _defaultMandatoryValue
+ : _mandatory,
+ _immediate);
}
}
@@ -542,7 +520,10 @@ public abstract class BasicMessageProduc
_logger.debug("Updating original message");
origMessage.setJMSPriority(message.getJMSPriority());
origMessage.setJMSTimestamp(message.getJMSTimestamp());
- _logger.debug("Setting JMSExpiration:" + message.getJMSExpiration());
+ if (_logger.isDebugEnabled())
+ {
+ _logger.debug("Setting JMSExpiration:" + message.getJMSExpiration());
+ }
origMessage.setJMSExpiration(message.getJMSExpiration());
origMessage.setJMSMessageID(message.getJMSMessageID());
}
@@ -646,6 +627,69 @@ public abstract class BasicMessageProduc
}
}
+ /**
+ * If true, messages will not get a timestamp.
+ */
+ protected boolean isDisableTimestamps()
+ {
+ return _disableTimestamps;
+ }
+
+ protected void setDisableTimestamps(boolean disableTimestamps)
+ {
+ _disableTimestamps = disableTimestamps;
+ }
+
+ protected void setDestination(AMQDestination destination)
+ {
+ _destination = destination;
+ }
+
+ protected AMQProtocolHandler getProtocolHandler()
+ {
+ return _protocolHandler;
+ }
+
+ protected void setProtocolHandler(AMQProtocolHandler protocolHandler)
+ {
+ _protocolHandler = protocolHandler;
+ }
+
+ protected int getChannelId()
+ {
+ return _channelId;
+ }
+
+ protected void setChannelId(int channelId)
+ {
+ _channelId = channelId;
+ }
+
+ protected void setSession(AMQSession session)
+ {
+ _session = session;
+ }
+
+ protected String getUserID()
+ {
+ return _userID;
+ }
+
+ protected void setUserID(String userID)
+ {
+ _userID = userID;
+ }
+
+ protected PublishMode getPublishMode()
+ {
+ return publishMode;
+ }
+
+ protected void setPublishMode(PublishMode publishMode)
+ {
+ this.publishMode = publishMode;
+ }
+
Logger getLogger()
{
return _logger;
Modified: qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_10.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_10.java?rev=1334037&r1=1334036&r2=1334037&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_10.java (original)
+++ qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_10.java Fri May 4 15:39:19 2012
@@ -61,9 +61,9 @@ public class BasicMessageProducer_0_10 e
BasicMessageProducer_0_10(AMQConnection connection, AMQDestination destination, boolean transacted, int channelId,
AMQSession session, AMQProtocolHandler protocolHandler, long producerId,
- boolean immediate, boolean mandatory) throws AMQException
+ Boolean immediate, Boolean mandatory) throws AMQException
{
- super(connection, destination, transacted, channelId, session, protocolHandler, producerId, immediate, mandatory);
+ super(_logger, connection, destination, transacted, channelId, session, protocolHandler, producerId, immediate, mandatory);
userIDBytes = Strings.toUTF8(getUserID());
}
Modified: qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_8.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_8.java?rev=1334037&r1=1334036&r2=1334037&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_8.java (original)
+++ qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/BasicMessageProducer_0_8.java Fri May 4 15:39:19 2012
@@ -33,6 +33,9 @@ import org.apache.qpid.framing.ContentHe
import org.apache.qpid.framing.ExchangeDeclareBody;
import org.apache.qpid.framing.MethodRegistry;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Queue;
@@ -42,11 +45,12 @@ import java.util.UUID;
public class BasicMessageProducer_0_8 extends BasicMessageProducer
{
+ private static final Logger _logger = LoggerFactory.getLogger(BasicMessageProducer_0_8.class);
BasicMessageProducer_0_8(AMQConnection connection, AMQDestination destination, boolean transacted, int channelId,
- AMQSession session, AMQProtocolHandler protocolHandler, long producerId, boolean immediate, boolean mandatory) throws AMQException
+ AMQSession session, AMQProtocolHandler protocolHandler, long producerId, Boolean immediate, Boolean mandatory) throws AMQException
{
- super(connection, destination,transacted,channelId,session, protocolHandler, producerId, immediate, mandatory);
+ super(_logger,connection, destination,transacted,channelId,session, protocolHandler, producerId, immediate, mandatory);
}
void declareDestination(AMQDestination destination)
Modified: qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/Closeable.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/Closeable.java?rev=1334037&r1=1334036&r2=1334037&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/Closeable.java (original)
+++ qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/Closeable.java Fri May 4 15:39:19 2012
@@ -81,7 +81,7 @@ public abstract class Closeable
}
/**
- * Checks if this is closis.
+ * Checks if this is closing.
*
* @return <tt>true</tt> if we are closing, <tt>false</tt> otherwise.
*/
@@ -90,6 +90,11 @@ public abstract class Closeable
return _closing.get();
}
+ public void resetClosedFlag()
+ {
+ _closed.set(false);
+ }
+
protected boolean setClosed()
{
return _closed.getAndSet(true);
Modified: qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/XAResourceImpl.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/XAResourceImpl.java?rev=1334037&r1=1334036&r2=1334037&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/XAResourceImpl.java (original)
+++ qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/XAResourceImpl.java Fri May 4 15:39:19 2012
@@ -308,13 +308,16 @@ public class XAResourceImpl implements X
_xaSession.createSession();
convertExecutionErrorToXAErr( e.getException().getErrorCode());
}
- Xid[] result = new Xid[res.getInDoubt().size()];
- int i = 0;
- for (Object obj : res.getInDoubt())
- {
- org.apache.qpid.transport.Xid xid = (org.apache.qpid.transport.Xid) obj;
- result[i] = new XidImpl(xid.getBranchId(), (int) xid.getFormat(), xid.getGlobalId());
- i++;
+ Xid[] result = new Xid[res.getInDoubt() != null ? res.getInDoubt().size() : 0];
+ if(result.length != 0)
+ {
+ int i = 0;
+ for (Object obj : res.getInDoubt())
+ {
+ org.apache.qpid.transport.Xid xid = (org.apache.qpid.transport.Xid) obj;
+ result[i] = new XidImpl(xid.getBranchId(), (int) xid.getFormat(), xid.getGlobalId());
+ i++;
+ }
}
return result;
}
@@ -436,6 +439,16 @@ public class XAResourceImpl implements X
}
}
+ /**
+ * Is this resource currently enlisted in a transaction?
+ *
+ * @return true if the resource is associated with a transaction, false otherwise.
+ */
+ public boolean isEnlisted()
+ {
+ return (_xid != null) ;
+ }
+
//------------------------------------------------------------------------
// Private methods
//------------------------------------------------------------------------
Modified: qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/XASessionImpl.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/XASessionImpl.java?rev=1334037&r1=1334036&r2=1334037&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/XASessionImpl.java (original)
+++ qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/XASessionImpl.java Fri May 4 15:39:19 2012
@@ -6,7 +6,7 @@
* "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
+ * 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
@@ -18,6 +18,7 @@
package org.apache.qpid.client;
import org.apache.qpid.client.message.MessageFactoryRegistry;
+import org.apache.qpid.transport.RangeSet;
import javax.jms.JMSException;
import javax.jms.QueueSession;
@@ -178,4 +179,17 @@ public class XASessionImpl extends AMQSe
{
return (TopicSession) getSession();
}
+
+ @Override
+ protected void acknowledgeImpl()
+ {
+ if (_xaResource.isEnlisted())
+ {
+ acknowledgeMessage(Long.MAX_VALUE, true) ;
+ }
+ else
+ {
+ super.acknowledgeImpl() ;
+ }
+ }
}
Modified: qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/handler/ClientMethodDispatcherImpl.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/handler/ClientMethodDispatcherImpl.java?rev=1334037&r1=1334036&r2=1334037&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/handler/ClientMethodDispatcherImpl.java (original)
+++ qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/handler/ClientMethodDispatcherImpl.java Fri May 4 15:39:19 2012
@@ -95,9 +95,9 @@ public class ClientMethodDispatcherImpl
public static ClientMethodDispatcherImpl newMethodDispatcher(ProtocolVersion version, AMQProtocolSession session)
{
- if (_logger.isInfoEnabled())
+ if (_logger.isDebugEnabled())
{
- _logger.info("New Method Dispatcher:" + session);
+ _logger.debug("New Method Dispatcher:" + session);
}
DispatcherFactory factory = _dispatcherFactories.get(version);
Modified: qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/message/FieldTableSupport.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/message/FieldTableSupport.java?rev=1334037&r1=1334036&r2=1334037&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/message/FieldTableSupport.java (original)
+++ qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/message/FieldTableSupport.java Fri May 4 15:39:19 2012
@@ -48,9 +48,12 @@ public class FieldTableSupport
public static Map<String,Object> convertToMap(FieldTable ft)
{
Map<String,Object> map = new HashMap<String,Object>();
- for (AMQShortString key: ft.keySet() )
+ if(ft != null)
{
- map.put(key.asString(), ft.getObject(key));
+ for (AMQShortString key: ft.keySet() )
+ {
+ map.put(key.asString(), ft.getObject(key));
+ }
}
return map;
Modified: qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java?rev=1334037&r1=1334036&r2=1334037&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java (original)
+++ qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java Fri May 4 15:39:19 2012
@@ -652,7 +652,8 @@ public class AMQProtocolHandler implemen
}
writeFrame(frame);
- return listener.blockForFrame(timeout);
+ long actualTimeout = timeout == -1 ? DEFAULT_SYNC_TIMEOUT : timeout;
+ return listener.blockForFrame(actualTimeout);
// When control resumes before this line, a reply will have been received
// that matches the criteria defined in the blocking listener
}
Modified: qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java?rev=1334037&r1=1334036&r2=1334037&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java (original)
+++ qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolSession.java Fri May 4 15:39:19 2012
@@ -108,7 +108,10 @@ public class AMQProtocolSession implemen
{
_protocolHandler = protocolHandler;
_protocolVersion = connection.getProtocolVersion();
- _logger.info("Using ProtocolVersion for Session:" + _protocolVersion);
+ if (_logger.isDebugEnabled())
+ {
+ _logger.debug("Using ProtocolVersion for Session:" + _protocolVersion);
+ }
_methodDispatcher = ClientMethodDispatcherImpl.newMethodDispatcher(ProtocolVersion.getLatestSupportedVersion(),
this);
_connection = connection;
@@ -302,7 +305,10 @@ public class AMQProtocolSession implemen
*/
public void closeSession(AMQSession session)
{
- _logger.debug("closeSession called on protocol session for session " + session.getChannelId());
+ if (_logger.isDebugEnabled())
+ {
+ _logger.debug("closeSession called on protocol session for session " + session.getChannelId());
+ }
final int channelId = session.getChannelId();
if (channelId <= 0)
{
@@ -393,7 +399,10 @@ public class AMQProtocolSession implemen
public void setProtocolVersion(final ProtocolVersion pv)
{
- _logger.info("Setting ProtocolVersion to :" + pv);
+ if (_logger.isDebugEnabled())
+ {
+ _logger.debug("Setting ProtocolVersion to :" + pv);
+ }
_protocolVersion = pv;
_methodRegistry = MethodRegistry.getMethodRegistry(pv);
_methodDispatcher = ClientMethodDispatcherImpl.newMethodDispatcher(pv, this);
Modified: qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/util/BlockingWaiter.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/util/BlockingWaiter.java?rev=1334037&r1=1334036&r2=1334037&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/util/BlockingWaiter.java (original)
+++ qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/client/util/BlockingWaiter.java Fri May 4 15:39:19 2012
@@ -39,7 +39,7 @@ import java.util.concurrent.locks.Reentr
* differs from a 'rendezvous' in that sense.
*
* <p/>BlockingWaiters are used to coordinate when waiting for an an event that expect a response.
- * They are always used in a 'one-shot' manner, that is, to recieve just one response. Usually the caller has to register
+ * They are always used in a 'one-shot' manner, that is, to receive just one response. Usually the caller has to register
* them as method listeners with an event dispatcher and remember to de-register them (in a finally block) once they
* have been completed.
*
@@ -51,12 +51,12 @@ import java.util.concurrent.locks.Reentr
* <p/><table id="crc"><caption>CRC Card</caption>
* <tr><th> Responsibilities <th> Collaborations </td>
* <tr><td> Accept generic objects as events for processing via {@link #process}. <td>
- * <tr><td> Delegate handling and undserstanding of the object to a concrete implementation. <td>
+ * <tr><td> Delegate handling and understanding of the object to a concrete implementation. <td>
* <tr><td> Block until {@link #process} determines that waiting is no longer required <td>
* <tr><td> Propagate the most recent exception to the consumer.<td>
* </table>
*
- * @todo Interuption is caught but not handled. This could be allowed to fall through. This might actually be usefull
+ * @todo Interruption is caught but not handled. This could be allowed to fall through. This might actually be useful
* for fail-over where a thread is blocking when failure happens, it could be interrupted to abandon or retry
* when this happens. At the very least, restore the interrupted status flag.
* @todo If the retrotranslator can handle it, could use a SynchronousQueue to implement this rendezvous. Need to
@@ -84,13 +84,13 @@ public abstract class BlockingWaiter<T>
/** Used to hold the most recent exception that is passed to the {@link #error(Exception)} method. */
private volatile Exception _error;
- /** Holds the incomming Object. */
+ /** Holds the incoming Object. */
private Object _doneObject = null;
private AtomicBoolean _waiting = new AtomicBoolean(false);
private boolean _closed = false;
/**
- * Delegates processing of the incomming object to the handler.
+ * Delegates processing of the incoming object to the handler.
*
* @param object The object to process.
*
@@ -146,6 +146,11 @@ public abstract class BlockingWaiter<T>
*/
public Object block(long timeout) throws AMQException, FailoverException
{
+ if (timeout < 0)
+ {
+ throw new IllegalArgumentException("timeout must be zero or greater");
+ }
+
long nanoTimeout = TimeUnit.MILLISECONDS.toNanos(timeout);
_lock.lock();
@@ -165,26 +170,18 @@ public abstract class BlockingWaiter<T>
{
try
{
- if (timeout == -1)
- {
- _receivedCondition.await();
- }
- else
- {
- nanoTimeout = _receivedCondition.awaitNanos(nanoTimeout);
+ nanoTimeout = _receivedCondition.awaitNanos(nanoTimeout);
- if (nanoTimeout <= 0 && !_ready && _error == null)
- {
- _error = new AMQTimeoutException("Server did not respond in a timely fashion", null);
- _ready = true;
- }
+ if (nanoTimeout <= 0 && !_ready && _error == null)
+ {
+ _error = new AMQTimeoutException("Server did not respond in a timely fashion", null);
+ _ready = true;
}
}
catch (InterruptedException e)
{
_logger.error(e.getMessage(), e);
- // IGNORE -- //fixme this isn't ideal as being interrupted isn't equivellant to sucess
-
+ // IGNORE -- //fixme this isn't ideal as being interrupted isn't equivalent to success
}
}
}
@@ -285,8 +282,8 @@ public abstract class BlockingWaiter<T>
/**
* Close this Waiter so that no more errors are processed.
* This is a preventative method to ensure that a second error thread does not get stuck in the error method after
- * the await has returned. This has not happend but in practise but if two errors occur on the Connection at
- * the same time then it is conceiveably possible for the second to get stuck if the first one is processed by a
+ * the await has returned. This has not happened but in practise but if two errors occur on the Connection at
+ * the same time then it is conceivably possible for the second to get stuck if the first one is processed by a
* waiter.
*
* Once closed any attempt to wait will throw an exception.
Modified: qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/jndi/PropertiesFileInitialContextFactory.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/jndi/PropertiesFileInitialContextFactory.java?rev=1334037&r1=1334036&r2=1334037&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/jndi/PropertiesFileInitialContextFactory.java (original)
+++ qpid/branches/asyncstore/java/client/src/main/java/org/apache/qpid/jndi/PropertiesFileInitialContextFactory.java Fri May 4 15:39:19 2012
@@ -20,8 +20,26 @@
*/
package org.apache.qpid.jndi;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import java.io.BufferedInputStream;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.Hashtable;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ConcurrentHashMap;
+
+import javax.jms.ConnectionFactory;
+import javax.jms.Destination;
+import javax.jms.Queue;
+import javax.jms.Topic;
+import javax.naming.ConfigurationException;
+import javax.naming.Context;
+import javax.naming.NamingException;
+import javax.naming.spi.InitialContextFactory;
import org.apache.qpid.client.AMQConnectionFactory;
import org.apache.qpid.client.AMQDestination;
@@ -33,23 +51,10 @@ import org.apache.qpid.framing.AMQShortS
import org.apache.qpid.url.BindingURL;
import org.apache.qpid.url.URLSyntaxException;
import org.apache.qpid.util.Strings;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
-import javax.jms.ConnectionFactory;
-import javax.jms.Destination;
-import javax.jms.Queue;
-import javax.jms.Topic;
-import javax.naming.ConfigurationException;
-import javax.naming.Context;
-import javax.naming.NamingException;
-import javax.naming.spi.InitialContextFactory;
-import java.io.BufferedInputStream;
-import java.io.FileInputStream;
-import java.io.IOException;
-import java.util.Hashtable;
-import java.util.Iterator;
-import java.util.Map;
-import java.util.Properties;
-import java.util.concurrent.ConcurrentHashMap;
public class PropertiesFileInitialContextFactory implements InitialContextFactory
{
@@ -60,6 +65,7 @@ public class PropertiesFileInitialContex
private String QUEUE_PREFIX = "queue.";
private String TOPIC_PREFIX = "topic.";
+ @SuppressWarnings({ "rawtypes", "unchecked" })
public Context getInitialContext(Hashtable environment) throws NamingException
{
Map data = new ConcurrentHashMap();
@@ -68,6 +74,7 @@ public class PropertiesFileInitialContex
{
String file = null;
+
if (environment.containsKey(Context.PROVIDER_URL))
{
file = (String) environment.get(Context.PROVIDER_URL);
@@ -77,13 +84,23 @@ public class PropertiesFileInitialContex
file = System.getProperty(Context.PROVIDER_URL);
}
+ // Load the properties specified
if (file != null)
{
_logger.info("Loading Properties from:" + file);
+ BufferedInputStream inputStream = null;
- // Load the properties specified
- BufferedInputStream inputStream = new BufferedInputStream(new FileInputStream(file));
+ if(file.contains("file:"))
+ {
+ inputStream = new BufferedInputStream(new FileInputStream(new File(new URI(file))));
+ }
+ else
+ {
+ inputStream = new BufferedInputStream(new FileInputStream(file));
+ }
+
Properties p = new Properties();
+
try
{
p.load(inputStream);
@@ -119,6 +136,11 @@ public class PropertiesFileInitialContex
_logger.warn("Unable to load property file specified in Provider_URL:" + environment.get(Context.PROVIDER_URL) +"\n" +
"Due to:"+ioe.getMessage());
}
+ catch(URISyntaxException uoe)
+ {
+ _logger.warn("Unable to load property file specified in Provider_URL:" + environment.get(Context.PROVIDER_URL) +"\n" +
+ "Due to:"+uoe.getMessage());
+ }
createConnectionFactories(data, environment);
Modified: qpid/branches/asyncstore/java/client/src/test/java/org/apache/qpid/test/unit/message/TestAMQSession.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/client/src/test/java/org/apache/qpid/test/unit/message/TestAMQSession.java?rev=1334037&r1=1334036&r2=1334037&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/client/src/test/java/org/apache/qpid/test/unit/message/TestAMQSession.java (original)
+++ qpid/branches/asyncstore/java/client/src/test/java/org/apache/qpid/test/unit/message/TestAMQSession.java Fri May 4 15:39:19 2012
@@ -145,7 +145,7 @@ public class TestAMQSession extends AMQS
}
public void sendQueueDeclare(AMQDestination amqd, AMQProtocolHandler protocolHandler,
- boolean nowait) throws AMQException, FailoverException
+ boolean nowait, boolean passive) throws AMQException, FailoverException
{
}
Modified: qpid/branches/asyncstore/java/common.xml
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common.xml?rev=1334037&r1=1334036&r2=1334037&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common.xml (original)
+++ qpid/branches/asyncstore/java/common.xml Fri May 4 15:39:19 2012
@@ -23,7 +23,7 @@
<dirname property="project.root" file="${ant.file.common}"/>
<property name="project.name" value="qpid"/>
- <property name="project.version" value="0.15"/>
+ <property name="project.version" value="0.17"/>
<property name="project.url" value="http://qpid.apache.org"/>
<property name="project.groupid" value="org.apache.qpid"/>
<property name="project.namever" value="${project.name}-${project.version}"/>
Modified: qpid/branches/asyncstore/java/common/src/main/java/common.bnd
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/src/main/java/common.bnd?rev=1334037&r1=1334036&r2=1334037&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/src/main/java/common.bnd (original)
+++ qpid/branches/asyncstore/java/common/src/main/java/common.bnd Fri May 4 15:39:19 2012
@@ -17,7 +17,7 @@
# under the License.
#
-ver: 0.15.0
+ver: 0.17.0
Bundle-SymbolicName: qpid-common
Bundle-Version: ${ver}
Modified: qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/codec/MarkableDataInput.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/codec/MarkableDataInput.java?rev=1334037&r1=1334036&r2=1334037&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/codec/MarkableDataInput.java (original)
+++ qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/codec/MarkableDataInput.java Fri May 4 15:39:19 2012
@@ -1,3 +1,23 @@
+/*
+ *
+ * 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.codec;
import org.apache.qpid.framing.AMQShortString;
Modified: qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/common/AMQPFilterTypes.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/common/AMQPFilterTypes.java?rev=1334037&r1=1334036&r2=1334037&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/common/AMQPFilterTypes.java (original)
+++ qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/common/AMQPFilterTypes.java Fri May 4 15:39:19 2012
@@ -34,11 +34,13 @@ public enum AMQPFilterTypes
{
JMS_SELECTOR("x-filter-jms-selector"),
NO_CONSUME("x-filter-no-consume"),
- AUTO_CLOSE("x-filter-auto-close");
+ AUTO_CLOSE("x-filter-auto-close"),
+ NO_LOCAL("x-qpid-no-local");
/** The identifying string for the filter type. */
private final AMQShortString _value;
+
/**
* Creates a new filter type from its identifying string.
*
Modified: qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/framing/AMQShortString.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/framing/AMQShortString.java?rev=1334037&r1=1334036&r2=1334037&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/framing/AMQShortString.java (original)
+++ qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/framing/AMQShortString.java Fri May 4 15:39:19 2012
@@ -858,4 +858,9 @@ public final class AMQShortString implem
}
}
+ public static String toString(AMQShortString amqShortString)
+ {
+ return amqShortString == null ? null : amqShortString.asString();
+ }
+
}
Modified: qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/framing/ByteArrayDataInput.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/framing/ByteArrayDataInput.java?rev=1334037&r1=1334036&r2=1334037&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/framing/ByteArrayDataInput.java (original)
+++ qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/framing/ByteArrayDataInput.java Fri May 4 15:39:19 2012
@@ -1,3 +1,23 @@
+/*
+ *
+ * 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.framing;
import org.apache.qpid.codec.MarkableDataInput;
Modified: qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderBody.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderBody.java?rev=1334037&r1=1334036&r2=1334037&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderBody.java (original)
+++ qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/framing/ContentHeaderBody.java Fri May 4 15:39:19 2012
@@ -75,17 +75,6 @@ public class ContentHeaderBody implement
return TYPE;
}
- protected void populateFromBuffer(DataInputStream buffer, long size)
- throws AMQFrameDecodingException, AMQProtocolVersionException, IOException
- {
- classId = buffer.readUnsignedShort();
- weight = buffer.readUnsignedShort();
- bodySize = buffer.readLong();
- int propertyFlags = buffer.readUnsignedShort();
- ContentHeaderPropertiesFactory factory = ContentHeaderPropertiesFactory.getInstance();
- properties = factory.createContentHeaderProperties(classId, propertyFlags, buffer, (int)size - 14);
- }
-
/**
* Helper method that is used currently by the persistence layer (by BDB at the moment).
* @param buffer
Modified: qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/framing/EncodingUtils.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/framing/EncodingUtils.java?rev=1334037&r1=1334036&r2=1334037&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/framing/EncodingUtils.java (original)
+++ qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/framing/EncodingUtils.java Fri May 4 15:39:19 2012
@@ -264,7 +264,6 @@ public class EncodingUtils
public static void writeLongStringBytes(DataOutput buffer, String s) throws IOException
{
- assert (s == null) || (s.length() <= 0xFFFE);
if (s != null)
{
int len = s.length();
@@ -286,7 +285,6 @@ public class EncodingUtils
public static void writeLongStringBytes(DataOutput buffer, char[] s) throws IOException
{
- assert (s == null) || (s.length <= 0xFFFE);
if (s != null)
{
int len = s.length;
@@ -307,7 +305,6 @@ public class EncodingUtils
public static void writeLongStringBytes(DataOutput buffer, byte[] bytes) throws IOException
{
- assert (bytes == null) || (bytes.length <= 0xFFFE);
if (bytes != null)
{
writeUnsignedInteger(buffer, bytes.length);
Modified: qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/framing/ExtendedDataInput.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/framing/ExtendedDataInput.java?rev=1334037&r1=1334036&r2=1334037&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/framing/ExtendedDataInput.java (original)
+++ qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/framing/ExtendedDataInput.java Fri May 4 15:39:19 2012
@@ -1,3 +1,23 @@
+/*
+ *
+ * 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.framing;
import java.io.DataInput;
Modified: qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/properties/ConnectionStartProperties.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/properties/ConnectionStartProperties.java?rev=1334037&r1=1334036&r2=1334037&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/properties/ConnectionStartProperties.java (original)
+++ qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/properties/ConnectionStartProperties.java Fri May 4 15:39:19 2012
@@ -74,14 +74,19 @@ public class ConnectionStartProperties
public static String getPlatformInfo()
{
- StringBuffer fullSystemInfo = new StringBuffer();
- fullSystemInfo.append(System.getProperty("java.runtime.name"));
- fullSystemInfo.append(", " + System.getProperty("java.runtime.version"));
- fullSystemInfo.append(", " + System.getProperty("java.vendor"));
- fullSystemInfo.append(", " + System.getProperty("os.arch"));
- fullSystemInfo.append(", " + System.getProperty("os.name"));
- fullSystemInfo.append(", " + System.getProperty("os.version"));
- fullSystemInfo.append(", " + System.getProperty("sun.os.patch.level"));
+ StringBuilder fullSystemInfo = new StringBuilder(System.getProperty("java.runtime.name"));
+ fullSystemInfo.append(", ");
+ fullSystemInfo.append(System.getProperty("java.runtime.version"));
+ fullSystemInfo.append(", ");
+ fullSystemInfo.append(System.getProperty("java.vendor"));
+ fullSystemInfo.append(", ");
+ fullSystemInfo.append(System.getProperty("os.arch"));
+ fullSystemInfo.append(", ");
+ fullSystemInfo.append(System.getProperty("os.name"));
+ fullSystemInfo.append(", ");
+ fullSystemInfo.append(System.getProperty("os.version"));
+ fullSystemInfo.append(", ");
+ fullSystemInfo.append(System.getProperty("sun.os.patch.level"));
return fullSystemInfo.toString();
}
Modified: qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/Connection.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/Connection.java?rev=1334037&r1=1334036&r2=1334037&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/Connection.java (original)
+++ qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/Connection.java Fri May 4 15:39:19 2012
@@ -216,6 +216,7 @@ public class Connection extends Connecti
conSettings = settings;
state = OPENING;
userID = settings.getUsername();
+ connectionLost.set(false);
securityLayer = SecurityLayerFactory.newInstance(getConnectionSettings());
Modified: qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/NetworkTransportConfiguration.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/NetworkTransportConfiguration.java?rev=1334037&r1=1334036&r2=1334037&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/NetworkTransportConfiguration.java (original)
+++ qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/NetworkTransportConfiguration.java Fri May 4 15:39:19 2012
@@ -20,6 +20,8 @@
*/
package org.apache.qpid.transport;
+import java.net.InetSocketAddress;
+
/**
* This interface provides a means for NetworkDrivers to configure TCP options such as incoming and outgoing
* buffer sizes and set particular options on the socket. NetworkDrivers should honour the values returned
@@ -43,4 +45,6 @@ public interface NetworkTransportConfigu
String getTransport();
Integer getConnectorProcessors();
+
+ InetSocketAddress getAddress();
}
Modified: qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/Range.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/Range.java?rev=1334037&r1=1334036&r2=1334037&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/Range.java (original)
+++ qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/Range.java Fri May 4 15:39:19 2012
@@ -122,6 +122,11 @@ public abstract class Range implements R
throw new UnsupportedOperationException();
}
+ public void subtract(RangeSet rangeSet)
+ {
+ throw new UnsupportedOperationException();
+ }
+
public RangeSet copy()
{
RangeSet rangeSet = RangeSetFactory.createRangeSet();
Modified: qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/RangeSet.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/RangeSet.java?rev=1334037&r1=1334036&r2=1334037&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/RangeSet.java (original)
+++ qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/RangeSet.java Fri May 4 15:39:19 2012
@@ -49,6 +49,8 @@ public interface RangeSet extends Iterab
void add(int value);
+ void subtract(final RangeSet other);
+
void clear();
RangeSet copy();
Modified: qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/RangeSetImpl.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/RangeSetImpl.java?rev=1334037&r1=1334036&r2=1334037&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/RangeSetImpl.java (original)
+++ qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/RangeSetImpl.java Fri May 4 15:39:19 2012
@@ -150,6 +150,68 @@ public class RangeSetImpl implements Ran
ranges.clear();
}
+ public void subtract(final RangeSet other)
+ {
+ final Iterator<Range> otherIter = other.iterator() ;
+ if (otherIter.hasNext())
+ {
+ Range otherRange = otherIter.next();
+ final ListIterator<Range> iter = ranges.listIterator() ;
+ if (iter.hasNext())
+ {
+ Range range = iter.next();
+ do
+ {
+ if (otherRange.getUpper() < range.getLower())
+ {
+ otherRange = nextRange(otherIter) ;
+ }
+ else if (range.getUpper() < otherRange.getLower())
+ {
+ range = nextRange(iter) ;
+ }
+ else
+ {
+ final boolean first = range.getLower() < otherRange.getLower() ;
+ final boolean second = otherRange.getUpper() < range.getUpper() ;
+
+ if (first)
+ {
+ iter.set(Range.newInstance(range.getLower(), otherRange.getLower()-1)) ;
+ if (second)
+ {
+ iter.add(Range.newInstance(otherRange.getUpper()+1, range.getUpper())) ;
+ iter.previous() ;
+ range = iter.next() ;
+ }
+ else
+ {
+ range = nextRange(iter) ;
+ }
+ }
+ else if (second)
+ {
+ range = Range.newInstance(otherRange.getUpper()+1, range.getUpper()) ;
+ iter.set(range) ;
+ otherRange = nextRange(otherIter) ;
+ }
+ else
+ {
+ iter.remove() ;
+ range = nextRange(iter) ;
+ }
+ }
+ }
+ while ((otherRange != null) && (range != null)) ;
+ }
+ }
+ }
+
+ private Range nextRange(final Iterator<Range> iter)
+ {
+ return (iter.hasNext() ? iter.next() : null) ;
+ }
+
public RangeSet copy()
{
return new org.apache.qpid.transport.RangeSetImpl(this);
Modified: qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/ServerDelegate.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/ServerDelegate.java?rev=1334037&r1=1334036&r2=1334037&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/ServerDelegate.java (original)
+++ qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/ServerDelegate.java Fri May 4 15:39:19 2012
@@ -191,17 +191,11 @@ public class ServerDelegate extends Conn
@Override
public void sessionAttach(Connection conn, SessionAttach atc)
{
- sessionAttachImpl(conn, atc);
- }
-
- protected Session sessionAttachImpl(Connection conn, SessionAttach atc)
- {
Session ssn = getSession(conn, atc);
conn.map(ssn, atc.getChannel());
+ conn.registerSession(ssn);
ssn.sessionAttached(atc.getName());
ssn.setState(Session.State.OPEN);
-
- return ssn;
}
protected void setConnectionTuneOkChannelMax(final Connection conn, final int okChannelMax)
Modified: qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/Session.java
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/Session.java?rev=1334037&r1=1334036&r2=1334037&view=diff
==============================================================================
--- qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/Session.java (original)
+++ qpid/branches/asyncstore/java/common/src/main/java/org/apache/qpid/transport/Session.java Fri May 4 15:39:19 2012
@@ -161,7 +161,7 @@ public class Session extends SessionInvo
this.expiry = expiry;
}
- void setClose(boolean close)
+ protected void setClose(boolean close)
{
this.closing = close;
}
@@ -513,20 +513,12 @@ public class Session extends SessionInvo
void knownComplete(RangeSet kc)
{
- synchronized (processedLock)
+ if (kc.size() > 0)
{
- RangeSet newProcessed = RangeSetFactory.createRangeSet();
- for (Range pr : processed)
+ synchronized (processedLock)
{
- for (Range kr : kc)
- {
- for (Range r : pr.subtract(kr))
- {
- newProcessed.add(r);
- }
- }
+ processed.subtract(kc) ;
}
- this.processed = newProcessed;
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org