You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by or...@apache.org on 2017/02/21 21:20:44 UTC
svn commit: r1783945 - in /qpid/java/trunk/broker-core/src:
main/java/org/apache/qpid/server/message/RoutingResult.java
test/java/org/apache/qpid/server/message/RoutingResultTest.java
Author: orudyy
Date: Tue Feb 21 21:20:44 2017
New Revision: 1783945
URL: http://svn.apache.org/viewvc?rev=1783945&view=rev
Log:
QPID-7678,QPID-6028: Revert previous changes as it was a deliberate decision: if message could not be enqueued on all queues it should not be enqueued on any
Removed:
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/message/RoutingResultTest.java
Modified:
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/message/RoutingResult.java
Modified: qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/message/RoutingResult.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/message/RoutingResult.java?rev=1783945&r1=1783944&r2=1783945&view=diff
==============================================================================
--- qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/message/RoutingResult.java (original)
+++ qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/message/RoutingResult.java Tue Feb 21 21:20:44 2017
@@ -88,44 +88,56 @@ public class RoutingResult<M extends Ser
public int send(ServerTransaction txn,
final Action<? super MessageInstance> postEnqueueAction)
{
- final ArrayList<BaseQueue> queues = new ArrayList<>();
for(BaseQueue q : _queues)
{
- if(_message.isResourceAcceptable(q) && !_message.isReferenced(q))
+ if(!_message.isResourceAcceptable(q))
{
- queues.add(q);
+ return 0;
}
}
+ final BaseQueue[] baseQueues;
- if (!queues.isEmpty())
+ if(_message.isReferenced())
{
- txn.enqueue(queues, _message, new ServerTransaction.EnqueueAction()
+ ArrayList<BaseQueue> uniqueQueues = new ArrayList<>(_queues.size());
+ for(BaseQueue q : _queues)
{
- MessageReference _reference = _message.newReference();
+ if(!_message.isReferenced(q))
+ {
+ uniqueQueues.add(q);
+ }
+ }
+ baseQueues = uniqueQueues.toArray(new BaseQueue[uniqueQueues.size()]);
+ }
+ else
+ {
+ baseQueues = _queues.toArray(new BaseQueue[_queues.size()]);
+ }
+ txn.enqueue(_queues, _message, new ServerTransaction.EnqueueAction()
+ {
+ MessageReference _reference = _message.newReference();
- public void postCommit(MessageEnqueueRecord... records)
+ public void postCommit(MessageEnqueueRecord... records)
+ {
+ try
{
- try
+ for(int i = 0; i < baseQueues.length; i++)
{
- for (int i = 0; i < queues.size(); i++)
- {
- queues.get(i).enqueue(_message, postEnqueueAction, records[i]);
- }
- }
- finally
- {
- _reference.release();
+ baseQueues[i].enqueue(_message, postEnqueueAction, records[i]);
}
}
-
- public void onRollback()
+ finally
{
_reference.release();
}
- });
- }
+ }
- return queues.size();
+ public void onRollback()
+ {
+ _reference.release();
+ }
+ });
+ return _queues.size();
}
public boolean hasRoutes()
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org