You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by kw...@apache.org on 2015/02/13 18:07:51 UTC
svn commit: r1659606 - in /qpid/branches/QPID-6262-JavaBrokerNIO: ./ qpid/
qpid/cpp/managementgen/qmfgen/ qpid/cpp/rubygen/ qpid/cpp/src/
qpid/cpp/src/qpid/client/ qpid/cpp/src/qpid/client/amqp0_10/
qpid/doc/book/src/cpp-broker/ qpid/java/ qpid/java/cl...
Author: kwall
Date: Fri Feb 13 17:07:50 2015
New Revision: 1659606
URL: http://svn.apache.org/r1659606
Log:
merged from trunk
Modified:
qpid/branches/QPID-6262-JavaBrokerNIO/ (props changed)
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/ (props changed)
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/managementgen/qmfgen/generate.py
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/rubygen/amqpgen.rb
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/ (props changed)
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/client/QueueOptions.cpp
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/client/QueueOptions.h
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/client/amqp0_10/SenderImpl.h
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/doc/book/src/cpp-broker/Cheat-Sheet-for-configuring-Queue-Options.xml
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/ (props changed)
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate.java
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_8_0.java
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/ChannelToSessionMap.java
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/util/FlowControllingBlockingQueue.java
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/python/ (props changed)
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/python/qpid/messaging/driver.py
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/python/qpid/messaging/endpoints.py
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/python/qpid/tests/messaging/endpoints.py
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/tools/src/java/ (props changed)
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/tools/src/java/qpid-qmf2-tools/src/main/java/org/apache/qpid/qmf2/tools/ (props changed)
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/tools/src/java/qpid-qmf2-tools/src/main/java/org/apache/qpid/qmf2/tools/QpidConfig.java
qpid/branches/QPID-6262-JavaBrokerNIO/qpid/tools/src/py/qpid-config
Propchange: qpid/branches/QPID-6262-JavaBrokerNIO/
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Fri Feb 13 17:07:50 2015
@@ -3,4 +3,4 @@
/qpid/branches/java-broker-bdb-ha2:1576683-1583556
/qpid/branches/java-network-refactor:805429-825319
/qpid/branches/mcpierce-QPID-4719:1477004-1477093
-/qpid/trunk:1643238-1658732
+/qpid/trunk:1643238-1659605
Propchange: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Fri Feb 13 17:07:50 2015
@@ -6,4 +6,4 @@
/qpid/branches/mcpierce-QPID-4719/qpid:1477004-1477093
/qpid/branches/qpid-2935/qpid:1061302-1072333
/qpid/branches/qpid-3346/qpid:1144319-1179855
-/qpid/trunk/qpid:1643238-1658732
+/qpid/trunk/qpid:1643238-1659605
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/managementgen/qmfgen/generate.py
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/managementgen/qmfgen/generate.py?rev=1659606&r1=1659605&r2=1659606&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/managementgen/qmfgen/generate.py (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/managementgen/qmfgen/generate.py Fri Feb 13 17:07:50 2015
@@ -257,6 +257,8 @@ class CMakeLists(Makefile):
class Generator:
+ verbose = False
+
"""
This class manages code generation using template files. It is instantiated
once for an entire code generation session.
@@ -350,7 +352,9 @@ class Generator:
pass
os.rename (tempFile, target)
- print "Generated:", target
+
+ if self.verbose:
+ print "Generated:", target
def targetPackageFile (self, schema, templateFile):
dot = templateFile.find(".")
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/rubygen/amqpgen.rb
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/rubygen/amqpgen.rb?rev=1659606&r1=1659605&r2=1659606&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/rubygen/amqpgen.rb (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/rubygen/amqpgen.rb Fri Feb 13 17:07:50 2015
@@ -489,6 +489,7 @@ class Generator
@prefix=[''] # For indentation or comments.
@indentstr=' ' # One indent level.
@outdent=2
+ @verbose=false
end
# Declare next file to be public API
@@ -504,10 +505,14 @@ class Generator
@out=String.new # Generate in memory first
yield if block
if @path.exist? and @path.read == @out
- puts "Skipped #{@path} - unchanged" # Dont generate if unchanged
+ if @verbose
+ puts "Skipped #{@path} - unchanged" # Dont generate if unchanged
+ end
else
@path.open('w') { |f| f << @out }
- puts "Generated #{@path}"
+ if @verbose
+ puts "Generated #{@path}"
+ end
end
end
end
Propchange: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Fri Feb 13 17:07:50 2015
@@ -6,4 +6,4 @@
/qpid/branches/java-network-refactor/qpid/cpp/src:805429-825319
/qpid/branches/qpid-2935/qpid/cpp/src:1061302-1072333
/qpid/branches/qpid-3346/qpid/cpp/src:1144319-1179855
-/qpid/trunk/qpid/cpp/src:1643238-1658732
+/qpid/trunk/qpid/cpp/src:1643238-1659605
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/client/QueueOptions.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/client/QueueOptions.cpp?rev=1659606&r1=1659605&r2=1659606&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/client/QueueOptions.cpp (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/client/QueueOptions.cpp Fri Feb 13 17:07:50 2015
@@ -40,8 +40,6 @@ const std::string QueueOptions::strRING_
const std::string QueueOptions::strLastValueQueue("qpid.last_value_queue");
const std::string QueueOptions::strLVQMatchProperty("qpid.LVQ_key");
const std::string QueueOptions::strLastValueQueueNoBrowse("qpid.last_value_queue_no_browse");
-const std::string QueueOptions::strQueueEventMode("qpid.queue_event_generation");
-
QueueOptions::~QueueOptions()
{}
@@ -101,11 +99,6 @@ void QueueOptions::clearOrdering()
erase(strLastValueQueue);
}
-void QueueOptions::enableQueueEvents(bool enqueueOnly)
-{
- setInt(strQueueEventMode, enqueueOnly ? ENQUEUE_ONLY : ENQUEUE_AND_DEQUEUE);
-}
-
}
}
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/client/QueueOptions.h
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/client/QueueOptions.h?rev=1659606&r1=1659605&r2=1659606&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/client/QueueOptions.h (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/client/QueueOptions.h Fri Feb 13 17:07:50 2015
@@ -75,28 +75,6 @@ class QPID_CLIENT_CLASS_EXTERN QueueOpti
*/
QPID_CLIENT_EXTERN void clearOrdering();
- /**
- * Turns on event generation for this queue (either enqueue only
- * or for enqueue and dequeue events); the events can then be
- * processed by a regsitered broker plugin.
- *
- * DEPRECATED
- *
- * This is confusing to anyone who sees only the function call
- * and not the variable name / doxygen. Consider the following call:
- *
- * options.enableQueueEvents(false);
- *
- * It looks like it disables queue events, but what it really does is
- * enable both enqueue and dequeue events.
- *
- * Use setInt() instead:
- *
- * options.setInt("qpid.queue_event_generation", 2);
- */
-
- QPID_CLIENT_EXTERN void enableQueueEvents(bool enqueueOnly);
-
static QPID_CLIENT_EXTERN const std::string strMaxCountKey;
static QPID_CLIENT_EXTERN const std::string strMaxSizeKey;
static QPID_CLIENT_EXTERN const std::string strTypeKey;
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/client/amqp0_10/SenderImpl.h
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/client/amqp0_10/SenderImpl.h?rev=1659606&r1=1659605&r2=1659606&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/client/amqp0_10/SenderImpl.h (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/cpp/src/qpid/client/amqp0_10/SenderImpl.h Fri Feb 13 17:07:50 2015
@@ -36,7 +36,7 @@ namespace amqp0_10 {
class AddressResolution;
class MessageSink;
-struct OutgoingMessage;
+class OutgoingMessage;
/**
*
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/doc/book/src/cpp-broker/Cheat-Sheet-for-configuring-Queue-Options.xml
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/doc/book/src/cpp-broker/Cheat-Sheet-for-configuring-Queue-Options.xml?rev=1659606&r1=1659605&r2=1659606&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/doc/book/src/cpp-broker/Cheat-Sheet-for-configuring-Queue-Options.xml (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/doc/book/src/cpp-broker/Cheat-Sheet-for-configuring-Queue-Options.xml Fri Feb 13 17:07:50 2015
@@ -179,34 +179,7 @@
<section role="h3" id="CheatSheetforconfiguringQueueOptions-Settingadditionalbehaviors"><title>
Setting additional behaviors
</title>
- <section role="h4" id="CheatSheetforconfiguringQueueOptions-Queueeventgeneration"><title>
- Queue
- event generation
- </title>
- <para>
- This option is used to determine whether enqueue/dequeue events
- representing changes made to queue state are generated. These
- events can then be processed by plugins such as that used for
- <xref linkend="queue-state-replication"/>.
- </para><para>
- Example:
- </para>
- <programlisting>
-#include "qpid/client/QueueOptions.h"
-
- QueueOptions options;
- options.enableQueueEvents(1);
- session.queueDeclare(arg::queue="my-queue", arg::arguments=options);
-</programlisting>
- <para>
- The boolean option indicates whether only enqueue events should
- be generated. The key set by this is
- 'qpid.queue_event_generation' and the value is and integer value
- of 1 (to replicate only enqueue events) or 2 (to replicate both
- enqueue and dequeue events).
- </para>
<!--h3--></section>
- <!--h4--></section>
<section role="h3" id="CheatSheetforconfiguringQueueOptions-OtherClients"><title>
Other
Clients
Propchange: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Fri Feb 13 17:07:50 2015
@@ -9,4 +9,4 @@
/qpid/branches/java-network-refactor/qpid/java:805429-821809
/qpid/branches/qpid-2935/qpid/java:1061302-1072333
/qpid/trunk/qpid:796646-796653
-/qpid/trunk/qpid/java:1643238-1659391
+/qpid/trunk/qpid/java:1643238-1659605
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java?rev=1659606&r1=1659605&r2=1659606&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnection.java Fri Feb 13 17:07:50 2015
@@ -1216,11 +1216,6 @@ public class AMQConnection extends Close
return _failoverMutex;
}
- public void failoverPrep()
- {
- _delegate.failoverPrep();
- }
-
public void resubscribeSessions() throws JMSException, AMQException, FailoverException
{
_delegate.resubscribeSessions();
@@ -1653,4 +1648,46 @@ public class AMQConnection extends Close
{
return _messageCompressionThresholdSize;
}
+
+ void doWithAllLocks(Runnable r)
+ {
+ doWithAllLocks(r, _sessions.values());
+
+ }
+
+ private void doWithAllLocks(final Runnable r, final List<AMQSession> sessions)
+ {
+ if (!sessions.isEmpty())
+ {
+ AMQSession session = sessions.remove(0);
+
+ final Object dispatcherLock = session.getDispatcherLock();
+ if (dispatcherLock != null)
+ {
+ synchronized (dispatcherLock)
+ {
+ synchronized (session.getMessageDeliveryLock())
+ {
+ doWithAllLocks(r, sessions);
+ }
+ }
+ }
+ else
+ {
+ synchronized (session.getMessageDeliveryLock())
+ {
+ doWithAllLocks(r, sessions);
+ }
+ }
+ }
+ else
+ {
+ synchronized (getFailoverMutex())
+ {
+ r.run();
+ }
+ }
+ }
+
+
}
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate.java?rev=1659606&r1=1659605&r2=1659606&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate.java (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate.java Fri Feb 13 17:07:50 2015
@@ -52,8 +52,6 @@ public interface AMQConnectionDelegate
XASession createXASession(int ackMode) throws JMSException;
- void failoverPrep();
-
void resubscribeSessions() throws JMSException, AMQException, FailoverException;
void closeConnection(long timeout) throws JMSException, AMQException;
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java?rev=1659606&r1=1659605&r2=1659606&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_0_10.java Fri Feb 13 17:07:50 2015
@@ -27,6 +27,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicBoolean;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
@@ -249,7 +250,7 @@ public class AMQConnectionDelegate_0_10
List<AMQSession> sessions = new ArrayList<AMQSession>(_conn.getSessions().values());
for (AMQSession s : sessions)
{
- s.failoverPrep();
+ ((AMQSession_0_10)s).failoverPrep();
}
}
@@ -306,16 +307,21 @@ public class AMQConnectionDelegate_0_10
_qpidConnection.notifyFailoverRequired();
- synchronized (_conn.getFailoverMutex())
+ final AtomicBoolean failoverDone = new AtomicBoolean();
+
+ _conn.doWithAllLocks(new Runnable()
{
+ @Override
+ public void run()
+ {
try
{
if (_conn.firePreFailover(false) && _conn.attemptReconnection())
{
- _conn.failoverPrep();
+ failoverPrep();
_conn.resubscribeSessions();
_conn.fireFailoverComplete();
- return;
+ failoverDone.set(true);
}
}
catch (Exception e)
@@ -327,9 +333,19 @@ public class AMQConnectionDelegate_0_10
_conn.getProtocolHandler().getFailoverLatch().countDown();
_conn.getProtocolHandler().setFailoverLatch(null);
}
+
+ }
+ });
+
+
+ if (failoverDone.get())
+ {
+ return;
}
+
}
+
_conn.setClosed();
final ExceptionListener listener = _conn.getExceptionListenerNoCheck();
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_8_0.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_8_0.java?rev=1659606&r1=1659605&r2=1659606&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_8_0.java (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/AMQConnectionDelegate_8_0.java Fri Feb 13 17:07:50 2015
@@ -350,11 +350,6 @@ public class AMQConnectionDelegate_8_0 i
}
}
- public void failoverPrep()
- {
- // do nothing
- }
-
/**
* For all sessions, and for all consumers in those sessions, resubscribe. This is called during failover handling.
* The caller must hold the failover mutex before calling this method.
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java?rev=1659606&r1=1659605&r2=1659606&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java Fri Feb 13 17:07:50 2015
@@ -169,7 +169,7 @@ public abstract class AMQSession<C exten
private final Lock _subscriberDetails = new ReentrantLock(true);
private final Lock _subscriberAccess = new ReentrantLock(true);
- private final FlowControllingBlockingQueue _queue;
+ private final FlowControllingBlockingQueue<Dispatchable> _queue;
private final AtomicLong _highestDeliveryTag = new AtomicLong(-1);
private final AtomicLong _rollbackMark = new AtomicLong(-1);
@@ -358,7 +358,7 @@ public abstract class AMQSession<C exten
if (_acknowledgeMode == NO_ACKNOWLEDGE)
{
_queue =
- new FlowControllingBlockingQueue(_prefetchHighMark, _prefetchLowMark,
+ new FlowControllingBlockingQueue<Dispatchable>(_prefetchHighMark, _prefetchLowMark,
new FlowControllingBlockingQueue.ThresholdListener()
{
private final AtomicBoolean _suspendState = new AtomicBoolean();
@@ -423,7 +423,7 @@ public abstract class AMQSession<C exten
}
else
{
- _queue = new FlowControllingBlockingQueue(_prefetchHighMark, null);
+ _queue = new FlowControllingBlockingQueue<Dispatchable>(_prefetchHighMark, null);
}
// Add creation logging to tie in with the existing close logging
@@ -1789,7 +1789,7 @@ public abstract class AMQSession<C exten
//in the pre-dispatch queue.
_usingDispatcherForCleanup = true;
- syncDispatchQueue();
+ syncDispatchQueue(false);
// Set to false before sending the recover as 0-8/9/9-1 will
//send messages back before the recover completes, and we
@@ -1881,7 +1881,7 @@ public abstract class AMQSession<C exten
setRollbackMark();
- syncDispatchQueue();
+ syncDispatchQueue(false);
_dispatcher.rollback();
@@ -2201,21 +2201,17 @@ public abstract class AMQSession<C exten
}
- void failoverPrep()
- {
- syncDispatchQueue();
- }
- void syncDispatchQueue()
+ void syncDispatchQueue(final boolean holdDispatchLock)
{
- if (Thread.currentThread() == _dispatcherThread)
+ if (Thread.currentThread() == _dispatcherThread || holdDispatchLock)
{
while (!super.isClosed() && !_queue.isEmpty())
{
Dispatchable disp;
try
{
- disp = (Dispatchable) _queue.take();
+ disp = _queue.take();
}
catch (InterruptedException e)
{
@@ -2267,7 +2263,7 @@ public abstract class AMQSession<C exten
Dispatchable disp;
try
{
- disp = (Dispatchable) _queue.take();
+ disp = _queue.take();
}
catch (InterruptedException e)
{
@@ -3086,7 +3082,7 @@ public abstract class AMQSession<C exten
private void rejectMessagesForConsumerTag(int consumerTag, boolean requeue, boolean rejectAllConsumers)
{
- Iterator messages = _queue.iterator();
+ Iterator<Dispatchable> messages = _queue.iterator();
if (_logger.isDebugEnabled())
{
_logger.debug("Rejecting messages from _queue for Consumer tag(" + consumerTag + ") (PDispatchQ) requeue:"
@@ -3237,6 +3233,12 @@ public abstract class AMQSession<C exten
public abstract void setFlowControl(final boolean active);
+ Object getDispatcherLock()
+ {
+ Dispatcher dispatcher = _dispatcher;
+ return dispatcher == null ? null : dispatcher._lock;
+ }
+
public interface Dispatchable
{
void dispatch(AMQSession ssn);
@@ -3389,10 +3391,18 @@ public abstract class AMQSession<C exten
try
{
- Dispatchable disp;
- while (((disp = (Dispatchable) _queue.take()) != null) && !_closed.get())
+
+ while (((_queue.blockingPeek()) != null) && !_closed.get())
{
- disp.dispatch(AMQSession.this);
+ synchronized (_lock)
+ {
+ Dispatchable disp = _queue.nonBlockingTake();
+
+ if(disp != null)
+ {
+ disp.dispatch(AMQSession.this);
+ }
+ }
}
}
catch (InterruptedException e)
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java?rev=1659606&r1=1659605&r2=1659606&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession_0_10.java Fri Feb 13 17:07:50 2015
@@ -235,7 +235,7 @@ public class AMQSession_0_10 extends AMQ
void failoverPrep()
{
- super.failoverPrep();
+ syncDispatchQueue(true);
clearUnacked();
}
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java?rev=1659606&r1=1659605&r2=1659606&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/BasicMessageConsumer_0_10.java Fri Feb 13 17:07:50 2015
@@ -412,7 +412,7 @@ public class BasicMessageConsumer_0_10 e
_capacity,
Option.UNRELIABLE);
}
- _0_10session.syncDispatchQueue();
+ _0_10session.syncDispatchQueue(false);
o = super.getMessageFromQueue(-1);
}
if (_capacity == 0)
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/ChannelToSessionMap.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/ChannelToSessionMap.java?rev=1659606&r1=1659605&r2=1659606&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/ChannelToSessionMap.java (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/ChannelToSessionMap.java Fri Feb 13 17:07:50 2015
@@ -22,13 +22,16 @@ package org.apache.qpid.client;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
public final class ChannelToSessionMap
{
- private final Map<Integer, AMQSession> _sessionMap = new ConcurrentHashMap<>();
+ private final Map<Integer, AMQSession> _sessionMap = Collections.synchronizedMap(new LinkedHashMap<Integer, AMQSession>());
private AtomicInteger _idFactory = new AtomicInteger(0);
private int _maxChannelID;
private int _minChannelID;
@@ -48,7 +51,7 @@ public final class ChannelToSessionMap
_sessionMap.remove(channelId);
}
- public Collection<AMQSession> values()
+ public List<AMQSession> values()
{
return new ArrayList<>(_sessionMap.values());
}
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/util/FlowControllingBlockingQueue.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/util/FlowControllingBlockingQueue.java?rev=1659606&r1=1659605&r2=1659606&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/util/FlowControllingBlockingQueue.java (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/java/client/src/main/java/org/apache/qpid/client/util/FlowControllingBlockingQueue.java Fri Feb 13 17:07:50 2015
@@ -20,13 +20,13 @@
*/
package org.apache.qpid.client.util;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
import java.util.Iterator;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
/**
* A blocking queue that emits events above a user specified threshold allowing the caller to take action (e.g. flow
* control) to try to prevent the queue growing (much) further. The underlying queue itself is not bounded therefore the
@@ -37,12 +37,12 @@ import java.util.concurrent.ConcurrentLi
* <p>
* TODO Make this implement java.util.Queue and hide the implementation. Then different queue types can be substituted.
*/
-public class FlowControllingBlockingQueue
+public class FlowControllingBlockingQueue<T>
{
private static final Logger _logger = LoggerFactory.getLogger(FlowControllingBlockingQueue.class);
/** This queue is bounded and is used to store messages before being dispatched to the consumer */
- private final Queue _queue = new ConcurrentLinkedQueue();
+ private final Queue<T> _queue = new ConcurrentLinkedQueue<T>();
private final int _flowControlHighThreshold;
private final int _flowControlLowThreshold;
@@ -82,9 +82,44 @@ public class FlowControllingBlockingQueu
}
}
- public Object take() throws InterruptedException
+ public T blockingPeek() throws InterruptedException
+ {
+ T o = _queue.peek();
+ if (o == null)
+ {
+ synchronized (this)
+ {
+ while ((o = _queue.peek()) == null)
+ {
+ wait();
+ }
+ }
+ }
+ return o;
+ }
+
+ public T nonBlockingTake() throws InterruptedException
+ {
+ T o = _queue.poll();
+
+ if (o != null && !disableFlowControl && _listener != null)
+ {
+ synchronized (_listener)
+ {
+ if (_count-- == _flowControlLowThreshold)
+ {
+ _listener.underThreshold(_count);
+ }
+ }
+
+ }
+
+ return o;
+ }
+
+ public T take() throws InterruptedException
{
- Object o = _queue.poll();
+ T o = _queue.poll();
if(o == null)
{
synchronized(this)
@@ -110,7 +145,7 @@ public class FlowControllingBlockingQueu
return o;
}
- public void add(Object o)
+ public void add(T o)
{
synchronized(this)
{
@@ -130,7 +165,7 @@ public class FlowControllingBlockingQueu
}
}
- public Iterator iterator()
+ public Iterator<T> iterator()
{
return _queue.iterator();
}
Propchange: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/python/
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Fri Feb 13 17:07:50 2015
@@ -3,4 +3,4 @@
/qpid/branches/java-network-refactor/qpid/python:805429-825319
/qpid/branches/qmfv2/qpid/python:902858,902894
/qpid/branches/qpid-2935/qpid/python:1061302-1072333
-/qpid/trunk/qpid/python:1643238-1658732
+/qpid/trunk/qpid/python:1643238-1659605
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/python/qpid/messaging/driver.py
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/python/qpid/messaging/driver.py?rev=1659606&r1=1659605&r2=1659606&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/python/qpid/messaging/driver.py (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/python/qpid/messaging/driver.py Fri Feb 13 17:07:50 2015
@@ -1368,7 +1368,8 @@ class Engine:
assert rcv.received < rcv.impending, "%s, %s" % (rcv.received, rcv.impending)
rcv.received += 1
log.debug("RCVD[%s]: %s", ssn.log_id, msg)
- ssn.incoming.append(msg)
+ ssn.message_received(msg)
+
def _decode(self, xfr):
dp = EMPTY_DP
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/python/qpid/messaging/endpoints.py
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/python/qpid/messaging/endpoints.py?rev=1659606&r1=1659605&r2=1659606&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/python/qpid/messaging/endpoints.py (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/python/qpid/messaging/endpoints.py Fri Feb 13 17:07:50 2015
@@ -569,6 +569,7 @@ class Session(Endpoint):
self.closed = False
self._lock = connection._lock
+ self._msg_received = None
def __repr__(self):
return "<Session %s>" % self.name
@@ -600,6 +601,11 @@ class Session(Endpoint):
if self.closed:
raise SessionClosed()
+ def message_received(self, msg):
+ self.incoming.append(msg)
+ if self._msg_received:
+ self._msg_received()
+
@synchronized
def sender(self, target, **options):
"""
@@ -685,6 +691,18 @@ class Session(Endpoint):
return None
@synchronized
+ def set_message_received_handler(self, handler):
+ """Register a callback that will be invoked when a message arrives on the
+ session. Use with caution: since this callback is invoked in the context
+ of the driver thread, it is not safe to call any of the public messaging
+ APIs from within this callback. The intent of the handler is to provide
+ an efficient way to notify the application that a message has arrived.
+ This can be useful for those applications that need to schedule a task
+ to poll for received messages without blocking in the messaging API.
+ """
+ self._msg_received = handler
+
+ @synchronized
def next_receiver(self, timeout=None):
if self._ecwait(lambda: self.incoming, timeout):
return self.incoming[0]._receiver
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/python/qpid/tests/messaging/endpoints.py
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/python/qpid/tests/messaging/endpoints.py?rev=1659606&r1=1659605&r2=1659606&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/python/qpid/tests/messaging/endpoints.py (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/python/qpid/tests/messaging/endpoints.py Fri Feb 13 17:07:50 2015
@@ -660,6 +660,31 @@ class SessionTests(Base):
except Detached:
pass
+ def testRxCallback(self):
+ """Verify that the callback is invoked when a message is received.
+ """
+ ADDR = 'test-rx_callback-queue; {create: always, delete: receiver}'
+ class CallbackHandler:
+ def __init__(self):
+ self.handler_called = False
+ def __call__(self):
+ self.handler_called = True
+ cb = CallbackHandler()
+ self.ssn.set_message_received_handler(cb)
+ rcv = self.ssn.receiver(ADDR)
+ rcv.capacity = UNLIMITED
+ snd = self.ssn.sender(ADDR)
+ assert not cb.handler_called
+ snd.send("Ping")
+ deadline = time.time() + self.timeout()
+ while time.time() < deadline:
+ if cb.handler_called:
+ break;
+ assert cb.handler_called
+ snd.close()
+ rcv.close()
+
+
RECEIVER_Q = 'test-receiver-queue; {create: always, delete: always}'
class ReceiverTests(Base):
Propchange: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/tools/src/java/
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Fri Feb 13 17:07:50 2015
@@ -5,3 +5,4 @@
/qpid/branches/mcpierce-QPID-4719/qpid/tools/src/java:1477004-1477093
/qpid/branches/qpid-2935/qpid/tools/src/java:1061302-1072333
/qpid/branches/qpid-3346/qpid/tools/src/java:1144319-1179855
+/qpid/trunk/qpid/tools/src/java:1643238-1659605
Propchange: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/tools/src/java/qpid-qmf2-tools/src/main/java/org/apache/qpid/qmf2/tools/
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Fri Feb 13 17:07:50 2015
@@ -5,3 +5,4 @@
/qpid/branches/mcpierce-QPID-4719/qpid/tools/src/java/src/main/java/org/apache/qpid/qmf2/tools:1477004-1477093
/qpid/branches/qpid-2935/qpid/tools/src/java/src/main/java/org/apache/qpid/qmf2/tools:1061302-1072333
/qpid/branches/qpid-3346/qpid/tools/src/java/src/main/java/org/apache/qpid/qmf2/tools:1144319-1179855
+/qpid/trunk/qpid/tools/src/java/qpid-qmf2-tools/src/main/java/org/apache/qpid/qmf2/tools:1643238-1659605
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/tools/src/java/qpid-qmf2-tools/src/main/java/org/apache/qpid/qmf2/tools/QpidConfig.java
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/tools/src/java/qpid-qmf2-tools/src/main/java/org/apache/qpid/qmf2/tools/QpidConfig.java?rev=1659606&r1=1659605&r2=1659606&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/tools/src/java/qpid-qmf2-tools/src/main/java/org/apache/qpid/qmf2/tools/QpidConfig.java (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/tools/src/java/qpid-qmf2-tools/src/main/java/org/apache/qpid/qmf2/tools/QpidConfig.java Fri Feb 13 17:07:50 2015
@@ -344,7 +344,6 @@ public final class QpidConfig
private String _order = "fifo";
private boolean _msgSequence = false;
private boolean _ive = false;
- private long _eventGeneration = 0;
private String _file = null;
// New to Qpid 0.10 qpid-config
@@ -364,7 +363,6 @@ public final class QpidConfig
private static final String LVQNB = "qpid.last_value_queue_no_browse";
private static final String MSG_SEQUENCE = "qpid.msg_sequence";
private static final String IVE = "qpid.ive";
- private static final String QUEUE_EVENT_GENERATION = "qpid.queue_event_generation";
private static final String FLOW_STOP_COUNT = "qpid.flow_stop_count";
private static final String FLOW_RESUME_COUNT = "qpid.flow_resume_count";
private static final String FLOW_STOP_SIZE = "qpid.flow_stop_size";
@@ -387,7 +385,6 @@ public final class QpidConfig
SPECIAL_ARGS.add(LVQNB);
SPECIAL_ARGS.add(MSG_SEQUENCE);
SPECIAL_ARGS.add(IVE);
- SPECIAL_ARGS.add(QUEUE_EVENT_GENERATION);
SPECIAL_ARGS.add(FLOW_STOP_COUNT);
SPECIAL_ARGS.add(FLOW_RESUME_COUNT);
SPECIAL_ARGS.add(FLOW_STOP_SIZE);
@@ -714,11 +711,6 @@ for (Map.Entry<String, Object> entry :
System.out.printf("--order lvq-no-browse ");
}
- if (args.containsKey(QUEUE_EVENT_GENERATION))
- {
- System.out.printf("--generate-queue-events=%d ", QmfData.getLong(args.get(QUEUE_EVENT_GENERATION)));
- }
-
if (queue.hasValue("altExchange"))
{
ObjectId altExchangeRef = queue.getRefValue("altExchange");
@@ -936,11 +928,6 @@ for (Map.Entry<String, Object> entry :
properties.put(LVQNB, 1l);
}
- if (_eventGeneration > 0)
- {
- properties.put(QUEUE_EVENT_GENERATION, _eventGeneration);
- }
-
if (_altExchange != null)
{
properties.put("alternate-exchange", _altExchange);
@@ -1393,11 +1380,6 @@ for (Map.Entry<String, Object> entry :
_ive = true;
}
- if (opt[0].equals("--generate-queue-events"))
- {
- _eventGeneration = Long.parseLong(opt[1]);
- }
-
if (opt[0].equals("--force"))
{
_ifEmpty = false;
Modified: qpid/branches/QPID-6262-JavaBrokerNIO/qpid/tools/src/py/qpid-config
URL: http://svn.apache.org/viewvc/qpid/branches/QPID-6262-JavaBrokerNIO/qpid/tools/src/py/qpid-config?rev=1659606&r1=1659605&r2=1659606&view=diff
==============================================================================
--- qpid/branches/QPID-6262-JavaBrokerNIO/qpid/tools/src/py/qpid-config (original)
+++ qpid/branches/QPID-6262-JavaBrokerNIO/qpid/tools/src/py/qpid-config Fri Feb 13 17:07:50 2015
@@ -147,7 +147,6 @@ POLICY_TYPE = "qpid.policy_type"
LVQ_KEY = "qpid.last_value_queue_key"
MSG_SEQUENCE = "qpid.msg_sequence"
IVE = "qpid.ive"
-QUEUE_EVENT_GENERATION = "qpid.queue_event_generation"
FLOW_STOP_COUNT = "qpid.flow_stop_count"
FLOW_RESUME_COUNT = "qpid.flow_resume_count"
FLOW_STOP_SIZE = "qpid.flow_stop_size"
@@ -164,7 +163,7 @@ REPLICATE = "qpid.replicate"
SPECIAL_ARGS=[
FILECOUNT,FILESIZE,EFP_PARTITION_NUM,EFP_POOL_FILE_SIZE,
MAX_QUEUE_SIZE,MAX_QUEUE_COUNT,POLICY_TYPE,
- LVQ_KEY,MSG_SEQUENCE,IVE,QUEUE_EVENT_GENERATION,
+ LVQ_KEY,MSG_SEQUENCE,IVE,
FLOW_STOP_COUNT,FLOW_RESUME_COUNT,FLOW_STOP_SIZE,FLOW_RESUME_SIZE,
MSG_GROUP_HDR_KEY,SHARED_MSG_GROUP,REPLICATE]
@@ -541,7 +540,6 @@ class BrokerManager:
if MAX_QUEUE_COUNT in args: print "--max-queue-count=%s" % args[MAX_QUEUE_COUNT],
if POLICY_TYPE in args: print "--limit-policy=%s" % args[POLICY_TYPE].replace("_", "-"),
if LVQ_KEY in args: print "--lvq-key=%s" % args[LVQ_KEY],
- if QUEUE_EVENT_GENERATION in args: print "--generate-queue-events=%s" % args[QUEUE_EVENT_GENERATION],
if q.altExchange:
print "--alternate-exchange=%s" % q.altExchange,
if FLOW_STOP_SIZE in args: print "--flow-stop-size=%s" % args[FLOW_STOP_SIZE],
@@ -638,8 +636,6 @@ class BrokerManager:
if config._lvq_key:
declArgs[LVQ_KEY] = config._lvq_key
- if config._eventGeneration:
- declArgs[QUEUE_EVENT_GENERATION] = config._eventGeneration
if config._flowStopSize is not None:
declArgs[FLOW_STOP_SIZE] = config._flowStopSize
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org