You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rg...@apache.org on 2014/08/25 17:03:05 UTC
svn commit: r1620333 [2/2] - in /qpid/branches/0.30/qpid/java: ./
bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/
broker-core/src/main/java/org/apache/qpid/server/consumer/
broker-core/src/main/java/org/apache/qpid/server/exchange/ brok...
Modified: qpid/branches/0.30/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/SendingLink_1_0.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.30/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/SendingLink_1_0.java?rev=1620333&r1=1620332&r2=1620333&view=diff
==============================================================================
--- qpid/branches/0.30/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/SendingLink_1_0.java (original)
+++ qpid/branches/0.30/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/SendingLink_1_0.java Mon Aug 25 15:03:03 2014
@@ -636,19 +636,21 @@ public class SendingLink_1_0 implements
AutoCommitTransaction txn = new AutoCommitTransaction(_vhost.getMessageStore());
if(_consumer.acquires())
{
- txn.dequeue(Collections.singleton(queueEntry),
- new ServerTransaction.Action()
- {
- public void postCommit()
- {
- queueEntry.delete();
- }
-
- public void onRollback()
- {
- //To change body of implemented methods use File | Settings | File Templates.
- }
- });
+ if(queueEntry.acquire() || queueEntry.isAcquired())
+ {
+ txn.dequeue(Collections.singleton(queueEntry),
+ new ServerTransaction.Action()
+ {
+ public void postCommit()
+ {
+ queueEntry.delete();
+ }
+
+ public void onRollback()
+ {
+ }
+ });
+ }
}
}
else if(outcome instanceof Released)
Modified: qpid/branches/0.30/qpid/java/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNode.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.30/qpid/java/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNode.java?rev=1620333&r1=1620332&r2=1620333&view=diff
==============================================================================
--- qpid/branches/0.30/qpid/java/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNode.java (original)
+++ qpid/branches/0.30/qpid/java/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNode.java Mon Aug 25 15:03:03 2014
@@ -1071,6 +1071,12 @@ class ManagementNode implements MessageS
}
@Override
+ public boolean removeAcquisitionFromConsumer(final ConsumerImpl consumer)
+ {
+ return false;
+ }
+
+ @Override
public void setRedelivered()
{
@@ -1119,6 +1125,18 @@ class ManagementNode implements MessageS
}
@Override
+ public boolean lockAcquisition()
+ {
+ return false;
+ }
+
+ @Override
+ public boolean unlockAcquisition()
+ {
+ return false;
+ }
+
+ @Override
public int getMaximumDeliveryCount()
{
return 0;
Modified: qpid/branches/0.30/qpid/java/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementResponse.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.30/qpid/java/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementResponse.java?rev=1620333&r1=1620332&r2=1620333&view=diff
==============================================================================
--- qpid/branches/0.30/qpid/java/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementResponse.java (original)
+++ qpid/branches/0.30/qpid/java/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementResponse.java Mon Aug 25 15:03:03 2014
@@ -90,6 +90,12 @@ class ManagementResponse implements Mess
}
@Override
+ public boolean removeAcquisitionFromConsumer(final ConsumerImpl consumer)
+ {
+ return consumer == _consumer;
+ }
+
+ @Override
public void setRedelivered()
{
_isRedelivered = true;
@@ -138,6 +144,18 @@ class ManagementResponse implements Mess
}
@Override
+ public boolean lockAcquisition()
+ {
+ return false;
+ }
+
+ @Override
+ public boolean unlockAcquisition()
+ {
+ return false;
+ }
+
+ @Override
public int getMaximumDeliveryCount()
{
return 0;
@@ -190,7 +208,7 @@ class ManagementResponse implements Mess
@Override
public void delete()
{
- // TODO
+ _isDeleted = true;
}
@Override
Modified: qpid/branches/0.30/qpid/java/broker-plugins/management-http/src/main/java/org/apache/qpid/server/management/plugin/servlet/rest/MessageServlet.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.30/qpid/java/broker-plugins/management-http/src/main/java/org/apache/qpid/server/management/plugin/servlet/rest/MessageServlet.java?rev=1620333&r1=1620332&r2=1620333&view=diff
==============================================================================
--- qpid/branches/0.30/qpid/java/broker-plugins/management-http/src/main/java/org/apache/qpid/server/management/plugin/servlet/rest/MessageServlet.java (original)
+++ qpid/branches/0.30/qpid/java/broker-plugins/management-http/src/main/java/org/apache/qpid/server/management/plugin/servlet/rest/MessageServlet.java Mon Aug 25 15:03:03 2014
@@ -33,6 +33,7 @@ import javax.servlet.http.HttpServletRes
import org.apache.log4j.Logger;
import org.codehaus.jackson.map.ObjectMapper;
import org.codehaus.jackson.map.SerializationConfig;
+
import org.apache.qpid.server.consumer.ConsumerImpl;
import org.apache.qpid.server.message.AMQMessageHeader;
import org.apache.qpid.server.message.MessageDeletedException;
@@ -44,6 +45,7 @@ import org.apache.qpid.server.queue.Queu
import org.apache.qpid.server.queue.QueueEntryVisitor;
import org.apache.qpid.server.security.SecurityManager;
import org.apache.qpid.server.security.access.Operation;
+import org.apache.qpid.server.store.TransactionLogResource;
public class MessageServlet extends AbstractServlet
{
@@ -212,7 +214,11 @@ public class MessageServlet extends Abst
@Override
protected void updateEntry(QueueEntry entry, VirtualHost.Transaction txn)
{
- txn.move(entry, _destinationQueue);
+ ServerMessage msg = entry.getMessage();
+ if(msg != null && !msg.isReferenced((TransactionLogResource)_destinationQueue))
+ {
+ txn.move(entry, _destinationQueue);
+ }
}
}
@@ -229,7 +235,11 @@ public class MessageServlet extends Abst
@Override
protected void updateEntry(QueueEntry entry, VirtualHost.Transaction txn)
{
- txn.copy(entry, _destinationQueue);
+ ServerMessage msg = entry.getMessage();
+ if(msg != null && !msg.isReferenced((TransactionLogResource)_destinationQueue))
+ {
+ txn.copy(entry, _destinationQueue);
+ }
}
}
Modified: qpid/branches/0.30/qpid/java/broker-plugins/management-http/src/main/java/resources/addBinding.html
URL: http://svn.apache.org/viewvc/qpid/branches/0.30/qpid/java/broker-plugins/management-http/src/main/java/resources/addBinding.html?rev=1620333&r1=1620332&r2=1620333&view=diff
==============================================================================
--- qpid/branches/0.30/qpid/java/broker-plugins/management-http/src/main/java/resources/addBinding.html (original)
+++ qpid/branches/0.30/qpid/java/broker-plugins/management-http/src/main/java/resources/addBinding.html Mon Aug 25 15:03:03 2014
@@ -15,26 +15,48 @@
~ limitations under the License.
-->
<div class="dijitHidden">
- <div data-dojo-type="dijit.Dialog" style="width:600px;" data-dojo-props="title:'Add Binding'" id="addBinding">
- <form id="formAddBinding" method="post" dojoType="dijit.form.Form">
- <table cellpadding="0" cellspacing="2">
- <tr>
- <td valign="top"><strong>Exchange Name*: </strong></td>
- <td><div id="addBinding.selectExchangeDiv"></div></td>
- </tr>
- <tr>
- <td valign="top"><strong>Queue Name*: </strong></td>
- <td><div id="addBinding.selectQueueDiv"></div></td>
- </tr>
- <tr>
- <td valign="top"><strong>Binding Key*: </strong></td>
- <td><input type="text" required="true" name="name" id="formAddbinding.bindingKey" placeholder="Binding Key"
- dojoType="dijit.form.ValidationTextBox" missingMessage="A name must be supplied" /></td>
- </tr>
- </table>
+ <div data-dojo-type="dijit/Dialog" style="width:600px;" data-dojo-props="title:'Add Binding'" id="addBinding">
+ <form id="formAddBinding" method="post" data-dojo-type="dijit/form/Form">
+
+ <div class="clear">
+ <div class="formLabel-labelCell tableContainer-labelCell">Exchange Name*:</div>
+ <div class="formLabel-controlCell tableContainer-valueCell">
+ <div id="addBinding.selectExchangeDiv"></div>
+ </div>
+ </div>
+ <div class="clear">
+ <div class="formLabel-labelCell tableContainer-labelCell">Queue Name*: </div>
+ <div class="formLabel-controlCell tableContainer-valueCell">
+ <div id="addBinding.selectQueueDiv"></div>
+ </div>
+ </div>
+ <div class="clear">
+ <div class="formLabel-labelCell tableContainer-labelCell">Binding Key*:</div>
+ <div class="formLabel-controlCell tableContainer-valueCell">
+ <input type="text" id="formAddbinding.bindingKey"
+ data-dojo-type="dijit/form/ValidationTextBox"
+ data-dojo-props="
+ name: 'name',
+ placeHolder: 'Binding Key',
+ missingMessage: 'A binding key must be supplied',
+ title: 'Enter binding key'" />
+ </div>
+ </div>
+
+ <div class="clear formBox">
+ <fieldset>
+ <legend>Binding Arguments</legend>
+ <div class="editNoteBanner">NOTE: Only arguments with name and value will be submitted. To edit, please, click on a grid cell.</div>
+ <div id="formAddbinding.bindingArguments"></div>
+ <div>
+ <button data-dojo-type="dijit/form/Button" id="formAddbinding.addArgumentButton" type="button">Add</button>
+ <button data-dojo-type="dijit/form/Button" id="formAddbinding.deleteArgumentButton" type="button">Delete</button>
+ </div>
+ </fieldset>
+ </div>
+
<div class="dijitDialogPaneActionBar">
- <!-- submit buttons -->
- <input type="submit" value="Create Binding" label="Create Binding" dojoType="dijit.form.Button" />
+ <input type="submit" value="Create Binding" label="Create Binding" data-dojo-type="dijit/form/Button" />
</div>
</form>
</div>
Modified: qpid/branches/0.30/qpid/java/broker-plugins/management-http/src/main/java/resources/js/qpid/management/addBinding.js
URL: http://svn.apache.org/viewvc/qpid/branches/0.30/qpid/java/broker-plugins/management-http/src/main/java/resources/js/qpid/management/addBinding.js?rev=1620333&r1=1620332&r2=1620333&view=diff
==============================================================================
--- qpid/branches/0.30/qpid/java/broker-plugins/management-http/src/main/java/resources/js/qpid/management/addBinding.js (original)
+++ qpid/branches/0.30/qpid/java/broker-plugins/management-http/src/main/java/resources/js/qpid/management/addBinding.js Mon Aug 25 15:03:03 2014
@@ -23,6 +23,8 @@ define(["dojo/_base/xhr",
"dojo/_base/array",
"dojo/_base/event",
'dojo/_base/json',
+ "dojo/_base/lang",
+ "dojo/_base/declare",
"dojo/store/Memory",
"dijit/form/FilteringSelect",
"dijit/form/NumberSpinner", // required by the form
@@ -38,8 +40,77 @@ define(["dojo/_base/xhr",
"dijit/form/DateTextBox",
/* basic dojox classes */
"dojox/form/BusyButton", "dojox/form/CheckedMultiSelect",
+ "dojox/grid/EnhancedGrid",
+ "dojo/data/ObjectStore",
"dojo/domReady!"],
- function (xhr, dom, construct, win, registry, parser, array, event, json, Memory, FilteringSelect) {
+ function (xhr, dom, construct, win, registry, parser, array, event, json, lang, declare, Memory, FilteringSelect) {
+
+ var noLocalValues = new Memory({
+ data: [
+ {name:"", id:null},
+ {name:"true", id:true},
+ {name:"false", id:false}
+ ]
+ });
+
+ var xMatchValues = new Memory({
+ data: [
+ {name:"all", id:"all"},
+ {name:"any", id:"any"}
+ ]
+ });
+
+ var defaultBindingArguments = [
+ {id: 0, name:"x-filter-jms-selector", value: null},
+ {id: 1, name:"x-qpid-no-local", value: null}
+ ];
+
+ var GridWidgetProxy = declare("qpid.dojox.grid.cells.GridWidgetProxy", dojox.grid.cells._Widget, {
+ createWidget: function(inNode, inDatum, inRowIndex)
+ {
+ var WidgetClass = this.widgetClass;
+ var widgetProperties = this.getWidgetProps(inDatum);
+ var getWidgetProperties = widgetProperties.getWidgetProperties;
+ if (typeof getWidgetProperties == "function")
+ {
+ var item = this.grid.getItem(inRowIndex);
+ if (item)
+ {
+ var additionalWidgetProperties = getWidgetProperties(inDatum, inRowIndex, item);
+ if (additionalWidgetProperties)
+ {
+ WidgetClass = additionalWidgetProperties.widgetClass;
+ for(var prop in additionalWidgetProperties)
+ {
+ if(additionalWidgetProperties.hasOwnProperty(prop) && !widgetProperties[prop])
+ {
+ widgetProperties[prop] = additionalWidgetProperties[ prop ];
+ }
+ }
+ }
+ }
+ }
+ var widget = new WidgetClass(widgetProperties, inNode);
+ return widget;
+ },
+ getValue: function(inRowIndex)
+ {
+ if (this.widget)
+ {
+ return this.widget.get('value');
+ }
+ return null;
+ },
+ _finish: function(inRowIndex)
+ {
+ if (this.widget)
+ {
+ this.inherited(arguments);
+ this.widget.destroyRecursive();
+ this.widget = null;
+ }
+ }
+ });
var addBinding = {};
@@ -73,6 +144,28 @@ define(["dojo/_base/xhr",
if(addBinding.exchange) {
newBinding.exchange = addBinding.exchange;
}
+
+ addBinding.bindingArgumentsGrid.store.fetch({
+ onComplete:function(items,request)
+ {
+ if(items.length)
+ {
+ array.forEach(items, function(item)
+ {
+ if (item && item.name && item.value)
+ {
+ var bindingArguments = newBinding.arguments;
+ if (!bindingArguments)
+ {
+ bindingArguments = {};
+ newBinding.arguments = bindingArguments;
+ }
+ bindingArguments[item.name]=item.value;
+ }
+ });
+ }
+ }
+ });
return newBinding;
};
@@ -106,6 +199,98 @@ define(["dojo/_base/xhr",
});
+ var argumentsGridNode = dom.byId("formAddbinding.bindingArguments");
+ var objectStore = new dojo.data.ObjectStore({objectStore: new Memory({data:lang.clone(defaultBindingArguments), idProperty: "id"})});
+
+ var layout = [[
+ { name: "Argument Name", field: "name", width: "50%", editable: true },
+ { name: 'Argument Value', field: 'value', width: '50%', editable: true, type: GridWidgetProxy,
+ widgetProps: {
+ getWidgetProperties: function(inDatum, inRowIndex, item)
+ {
+ if (item.name == "x-qpid-no-local")
+ {
+ return {
+ labelAttr: "name",
+ searchAttr: "id",
+ selectOnClick: false,
+ query: { id: "*"},
+ required: false,
+ store: noLocalValues,
+ widgetClass: dijit.form.FilteringSelect
+ };
+ }
+ else if (item.name && item.name.toLowerCase() == "x-match")
+ {
+ return {
+ labelAttr: "name",
+ searchAttr: "id",
+ selectOnClick: false,
+ query: { id: "*"},
+ required: false,
+ store: xMatchValues,
+ widgetClass: dijit.form.FilteringSelect
+ };
+ }
+ return {widgetClass: dijit.form.TextBox };
+ }
+ }
+ }
+ ]];
+
+ var grid = new dojox.grid.EnhancedGrid({
+ selectionMode: "multiple",
+ store: objectStore,
+ singleClickEdit: true,
+ structure: layout,
+ height: "150px",
+ plugins: {indirectSelection: true}
+ }, argumentsGridNode);
+ grid.startup();
+
+ addBinding.bindingArgumentsGrid = grid;
+ addBinding.idGenerator = 1;
+ var addArgumentButton = registry.byId("formAddbinding.addArgumentButton");
+ var deleteArgumentButton = registry.byId("formAddbinding.deleteArgumentButton");
+
+ addArgumentButton.on("click",
+ function(event)
+ {
+ addBinding.idGenerator = addBinding.idGenerator + 1;
+ var newItem = {id:addBinding.idGenerator, name: "", value: ""};
+ grid.store.newItem(newItem);
+ grid.store.save();
+ grid.store.fetch(
+ {
+ onComplete:function(items,request)
+ {
+ var rowIndex = items.length - 1;
+ window.setTimeout(function()
+ {
+ grid.focus.setFocusIndex(rowIndex, 1 );
+ },10);
+ }
+ });
+ }
+ );
+
+ deleteArgumentButton.on("click",
+ function(event)
+ {
+ var data = grid.selection.getSelected();
+ if(data.length)
+ {
+ array.forEach(data, function(selectedItem) {
+ if (selectedItem !== null)
+ {
+ grid.store.deleteItem(selectedItem);
+ }
+ });
+ grid.store.save();
+ }
+ }
+ );
+
theForm.on("submit", function(e) {
event.stop(e);
@@ -154,7 +339,24 @@ define(["dojo/_base/xhr",
addBinding.exchange = obj.exchange;
registry.byId("formAddBinding").reset();
-
+ var grid = addBinding.bindingArgumentsGrid;
+ grid.store.fetch({
+ onComplete:function(items,request)
+ {
+ if(items.length)
+ {
+ array.forEach(items, function(item)
+ {
+ if (item !== null)
+ {
+ grid.store.deleteItem(item);
+ }
+ });
+ }
+ }
+ });
+ array.forEach(lang.clone(defaultBindingArguments), function(item) {grid.store.newItem(item); });
+ grid.store.save();
xhr.get({url: "api/latest/queue/" + encodeURIComponent(obj.virtualhostnode) + "/" + encodeURIComponent(obj.virtualhost) + "?depth=0",
handleAs: "json"}).then(
Modified: qpid/branches/0.30/qpid/java/broker-plugins/management-jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/QueueMBean.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.30/qpid/java/broker-plugins/management-jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/QueueMBean.java?rev=1620333&r1=1620332&r2=1620333&view=diff
==============================================================================
--- qpid/branches/0.30/qpid/java/broker-plugins/management-jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/QueueMBean.java (original)
+++ qpid/branches/0.30/qpid/java/broker-plugins/management-jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/QueueMBean.java Mon Aug 25 15:03:03 2014
@@ -59,6 +59,7 @@ import org.apache.qpid.server.model.Virt
import org.apache.qpid.server.queue.NotificationCheck;
import org.apache.qpid.server.queue.QueueEntry;
import org.apache.qpid.server.queue.QueueEntryVisitor;
+import org.apache.qpid.server.store.TransactionLogResource;
import org.apache.qpid.server.util.ServerScopedRuntimeException;
public class QueueMBean extends AMQManagedObject implements ManagedQueue, QueueNotificationListener
@@ -519,7 +520,8 @@ public class QueueMBean extends AMQManag
final long messageId = message.getMessageNumber();
if ((messageId >= fromMessageId)
- && (messageId <= toMessageId))
+ && (messageId <= toMessageId)
+ && !(message.isReferenced((TransactionLogResource)destinationQueue)))
{
txn.move(entry, destinationQueue);
}
@@ -571,8 +573,8 @@ public class QueueMBean extends AMQManag
}
VirtualHost<?,?,?> vhost = _queue.getParent(VirtualHost.class);
- final Queue<?> queue = vhost.getChildByName(Queue.class, toQueue);
- if (queue == null)
+ final Queue<?> destinationQueue = vhost.getChildByName(Queue.class, toQueue);
+ if (destinationQueue == null)
{
throw new OperationsException("No such queue \""+ toQueue +"\"");
}
@@ -591,9 +593,10 @@ public class QueueMBean extends AMQManag
final long messageId = message.getMessageNumber();
if ((messageId >= fromMessageId)
- && (messageId <= toMessageId))
+ && (messageId <= toMessageId)
+ && !(message.isReferenced((TransactionLogResource)destinationQueue)))
{
- txn.copy(entry, queue);
+ txn.copy(entry, destinationQueue);
}
}
Modified: qpid/branches/0.30/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.30/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java?rev=1620333&r1=1620332&r2=1620333&view=diff
==============================================================================
--- qpid/branches/0.30/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java (original)
+++ qpid/branches/0.30/qpid/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java Mon Aug 25 15:03:03 2014
@@ -20,8 +20,16 @@
*/
package org.apache.qpid.client.protocol;
-import org.apache.qpid.client.HeartbeatListener;
-import org.apache.qpid.util.BytesDataOutput;
+import java.io.IOException;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.Set;
+import java.util.concurrent.CopyOnWriteArraySet;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -31,6 +39,7 @@ import org.apache.qpid.AMQException;
import org.apache.qpid.AMQTimeoutException;
import org.apache.qpid.client.AMQConnection;
import org.apache.qpid.client.AMQSession;
+import org.apache.qpid.client.HeartbeatListener;
import org.apache.qpid.client.failover.FailoverException;
import org.apache.qpid.client.failover.FailoverHandler;
import org.apache.qpid.client.failover.FailoverState;
@@ -59,16 +68,7 @@ import org.apache.qpid.thread.Threading;
import org.apache.qpid.transport.Sender;
import org.apache.qpid.transport.TransportException;
import org.apache.qpid.transport.network.NetworkConnection;
-
-import java.io.IOException;
-import java.net.SocketAddress;
-import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.Iterator;
-import java.util.Set;
-import java.util.concurrent.CopyOnWriteArraySet;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
+import org.apache.qpid.util.BytesDataOutput;
/**
* AMQProtocolHandler is the client side protocol handler for AMQP, it handles all protocol events received from the
@@ -177,6 +177,7 @@ public class AMQProtocolHandler implemen
private long _lastReadTime = System.currentTimeMillis();
private long _lastWriteTime = System.currentTimeMillis();
private HeartbeatListener _heartbeatListener = HeartbeatListener.DEFAULT;
+ private Throwable _initialConnectionException;
/**
* Creates a new protocol handler, associated with the specified client connection instance.
@@ -214,6 +215,8 @@ public class AMQProtocolHandler implemen
// in order to execute AMQConnection#exceptionRecievedout out of synchronization block,
// otherwise it might deadlock with failover mutex
boolean failoverNotAllowed = false;
+ boolean failedWithoutConnecting = false;
+ Throwable initialConnectionException = null;
synchronized (this)
{
if (_logger.isDebugEnabled())
@@ -251,8 +254,11 @@ public class AMQProtocolHandler implemen
}
else
{
+ failedWithoutConnecting = true;
+ initialConnectionException = _initialConnectionException;
_logger.debug("We are in process of establishing the initial connection");
}
+ _initialConnectionException = null;
}
else
{
@@ -265,6 +271,16 @@ public class AMQProtocolHandler implemen
_connection.exceptionReceived(new AMQDisconnectedException(
"Server closed connection and reconnection not permitted.", _stateManager.getLastException()));
}
+ else if(failedWithoutConnecting)
+ {
+ if(initialConnectionException == null)
+ {
+ initialConnectionException = _stateManager.getLastException();
+ }
+ String message = initialConnectionException == null ? "" : initialConnectionException.getMessage();
+ _connection.exceptionReceived(new AMQDisconnectedException(
+ "Connection could not be established: " + message, initialConnectionException));
+ }
}
if (_logger.isDebugEnabled())
@@ -338,6 +354,7 @@ public class AMQProtocolHandler implemen
if (causeIsAConnectionProblem)
{
_logger.info("Connection exception caught therefore going to attempt failover: " + cause, cause);
+ _initialConnectionException = cause;
}
else
{
Modified: qpid/branches/0.30/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/ssl/SSLReceiver.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.30/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/ssl/SSLReceiver.java?rev=1620333&r1=1620332&r2=1620333&view=diff
==============================================================================
--- qpid/branches/0.30/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/ssl/SSLReceiver.java (original)
+++ qpid/branches/0.30/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/ssl/SSLReceiver.java Mon Aug 25 15:03:03 2014
@@ -20,17 +20,18 @@
*/
package org.apache.qpid.transport.network.security.ssl;
-import org.apache.qpid.transport.Receiver;
-import org.apache.qpid.transport.TransportException;
-import org.apache.qpid.transport.network.security.SSLStatus;
-import org.apache.qpid.transport.util.Logger;
+import java.nio.ByteBuffer;
import javax.net.ssl.SSLEngine;
import javax.net.ssl.SSLEngineResult;
import javax.net.ssl.SSLEngineResult.HandshakeStatus;
import javax.net.ssl.SSLEngineResult.Status;
import javax.net.ssl.SSLException;
-import java.nio.ByteBuffer;
+
+import org.apache.qpid.transport.Receiver;
+import org.apache.qpid.transport.TransportException;
+import org.apache.qpid.transport.network.security.SSLStatus;
+import org.apache.qpid.transport.util.Logger;
public class SSLReceiver implements Receiver<ByteBuffer>
{
@@ -192,7 +193,7 @@ public class SSLReceiver implements Rece
{
_sslStatus.getSslLock().notifyAll();
}
- exception(new TransportException("Error in SSLReceiver",e));
+ exception(new TransportException("Error in SSLReceiver: " + e.getMessage(),e));
}
}
Modified: qpid/branches/0.30/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/TimeToLiveTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.30/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/TimeToLiveTest.java?rev=1620333&r1=1620332&r2=1620333&view=diff
==============================================================================
--- qpid/branches/0.30/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/TimeToLiveTest.java (original)
+++ qpid/branches/0.30/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/TimeToLiveTest.java Mon Aug 25 15:03:03 2014
@@ -21,15 +21,8 @@
package org.apache.qpid.server.queue;
-import org.junit.Assert;
-import org.apache.log4j.Logger;
-
-import org.apache.qpid.client.AMQConnection;
-import org.apache.qpid.client.AMQDestination;
-import org.apache.qpid.client.AMQQueue;
-import org.apache.qpid.client.AMQSession;
-import org.apache.qpid.client.AMQTopic;
-import org.apache.qpid.test.utils.QpidBrokerTestCase;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
import javax.jms.Connection;
import javax.jms.JMSException;
@@ -39,8 +32,17 @@ import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TopicSubscriber;
-import java.util.concurrent.locks.Condition;
-import java.util.concurrent.locks.ReentrantLock;
+import javax.naming.NamingException;
+
+import org.apache.log4j.Logger;
+import org.junit.Assert;
+
+import org.apache.qpid.client.AMQConnection;
+import org.apache.qpid.client.AMQDestination;
+import org.apache.qpid.client.AMQQueue;
+import org.apache.qpid.client.AMQSession;
+import org.apache.qpid.client.AMQTopic;
+import org.apache.qpid.test.utils.QpidBrokerTestCase;
public class TimeToLiveTest extends QpidBrokerTestCase
{
@@ -53,18 +55,29 @@ public class TimeToLiveTest extends Qpid
private static final int MSG_COUNT = 50;
private static final long SERVER_TTL_TIMEOUT = 60000L;
+ public void testPassiveTTLWithPrefetch() throws Exception
+ {
+ doTestPassiveTTL(true);
+ }
+
public void testPassiveTTL() throws Exception
{
+ doTestPassiveTTL(false);
+
+ }
+
+ private void doTestPassiveTTL(boolean prefetchMessages) throws JMSException, NamingException
+ {
//Create Client 1
Connection clientConnection = getConnection();
-
+
Session clientSession = clientConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
- Queue queue = clientSession.createQueue(QUEUE);
-
+ Queue queue = clientSession.createQueue(QUEUE);
+
// Create then close the consumer so the queue is actually created
// Closing it then reopening it ensures that the consumer shouldn't get messages
// which should have expired and allows a shorter sleep period. See QPID-1418
-
+
MessageConsumer consumer = clientSession.createConsumer(queue);
consumer.close();
@@ -79,6 +92,12 @@ public class TimeToLiveTest extends Qpid
MessageProducer producer = producerSession.createProducer(queue);
+ consumer = clientSession.createConsumer(queue);
+ if(prefetchMessages)
+ {
+ clientConnection.start();
+ }
+
//Set TTL
int msg = 0;
producer.send(nextMessage(String.valueOf(msg), true, producerSession, producer));
@@ -96,7 +115,6 @@ public class TimeToLiveTest extends Qpid
producerSession.commit();
- consumer = clientSession.createConsumer(queue);
// Ensure we sleep the required amount of time.
ReentrantLock waitLock = new ReentrantLock();
@@ -124,6 +142,16 @@ public class TimeToLiveTest extends Qpid
}
+ if(prefetchMessages)
+ {
+ clientConnection.close();
+ clientConnection = getConnection();
+
+ clientSession = clientConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ queue = clientSession.createQueue(QUEUE);
+ consumer = clientSession.createConsumer(queue);
+ }
+
clientConnection.start();
//Receive Message 0
@@ -131,14 +159,14 @@ public class TimeToLiveTest extends Qpid
Message receivedFirst = consumer.receive(5000);
Message receivedSecond = consumer.receive(5000);
Message receivedThird = consumer.receive(1000);
-
+
// Log the messages to help diagnosis incase of failure
_logger.info("First:"+receivedFirst);
_logger.info("Second:"+receivedSecond);
_logger.info("Third:"+receivedThird);
// Only first and last messages sent should survive expiry
- Assert.assertNull("More messages received", receivedThird);
+ Assert.assertNull("More messages received", receivedThird);
Assert.assertNotNull("First message not received", receivedFirst);
Assert.assertTrue("First message doesn't have first set.", receivedFirst.getBooleanProperty("first"));
Modified: qpid/branches/0.30/qpid/java/systests/src/main/java/org/apache/qpid/systest/management/jmx/QueueManagementTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/0.30/qpid/java/systests/src/main/java/org/apache/qpid/systest/management/jmx/QueueManagementTest.java?rev=1620333&r1=1620332&r2=1620333&view=diff
==============================================================================
--- qpid/branches/0.30/qpid/java/systests/src/main/java/org/apache/qpid/systest/management/jmx/QueueManagementTest.java (original)
+++ qpid/branches/0.30/qpid/java/systests/src/main/java/org/apache/qpid/systest/management/jmx/QueueManagementTest.java Mon Aug 25 15:03:03 2014
@@ -456,6 +456,61 @@ public class QueueManagementTest extends
assertMessageIndicesOn(_destinationQueue, 0, 1, 2, 7, 8);
}
+
+ /**
+ * Tests {@link ManagedQueue#copyMessages(long, long, String)} interface.
+ */
+ public void testCopyMessagesBetweenQueuesWithDuplicates() throws Exception
+ {
+ final int numberOfMessagesToSend = 10;
+ sendMessage(_session, _sourceQueue, numberOfMessagesToSend);
+ syncSession(_session);
+ assertEquals("Unexpected queue depth after send",
+ numberOfMessagesToSend,
+ _managedSourceQueue.getMessageCount().intValue());
+
+ List<Long> amqMessagesIds = getAMQMessageIdsOn(_managedSourceQueue, 1, numberOfMessagesToSend);
+
+ // Copy first three messages to destination
+ long fromMessageId = amqMessagesIds.get(0);
+ long toMessageId = amqMessagesIds.get(2);
+ _managedSourceQueue.copyMessages(fromMessageId, toMessageId, _destinationQueueName);
+
+ assertEquals("Unexpected queue depth on destination queue after first copy",
+ 3,
+ _managedDestinationQueue.getMessageCount().intValue());
+ assertEquals("Unexpected queue depth on source queue after first copy",
+ numberOfMessagesToSend,
+ _managedSourceQueue.getMessageCount().intValue());
+
+ // Now copy a further two messages to destination
+ fromMessageId = amqMessagesIds.get(7);
+ toMessageId = amqMessagesIds.get(8);
+ _managedSourceQueue.copyMessages(fromMessageId, toMessageId, _destinationQueueName);
+ assertEquals("Unexpected queue depth on destination queue after second copy",
+ 5,
+ _managedDestinationQueue.getMessageCount().intValue());
+ assertEquals("Unexpected queue depth on source queue after second copy",
+ numberOfMessagesToSend,
+ _managedSourceQueue.getMessageCount().intValue());
+
+ // Attempt to copy mixture of messages already on and some not already on the queue
+
+ fromMessageId = amqMessagesIds.get(5);
+ toMessageId = amqMessagesIds.get(8);
+ _managedSourceQueue.copyMessages(fromMessageId, toMessageId, _destinationQueueName);
+ assertEquals("Unexpected queue depth on destination queue after second copy",
+ 7,
+ _managedDestinationQueue.getMessageCount().intValue());
+ assertEquals("Unexpected queue depth on source queue after second copy",
+ numberOfMessagesToSend,
+ _managedSourceQueue.getMessageCount().intValue());
+
+ assertMessageIndicesOn(_destinationQueue, 0, 1, 2, 7, 8, 5, 6);
+
+
+ }
+
public void testMoveMessagesBetweenQueuesWithActiveConsumerOnSourceQueue() throws Exception
{
setTestClientSystemProperty(ClientProperties.MAX_PREFETCH_PROP_NAME, new Integer(1).toString());
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org