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/10/21 23:11:33 UTC
svn commit: r1633466 - in /qpid/trunk/qpid/java:
broker-core/src/main/java/org/apache/qpid/server/model/port/
broker-core/src/main/java/org/apache/qpid/server/plugin/
broker-core/src/main/java/org/apache/qpid/server/protocol/
broker-core/src/test/java/...
Author: rgodfrey
Date: Tue Oct 21 21:11:31 2014
New Revision: 1633466
URL: http://svn.apache.org/r1633466
Log:
QPID-6175 : [Java Broker] allow maximum message size to be restricted
Modified:
qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/port/AmqpPort.java
qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/plugin/ProtocolEngineCreator.java
qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/protocol/AMQConnectionModel.java
qpid/trunk/qpid/java/broker-core/src/test/java/org/apache/qpid/server/consumer/MockConsumer.java
qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageTransferMessage.java
qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ProtocolEngineCreator_0_10.java
qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ProtocolEngine_0_10.java
qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java
qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSessionDelegate.java
qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/test/java/org/apache/qpid/server/protocol/v0_10/ServerSessionTest.java
qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java
qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java
qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/IncomingMessage.java
qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ProtocolEngineCreator_0_8.java
qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ProtocolEngineCreator_0_9.java
qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ProtocolEngineCreator_0_9_1.java
qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AMQChannelTest.java
qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngineTest.java
qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/BrokerTestHelper_0_8.java
qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/InternalTestProtocolSession.java
qpid/trunk/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java
qpid/trunk/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ProtocolEngineCreator_1_0_0_SASL.java
qpid/trunk/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ProtocolEngine_1_0_0_SASL.java
qpid/trunk/qpid/java/systests/src/test/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngineFactoryTest.java
Modified: qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/port/AmqpPort.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/port/AmqpPort.java?rev=1633466&r1=1633465&r2=1633466&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/port/AmqpPort.java (original)
+++ qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/model/port/AmqpPort.java Tue Oct 21 21:11:31 2014
@@ -59,6 +59,12 @@ public interface AmqpPort<X extends Amqp
@ManagedContextDefault(name = PORT_MAX_OPEN_CONNECTIONS)
int DEFAULT_MAX_OPEN_CONNECTIONS = -1;
+
+ String PORT_MAX_MESSAGE_SIZE = "qpid.port.max_message_size";
+
+ @ManagedContextDefault(name = PORT_MAX_MESSAGE_SIZE)
+ int DEFAULT_MAX_MESSAGE_SIZE = 0x1f40000; // 500Mb
+
String OPEN_CONNECTIONS_WARN_PERCENT = "qpid.port.open_connections_warn_percent";
@ManagedContextDefault(name = OPEN_CONNECTIONS_WARN_PERCENT)
Modified: qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/plugin/ProtocolEngineCreator.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/plugin/ProtocolEngineCreator.java?rev=1633466&r1=1633465&r2=1633466&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/plugin/ProtocolEngineCreator.java (original)
+++ qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/plugin/ProtocolEngineCreator.java Tue Oct 21 21:11:31 2014
@@ -21,18 +21,18 @@ package org.apache.qpid.server.plugin;/*
import org.apache.qpid.protocol.ServerProtocolEngine;
import org.apache.qpid.server.model.Broker;
-import org.apache.qpid.server.model.Port;
import org.apache.qpid.server.model.Protocol;
import org.apache.qpid.server.model.Transport;
+import org.apache.qpid.server.model.port.AmqpPort;
import org.apache.qpid.transport.network.NetworkConnection;
public interface ProtocolEngineCreator extends Pluggable
{
Protocol getVersion();
byte[] getHeaderIdentifier();
- ServerProtocolEngine newProtocolEngine(Broker broker,
+ ServerProtocolEngine newProtocolEngine(Broker<?> broker,
NetworkConnection network,
- Port port,
+ AmqpPort<?> port,
Transport transport,
long id);
}
Modified: qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/protocol/AMQConnectionModel.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/protocol/AMQConnectionModel.java?rev=1633466&r1=1633465&r2=1633466&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/protocol/AMQConnectionModel.java (original)
+++ qpid/trunk/qpid/java/broker-core/src/main/java/org/apache/qpid/server/protocol/AMQConnectionModel.java Tue Oct 21 21:11:31 2014
@@ -26,9 +26,9 @@ import java.util.List;
import org.apache.qpid.protocol.AMQConstant;
import org.apache.qpid.server.logging.LogSubject;
-import org.apache.qpid.server.model.Port;
import org.apache.qpid.server.model.Transport;
import org.apache.qpid.server.model.VirtualHost;
+import org.apache.qpid.server.model.port.AmqpPort;
import org.apache.qpid.server.stats.StatisticsGatherer;
import org.apache.qpid.server.util.Deletable;
@@ -89,7 +89,7 @@ public interface AMQConnectionModel<T ex
long getLastIoTime();
- Port<?> getPort();
+ AmqpPort<?> getPort();
Transport getTransport();
Modified: qpid/trunk/qpid/java/broker-core/src/test/java/org/apache/qpid/server/consumer/MockConsumer.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-core/src/test/java/org/apache/qpid/server/consumer/MockConsumer.java?rev=1633466&r1=1633465&r2=1633466&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker-core/src/test/java/org/apache/qpid/server/consumer/MockConsumer.java (original)
+++ qpid/trunk/qpid/java/broker-core/src/test/java/org/apache/qpid/server/consumer/MockConsumer.java Tue Oct 21 21:11:31 2014
@@ -46,10 +46,10 @@ import org.apache.qpid.server.model.Brok
import org.apache.qpid.server.model.ConfiguredObjectFactory;
import org.apache.qpid.server.model.ConfiguredObjectFactoryImpl;
import org.apache.qpid.server.model.Consumer;
-import org.apache.qpid.server.model.Port;
import org.apache.qpid.server.model.Session;
import org.apache.qpid.server.model.Transport;
import org.apache.qpid.server.model.VirtualHost;
+import org.apache.qpid.server.model.port.AmqpPort;
import org.apache.qpid.server.protocol.AMQConnectionModel;
import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.protocol.ConsumerListener;
@@ -618,7 +618,7 @@ public class MockConsumer implements Con
}
@Override
- public Port<?> getPort()
+ public AmqpPort<?> getPort()
{
return null;
}
Modified: qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageTransferMessage.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageTransferMessage.java?rev=1633466&r1=1633465&r2=1633466&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageTransferMessage.java (original)
+++ qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageTransferMessage.java Tue Oct 21 21:11:31 2014
@@ -20,13 +20,13 @@
*/
package org.apache.qpid.server.protocol.v0_10;
+import java.nio.ByteBuffer;
+
import org.apache.qpid.server.message.AMQMessageHeader;
import org.apache.qpid.server.message.AbstractServerMessageImpl;
import org.apache.qpid.server.store.StoredMessage;
import org.apache.qpid.transport.Header;
-import java.nio.ByteBuffer;
-
public class MessageTransferMessage extends AbstractServerMessageImpl<MessageTransferMessage, MessageMetaData_0_10>
{
@@ -51,13 +51,6 @@ public class MessageTransferMessage exte
return getMetaData().getMessageHeader();
}
- public boolean isRedelivered()
- {
- // The *Message* is never redelivered, only queue entries are... this is here so that filters
- // can run against the message on entry to an exchange
- return false;
- }
-
public long getSize()
{
return getMetaData().getSize();
Modified: qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ProtocolEngineCreator_0_10.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ProtocolEngineCreator_0_10.java?rev=1633466&r1=1633465&r2=1633466&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ProtocolEngineCreator_0_10.java (original)
+++ qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ProtocolEngineCreator_0_10.java Tue Oct 21 21:11:31 2014
@@ -25,9 +25,9 @@ import java.net.SocketAddress;
import org.apache.qpid.protocol.ServerProtocolEngine;
import org.apache.qpid.server.model.Broker;
-import org.apache.qpid.server.model.Port;
import org.apache.qpid.server.model.Protocol;
import org.apache.qpid.server.model.Transport;
+import org.apache.qpid.server.model.port.AmqpPort;
import org.apache.qpid.server.plugin.PluggableService;
import org.apache.qpid.server.plugin.ProtocolEngineCreator;
import org.apache.qpid.transport.ConnectionDelegate;
@@ -64,9 +64,9 @@ public class ProtocolEngineCreator_0_10
return AMQP_0_10_HEADER;
}
- public ServerProtocolEngine newProtocolEngine(Broker broker,
+ public ServerProtocolEngine newProtocolEngine(Broker<?> broker,
NetworkConnection network,
- Port port,
+ AmqpPort<?> port,
Transport transport,
long id)
{
@@ -80,12 +80,13 @@ public class ProtocolEngineCreator_0_10
fqdn, broker.getSubjectCreator(address, transport.isSecure())
);
- ServerConnection conn = new ServerConnection(id,broker);
+ ServerConnection conn = new ServerConnection(id,broker, port, transport);
conn.setConnectionDelegate(connDelegate);
conn.setRemoteAddress(network.getRemoteAddress());
conn.setLocalAddress(network.getLocalAddress());
- return new ProtocolEngine_0_10( conn, network, port, transport);
+
+ return new ProtocolEngine_0_10( conn, network);
}
Modified: qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ProtocolEngine_0_10.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ProtocolEngine_0_10.java?rev=1633466&r1=1633465&r2=1633466&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ProtocolEngine_0_10.java (original)
+++ qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ProtocolEngine_0_10.java Tue Oct 21 21:11:31 2014
@@ -32,7 +32,6 @@ import org.apache.log4j.Logger;
import org.apache.qpid.protocol.ServerProtocolEngine;
import org.apache.qpid.server.logging.messages.ConnectionMessages;
import org.apache.qpid.server.model.Port;
-import org.apache.qpid.server.model.Transport;
import org.apache.qpid.transport.Constant;
import org.apache.qpid.transport.Sender;
import org.apache.qpid.transport.network.Assembler;
@@ -57,14 +56,10 @@ public class ProtocolEngine_0_10 extend
private long _lastWriteTime;
public ProtocolEngine_0_10(ServerConnection conn,
- NetworkConnection network,
- Port port,
- Transport transport)
+ NetworkConnection network)
{
super(new Assembler(conn));
_connection = conn;
- _connection.setPort(port);
- _connection.setTransport(transport);
if(network != null)
{
Modified: qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java?rev=1633466&r1=1633465&r2=1633466&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java (original)
+++ qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java Tue Oct 21 21:11:31 2014
@@ -42,8 +42,8 @@ import org.apache.qpid.server.logging.Ev
import org.apache.qpid.server.logging.LogSubject;
import org.apache.qpid.server.logging.messages.ConnectionMessages;
import org.apache.qpid.server.model.Broker;
-import org.apache.qpid.server.model.Port;
import org.apache.qpid.server.model.Transport;
+import org.apache.qpid.server.model.port.AmqpPort;
import org.apache.qpid.server.protocol.AMQConnectionModel;
import org.apache.qpid.server.protocol.AMQSessionModel;
import org.apache.qpid.server.protocol.SessionModelListener;
@@ -75,7 +75,7 @@ public class ServerConnection extends Co
private final long _connectionId;
private final Object _reference = new Object();
private VirtualHostImpl<?,?,?> _virtualHost;
- private Port<?> _port;
+ private AmqpPort<?> _port;
private AtomicLong _lastIoTime = new AtomicLong();
private boolean _blocking;
private Transport _transport;
@@ -88,13 +88,24 @@ public class ServerConnection extends Co
private volatile boolean _stopped;
private int _messageCompressionThreshold;
+ private int _maxMessageSize;
- public ServerConnection(final long connectionId, Broker broker)
+ public ServerConnection(final long connectionId,
+ Broker<?> broker,
+ final AmqpPort<?> port,
+ final Transport transport)
{
_connectionId = connectionId;
_authorizedSubject.getPrincipals().add(new ConnectionPrincipal(this));
_broker = broker;
+ _port = port;
+ _transport = transport;
+
+ int maxMessageSize = port.getContextValue(Integer.class, AmqpPort.PORT_MAX_MESSAGE_SIZE);
+ _maxMessageSize = (maxMessageSize > 0) ? maxMessageSize : Integer.MAX_VALUE;
+
+
_messagesDelivered = new StatisticsCounter("messages-delivered-" + getConnectionId());
_dataDelivered = new StatisticsCounter("data-delivered-" + getConnectionId());
_messagesReceived = new StatisticsCounter("messages-received-" + getConnectionId());
@@ -203,16 +214,11 @@ public class ServerConnection extends Co
}
@Override
- public Port<?> getPort()
+ public AmqpPort<?> getPort()
{
return _port;
}
- public void setPort(Port<?> port)
- {
- _port = port;
- }
-
@Override
public Transport getTransport()
{
@@ -231,11 +237,6 @@ public class ServerConnection extends Co
return _stopped;
}
- public void setTransport(Transport transport)
- {
- _transport = transport;
- }
-
public void onOpen(final Runnable task)
{
_onOpenTask = task;
@@ -658,4 +659,9 @@ public class ServerConnection extends Co
{
return _messageCompressionThreshold;
}
+
+ public int getMaxMessageSize()
+ {
+ return _maxMessageSize;
+ }
}
Modified: qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSessionDelegate.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSessionDelegate.java?rev=1633466&r1=1633465&r2=1633466&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSessionDelegate.java (original)
+++ qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSessionDelegate.java Tue Oct 21 21:11:31 2014
@@ -333,16 +333,22 @@ public class ServerSessionDelegate exten
@Override
public void messageTransfer(Session ssn, final MessageTransfer xfr)
{
- if(((ServerSession)ssn).blockingTimeoutExceeded())
+ ServerSession serverSession = (ServerSession) ssn;
+ if(serverSession.blockingTimeoutExceeded())
{
getVirtualHost(ssn).getEventLogger().message(ChannelMessages.FLOW_CONTROL_IGNORED());
- ((ServerSession) ssn).close(AMQConstant.MESSAGE_TOO_LARGE,
- "Session flow control was requested, but not enforced by sender");
+ serverSession.close(AMQConstant.MESSAGE_TOO_LARGE,
+ "Session flow control was requested, but not enforced by sender");
+ }
+ else if(xfr.getBodySize() > serverSession.getConnection().getMaxMessageSize())
+ {
+ exception(ssn, xfr, ExecutionErrorCode.RESOURCE_LIMIT_EXCEEDED,
+ "Message size of " + xfr.getBodySize() + " greater than allowed maximum of " + serverSession.getConnection().getMaxMessageSize());
}
else
{
- final MessageDestination exchange = getDestinationForMessage(ssn, xfr);
+ final MessageDestination destination = getDestinationForMessage(ssn, xfr);
final DeliveryProperties delvProps =
xfr.getHeader() == null ? null : xfr.getHeader().getDeliveryProperties();
@@ -359,7 +365,7 @@ public class ServerSessionDelegate exten
virtualHost.getSecurityManager()
.authorisePublish(messageMetaData.isImmediate(),
messageMetaData.getRoutingKey(),
- exchange.getName(),
+ destination.getName(),
virtualHost.getName());
}
catch (AccessControlException e)
@@ -372,7 +378,6 @@ public class ServerSessionDelegate exten
final MessageStore store = virtualHost.getMessageStore();
final StoredMessage<MessageMetaData_0_10> storeMessage = createStoreMessage(xfr, messageMetaData, store);
- final ServerSession serverSession = (ServerSession) ssn;
final MessageTransferMessage message =
new MessageTransferMessage(storeMessage, serverSession.getReference());
MessageReference<MessageTransferMessage> reference = message.newReference();
@@ -400,7 +405,7 @@ public class ServerSessionDelegate exten
}
};
- int enqueues = serverSession.enqueue(message, instanceProperties, exchange);
+ int enqueues = serverSession.enqueue(message, instanceProperties, destination);
if (enqueues == 0)
{
@@ -414,7 +419,7 @@ public class ServerSessionDelegate exten
}
else
{
- virtualHost.getEventLogger().message(ExchangeMessages.DISCARDMSG(exchange.getName(),
+ virtualHost.getEventLogger().message(ExchangeMessages.DISCARDMSG(destination.getName(),
messageMetaData.getRoutingKey()));
}
}
Modified: qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/test/java/org/apache/qpid/server/protocol/v0_10/ServerSessionTest.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/test/java/org/apache/qpid/server/protocol/v0_10/ServerSessionTest.java?rev=1633466&r1=1633465&r2=1633466&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/test/java/org/apache/qpid/server/protocol/v0_10/ServerSessionTest.java (original)
+++ qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/test/java/org/apache/qpid/server/protocol/v0_10/ServerSessionTest.java Tue Oct 21 21:11:31 2014
@@ -22,11 +22,20 @@ import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
+import java.util.ArrayList;
+import java.util.List;
+
import org.apache.qpid.server.model.Broker;
+import org.apache.qpid.server.model.Transport;
+import org.apache.qpid.server.model.port.AmqpPort;
import org.apache.qpid.server.util.BrokerTestHelper;
import org.apache.qpid.server.virtualhost.VirtualHostImpl;
import org.apache.qpid.test.utils.QpidTestCase;
import org.apache.qpid.transport.Binary;
+import org.apache.qpid.transport.ExecutionErrorCode;
+import org.apache.qpid.transport.ExecutionException;
+import org.apache.qpid.transport.MessageTransfer;
+import org.apache.qpid.transport.Method;
public class ServerSessionTest extends QpidTestCase
{
@@ -60,16 +69,19 @@ public class ServerSessionTest extends Q
public void testCompareTo() throws Exception
{
- final Broker broker = mock(Broker.class);
+ final Broker<?> broker = mock(Broker.class);
when(broker.getContextValue(eq(Long.class), eq(Broker.CHANNEL_FLOW_CONTROL_ENFORCEMENT_TIMEOUT))).thenReturn(0l);
- ServerConnection connection = new ServerConnection(1, broker);
+ AmqpPort amqpPort = mock(AmqpPort.class);
+ when(amqpPort.getContextValue(eq(Integer.class), eq(AmqpPort.PORT_MAX_MESSAGE_SIZE))).thenReturn(AmqpPort.DEFAULT_MAX_MESSAGE_SIZE);
+
+ ServerConnection connection = new ServerConnection(1, broker, amqpPort, Transport.TCP);
connection.setVirtualHost(_virtualHost);
ServerSession session1 = new ServerSession(connection, new ServerSessionDelegate(),
new Binary(getName().getBytes()), 0);
// create a session with the same name but on a different connection
- ServerConnection connection2 = new ServerConnection(2, broker);
+ ServerConnection connection2 = new ServerConnection(2, broker, amqpPort, Transport.TCP);
connection2.setVirtualHost(_virtualHost);
ServerSession session2 = new ServerSession(connection2, new ServerSessionDelegate(),
new Binary(getName().getBytes()), 0);
@@ -78,5 +90,46 @@ public class ServerSessionTest extends Q
assertEquals("Unexpected compare result", 0, session1.compareTo(session1));
}
+ public void testOverlargeMessageTest() throws Exception
+ {
+ final Broker<?> broker = mock(Broker.class);
+ when(broker.getContextValue(eq(Long.class), eq(Broker.CHANNEL_FLOW_CONTROL_ENFORCEMENT_TIMEOUT))).thenReturn(0l);
+
+ AmqpPort port = mock(AmqpPort.class);
+ when(port.getContextValue(eq(Integer.class), eq(AmqpPort.PORT_MAX_MESSAGE_SIZE))).thenReturn(1024);
+ ServerConnection connection = new ServerConnection(1, broker, port, Transport.TCP);
+ connection.setVirtualHost(_virtualHost);
+ final List<Method> invokedMethods = new ArrayList<>();
+ ServerSession session = new ServerSession(connection, new ServerSessionDelegate(),
+ new Binary(getName().getBytes()), 0)
+ {
+ @Override
+ public void invoke(final Method m)
+ {
+ invokedMethods.add(m);
+ }
+ };
+
+ ServerSessionDelegate delegate = new ServerSessionDelegate();
+
+ MessageTransfer xfr = new MessageTransfer();
+ xfr.setBody(new byte[2048]);
+ delegate.messageTransfer(session, xfr);
+
+ assertFalse("No methods invoked - expecting at least 1", invokedMethods.isEmpty());
+ Method firstInvoked = invokedMethods.get(0);
+ assertTrue("First invoked method not execution error", firstInvoked instanceof ExecutionException);
+ assertEquals(ExecutionErrorCode.RESOURCE_LIMIT_EXCEEDED, ((ExecutionException)firstInvoked).getErrorCode());
+
+ invokedMethods.clear();
+
+ // test the boundary condition
+
+ xfr.setBody(new byte[1024]);
+ delegate.messageTransfer(session, xfr);
+
+ assertTrue("Methods invoked when not expecting any", invokedMethods.isEmpty());
+ }
+
}
Modified: qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java?rev=1633466&r1=1633465&r2=1633466&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java (original)
+++ qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java Tue Oct 21 21:11:31 2014
@@ -207,7 +207,6 @@ public class AMQChannel
private boolean _confirmOnPublish;
private long _confirmedMessageCounter;
-
public AMQChannel(AMQProtocolEngine connection, int channelId, final MessageStore messageStore)
{
_connection = connection;
@@ -578,9 +577,17 @@ public class AMQChannel
try
{
- _currentMessage.addContentBodyFrame(contentBody);
-
- deliverCurrentMessageIfComplete();
+ long currentSize = _currentMessage.addContentBodyFrame(contentBody);
+ if(currentSize > _currentMessage.getSize())
+ {
+ _connection.closeConnection(AMQConstant.FRAME_ERROR,
+ "More message data received than content header defined",
+ _channelId);
+ }
+ else
+ {
+ deliverCurrentMessageIfComplete();
+ }
}
catch (RuntimeException e)
{
@@ -2385,6 +2392,11 @@ public class AMQChannel
if(hasCurrentMessage())
{
+ if(bodySize > _connection.getMaxMessageSize())
+ {
+ closeChannel(AMQConstant.MESSAGE_TOO_LARGE,
+ "Message size of " + bodySize + " greater than allowed maximum of " + _connection.getMaxMessageSize());
+ }
publishContentHeader(new ContentHeaderBody(properties, bodySize));
}
else
Modified: qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java?rev=1633466&r1=1633465&r2=1633466&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java (original)
+++ qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java Tue Oct 21 21:11:31 2014
@@ -102,7 +102,7 @@ public class AMQProtocolEngine implement
private static final int REUSABLE_BYTE_BUFFER_CAPACITY = 65 * 1024;
public static final String BROKER_DEBUG_BINARY_DATA_LENGTH = "broker.debug.binaryDataLength";
public static final int DEFAULT_DEBUG_BINARY_DATA_LENGTH = 80;
- private final Port<?> _port;
+ private final AmqpPort<?> _port;
private final long _creationTime;
private AMQShortString _contextKey;
@@ -184,11 +184,12 @@ public class AMQProtocolEngine implement
private int _currentClassId;
private int _currentMethodId;
private int _binaryDataLimit;
+ private long _maxMessageSize;
- public AMQProtocolEngine(Broker broker,
+ public AMQProtocolEngine(Broker<?> broker,
final NetworkConnection network,
final long connectionId,
- Port port,
+ AmqpPort<?> port,
Transport transport)
{
_broker = broker;
@@ -202,6 +203,10 @@ public class AMQProtocolEngine implement
_binaryDataLimit = _broker.getContextKeys(false).contains(BROKER_DEBUG_BINARY_DATA_LENGTH)
? _broker.getContextValue(Integer.class, BROKER_DEBUG_BINARY_DATA_LENGTH)
: DEFAULT_DEBUG_BINARY_DATA_LENGTH;
+
+ int maxMessageSize = port.getContextValue(Integer.class, AmqpPort.PORT_MAX_MESSAGE_SIZE);
+ _maxMessageSize = (maxMessageSize > 0) ? (long) maxMessageSize : Long.MAX_VALUE;
+
_authorizedSubject.getPrincipals().add(new ConnectionPrincipal(this));
runAsSubject(new PrivilegedAction<Void>()
{
@@ -1164,7 +1169,7 @@ public class AMQProtocolEngine implement
}
@Override
- public Port<?> getPort()
+ public AmqpPort<?> getPort()
{
return _port;
}
@@ -1753,6 +1758,11 @@ public class AMQProtocolEngine implement
return _binaryDataLimit;
}
+ public long getMaxMessageSize()
+ {
+ return _maxMessageSize;
+ }
+
public final class WriteDeliverMethod
implements ClientDeliveryMethod
{
Modified: qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/IncomingMessage.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/IncomingMessage.java?rev=1633466&r1=1633465&r2=1633466&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/IncomingMessage.java (original)
+++ qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/IncomingMessage.java Tue Oct 21 21:11:31 2014
@@ -57,10 +57,11 @@ public class IncomingMessage
return _messagePublishInfo;
}
- public void addContentBodyFrame(final ContentBody contentChunk)
+ public long addContentBodyFrame(final ContentBody contentChunk)
{
_bodyLengthReceived += contentChunk.getSize();
_contentChunks.add(contentChunk);
+ return _bodyLengthReceived;
}
public boolean allContentReceived()
Modified: qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ProtocolEngineCreator_0_8.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ProtocolEngineCreator_0_8.java?rev=1633466&r1=1633465&r2=1633466&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ProtocolEngineCreator_0_8.java (original)
+++ qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ProtocolEngineCreator_0_8.java Tue Oct 21 21:11:31 2014
@@ -22,9 +22,9 @@ package org.apache.qpid.server.protocol.
import org.apache.qpid.protocol.ServerProtocolEngine;
import org.apache.qpid.server.model.Broker;
-import org.apache.qpid.server.model.Port;
import org.apache.qpid.server.model.Protocol;
import org.apache.qpid.server.model.Transport;
+import org.apache.qpid.server.model.port.AmqpPort;
import org.apache.qpid.server.plugin.PluggableService;
import org.apache.qpid.server.plugin.ProtocolEngineCreator;
import org.apache.qpid.transport.network.NetworkConnection;
@@ -58,9 +58,9 @@ public class ProtocolEngineCreator_0_8 i
return AMQP_0_8_HEADER;
}
- public ServerProtocolEngine newProtocolEngine(Broker broker,
+ public ServerProtocolEngine newProtocolEngine(Broker<?> broker,
NetworkConnection network,
- Port port,
+ AmqpPort<?> port,
Transport transport,
long id)
{
Modified: qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ProtocolEngineCreator_0_9.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ProtocolEngineCreator_0_9.java?rev=1633466&r1=1633465&r2=1633466&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ProtocolEngineCreator_0_9.java (original)
+++ qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ProtocolEngineCreator_0_9.java Tue Oct 21 21:11:31 2014
@@ -22,9 +22,9 @@ package org.apache.qpid.server.protocol.
import org.apache.qpid.protocol.ServerProtocolEngine;
import org.apache.qpid.server.model.Broker;
-import org.apache.qpid.server.model.Port;
import org.apache.qpid.server.model.Protocol;
import org.apache.qpid.server.model.Transport;
+import org.apache.qpid.server.model.port.AmqpPort;
import org.apache.qpid.server.plugin.PluggableService;
import org.apache.qpid.server.plugin.ProtocolEngineCreator;
import org.apache.qpid.transport.network.NetworkConnection;
@@ -58,9 +58,9 @@ public class ProtocolEngineCreator_0_9 i
return AMQP_0_9_HEADER;
}
- public ServerProtocolEngine newProtocolEngine(Broker broker,
+ public ServerProtocolEngine newProtocolEngine(Broker<?> broker,
NetworkConnection network,
- Port port,
+ AmqpPort<?> port,
Transport transport,
long id)
{
Modified: qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ProtocolEngineCreator_0_9_1.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ProtocolEngineCreator_0_9_1.java?rev=1633466&r1=1633465&r2=1633466&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ProtocolEngineCreator_0_9_1.java (original)
+++ qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ProtocolEngineCreator_0_9_1.java Tue Oct 21 21:11:31 2014
@@ -22,9 +22,9 @@ package org.apache.qpid.server.protocol.
import org.apache.qpid.protocol.ServerProtocolEngine;
import org.apache.qpid.server.model.Broker;
-import org.apache.qpid.server.model.Port;
import org.apache.qpid.server.model.Protocol;
import org.apache.qpid.server.model.Transport;
+import org.apache.qpid.server.model.port.AmqpPort;
import org.apache.qpid.server.plugin.PluggableService;
import org.apache.qpid.server.plugin.ProtocolEngineCreator;
import org.apache.qpid.transport.network.NetworkConnection;
@@ -59,9 +59,9 @@ public class ProtocolEngineCreator_0_9_1
return AMQP_0_9_1_HEADER;
}
- public ServerProtocolEngine newProtocolEngine(Broker broker,
+ public ServerProtocolEngine newProtocolEngine(Broker<?> broker,
NetworkConnection network,
- Port port,
+ AmqpPort<?> port,
Transport transport,
long id)
{
Modified: qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AMQChannelTest.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AMQChannelTest.java?rev=1633466&r1=1633465&r2=1633466&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AMQChannelTest.java (original)
+++ qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AMQChannelTest.java Tue Oct 21 21:11:31 2014
@@ -20,20 +20,28 @@
*/
package org.apache.qpid.server.protocol.v0_8;
+import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
+import java.util.ArrayList;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
+import org.apache.qpid.framing.AMQDataBlock;
+import org.apache.qpid.framing.AMQFrame;
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.framing.BasicContentHeaderProperties;
+import org.apache.qpid.framing.ChannelCloseBody;
import org.apache.qpid.framing.ContentHeaderBody;
import org.apache.qpid.framing.MessagePublishInfo;
+import org.apache.qpid.protocol.AMQConstant;
import org.apache.qpid.server.configuration.BrokerProperties;
import org.apache.qpid.server.exchange.ExchangeImpl;
import org.apache.qpid.server.message.MessageContentSource;
import org.apache.qpid.server.model.Broker;
+import org.apache.qpid.server.model.port.AmqpPort;
import org.apache.qpid.server.util.BrokerTestHelper;
import org.apache.qpid.server.virtualhost.VirtualHostImpl;
import org.apache.qpid.test.utils.QpidTestCase;
@@ -52,7 +60,10 @@ public class AMQChannelTest extends Qpid
BrokerTestHelper.setUp();
_virtualHost = BrokerTestHelper.createVirtualHost(getTestName());
_broker = BrokerTestHelper.createBrokerMock();
- _protocolSession = new InternalTestProtocolSession(_virtualHost, _broker)
+ AmqpPort port = mock(AmqpPort.class);
+ when(port.getContextValue(eq(Integer.class), eq(AmqpPort.PORT_MAX_MESSAGE_SIZE))).thenReturn(AmqpPort.DEFAULT_MAX_MESSAGE_SIZE);
+
+ _protocolSession = new InternalTestProtocolSession(_virtualHost, _broker, port)
{
@Override
public void writeReturn(MessagePublishInfo messagePublishInfo,
@@ -86,8 +97,10 @@ public class AMQChannelTest extends Qpid
{
AMQChannel channel1 = new AMQChannel(_protocolSession, 1, _virtualHost.getMessageStore());
+ AmqpPort port = mock(AmqpPort.class);
+ when(port.getContextValue(eq(Integer.class), eq(AmqpPort.PORT_MAX_MESSAGE_SIZE))).thenReturn(AmqpPort.DEFAULT_MAX_MESSAGE_SIZE);
// create a channel with the same channelId but on a different session
- AMQChannel channel2 = new AMQChannel(new InternalTestProtocolSession(_virtualHost, _broker), 1, _virtualHost.getMessageStore());
+ AMQChannel channel2 = new AMQChannel(new InternalTestProtocolSession(_virtualHost, _broker, port), 1, _virtualHost.getMessageStore());
assertFalse("Unexpected compare result", channel1.compareTo(channel2) == 0);
assertEquals("Unexpected compare result", 0, channel1.compareTo(channel1));
}
@@ -135,4 +148,33 @@ public class AMQChannelTest extends Qpid
assertEquals("Unexpected number of replies", 0, _replies.size());
}
+ public void testOverlargeMessage() throws Exception
+ {
+
+ AmqpPort port = mock(AmqpPort.class);
+ when(port.getContextValue(eq(Integer.class), eq(AmqpPort.PORT_MAX_MESSAGE_SIZE))).thenReturn(1024);
+ final List<AMQDataBlock> frames = new ArrayList<>();
+ _protocolSession = new InternalTestProtocolSession(_virtualHost, _broker, port)
+ {
+ @Override
+ public synchronized void writeFrame(final AMQDataBlock frame)
+ {
+ frames.add(frame);
+ }
+ };
+
+ AMQChannel channel = new AMQChannel(_protocolSession, 1, _virtualHost.getMessageStore());
+
+ channel.receiveBasicPublish(AMQShortString.EMPTY_STRING, AMQShortString.EMPTY_STRING, false, false);
+
+ final BasicContentHeaderProperties properties = new BasicContentHeaderProperties();
+ channel.receiveMessageHeader(properties, 2048l);
+
+ frames.toString();
+
+ assertEquals(1, frames.size());
+ assertEquals(ChannelCloseBody.class, ((AMQFrame) frames.get(0)).getBodyFrame().getClass());
+ assertEquals(AMQConstant.MESSAGE_TOO_LARGE.getCode(), ((ChannelCloseBody)((AMQFrame)frames.get(0)).getBodyFrame()).getReplyCode());
+ }
+
}
Modified: qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngineTest.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngineTest.java?rev=1633466&r1=1633465&r2=1633466&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngineTest.java (original)
+++ qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngineTest.java Tue Oct 21 21:11:31 2014
@@ -20,6 +20,7 @@
*/
package org.apache.qpid.server.protocol.v0_8;
+import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -29,16 +30,16 @@ import java.util.Map;
import org.apache.qpid.framing.FieldTable;
import org.apache.qpid.properties.ConnectionStartProperties;
import org.apache.qpid.server.model.Broker;
-import org.apache.qpid.server.model.Port;
import org.apache.qpid.server.model.Transport;
+import org.apache.qpid.server.model.port.AmqpPort;
import org.apache.qpid.server.util.BrokerTestHelper;
import org.apache.qpid.test.utils.QpidTestCase;
import org.apache.qpid.transport.network.NetworkConnection;
public class AMQProtocolEngineTest extends QpidTestCase
{
- private Broker _broker;
- private Port _port;
+ private Broker<?> _broker;
+ private AmqpPort<?> _port;
private NetworkConnection _network;
private Transport _transport;
@@ -49,7 +50,9 @@ public class AMQProtocolEngineTest exten
_broker = BrokerTestHelper.createBrokerMock();
when(_broker.getConnection_closeWhenNoRoute()).thenReturn(true);
- _port = mock(Port.class);
+ _port = mock(AmqpPort.class);
+ when(_port.getContextValue(eq(Integer.class), eq(AmqpPort.PORT_MAX_MESSAGE_SIZE))).thenReturn(AmqpPort.DEFAULT_MAX_MESSAGE_SIZE);
+
_network = mock(NetworkConnection.class);
_transport = Transport.TCP;
}
Modified: qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/BrokerTestHelper_0_8.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/BrokerTestHelper_0_8.java?rev=1633466&r1=1633465&r2=1633466&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/BrokerTestHelper_0_8.java (original)
+++ qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/BrokerTestHelper_0_8.java Tue Oct 21 21:11:31 2014
@@ -20,12 +20,17 @@
*/
package org.apache.qpid.server.protocol.v0_8;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
import org.apache.qpid.AMQException;
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.framing.BasicContentHeaderProperties;
import org.apache.qpid.framing.ContentHeaderBody;
import org.apache.qpid.framing.MessagePublishInfo;
import org.apache.qpid.server.message.MessageDestination;
+import org.apache.qpid.server.model.port.AmqpPort;
import org.apache.qpid.server.util.BrokerTestHelper;
import org.apache.qpid.server.virtualhost.VirtualHostImpl;
@@ -58,7 +63,10 @@ public class BrokerTestHelper_0_8 extend
public static InternalTestProtocolSession createProtocolSession(String hostName) throws Exception
{
VirtualHostImpl virtualHost = createVirtualHost(hostName);
- return new InternalTestProtocolSession(virtualHost, createBrokerMock());
+
+ AmqpPort port = mock(AmqpPort.class);
+ when(port.getContextValue(eq(Integer.class), eq(AmqpPort.PORT_MAX_MESSAGE_SIZE))).thenReturn(AmqpPort.DEFAULT_MAX_MESSAGE_SIZE);
+ return new InternalTestProtocolSession(virtualHost, createBrokerMock(), port);
}
public static void publishMessages(AMQChannel channel, int numberOfMessages, String queueName, String exchangeName)
Modified: qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/InternalTestProtocolSession.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/InternalTestProtocolSession.java?rev=1633466&r1=1633465&r2=1633466&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/InternalTestProtocolSession.java (original)
+++ qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/InternalTestProtocolSession.java Tue Oct 21 21:11:31 2014
@@ -46,6 +46,7 @@ import org.apache.qpid.server.message.In
import org.apache.qpid.server.message.MessageContentSource;
import org.apache.qpid.server.message.ServerMessage;
import org.apache.qpid.server.model.Broker;
+import org.apache.qpid.server.model.port.AmqpPort;
import org.apache.qpid.server.security.auth.AuthenticatedPrincipal;
import org.apache.qpid.server.security.auth.UsernamePrincipal;
import org.apache.qpid.server.virtualhost.VirtualHostImpl;
@@ -60,9 +61,9 @@ public class InternalTestProtocolSession
private AtomicInteger _deliveryCount = new AtomicInteger(0);
private static final AtomicLong ID_GENERATOR = new AtomicLong(0);
- public InternalTestProtocolSession(VirtualHostImpl virtualHost, Broker broker) throws AMQException
+ public InternalTestProtocolSession(VirtualHostImpl virtualHost, Broker<?> broker, final AmqpPort<?> port) throws AMQException
{
- super(broker, new TestNetworkConnection(), ID_GENERATOR.getAndIncrement(), null, null);
+ super(broker, new TestNetworkConnection(), ID_GENERATOR.getAndIncrement(), port, null);
_channelDelivers = new HashMap<Integer, Map<String, LinkedList<DeliveryPair>>>();
Modified: qpid/trunk/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java?rev=1633466&r1=1633465&r2=1633466&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java (original)
+++ qpid/trunk/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java Tue Oct 21 21:11:31 2014
@@ -47,7 +47,6 @@ import org.apache.qpid.protocol.AMQConst
import org.apache.qpid.server.connection.ConnectionPrincipal;
import org.apache.qpid.server.logging.LogSubject;
import org.apache.qpid.server.model.Broker;
-import org.apache.qpid.server.model.Port;
import org.apache.qpid.server.model.Transport;
import org.apache.qpid.server.model.port.AmqpPort;
import org.apache.qpid.server.protocol.AMQConnectionModel;
@@ -62,8 +61,8 @@ import org.apache.qpid.server.virtualhos
public class Connection_1_0 implements ConnectionEventListener, AMQConnectionModel<Connection_1_0,Session_1_0>
{
- private final Port<?> _port;
- private final Broker _broker;
+ private final AmqpPort<?> _port;
+ private final Broker<?> _broker;
private final SubjectCreator _subjectCreator;
private VirtualHostImpl _vhost;
private final Transport _transport;
@@ -102,10 +101,10 @@ public class Connection_1_0 implements C
private boolean _closedOnOpen;
- public Connection_1_0(Broker broker,
+ public Connection_1_0(Broker<?> broker,
ConnectionEndpoint conn,
long connectionId,
- Port port,
+ AmqpPort<?> port,
Transport transport, final SubjectCreator subjectCreator)
{
_broker = broker;
@@ -359,7 +358,7 @@ public class Connection_1_0 implements C
}
@Override
- public Port<?> getPort()
+ public AmqpPort<?> getPort()
{
return _port;
}
Modified: qpid/trunk/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ProtocolEngineCreator_1_0_0_SASL.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ProtocolEngineCreator_1_0_0_SASL.java?rev=1633466&r1=1633465&r2=1633466&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ProtocolEngineCreator_1_0_0_SASL.java (original)
+++ qpid/trunk/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ProtocolEngineCreator_1_0_0_SASL.java Tue Oct 21 21:11:31 2014
@@ -22,9 +22,9 @@ package org.apache.qpid.server.protocol.
import org.apache.qpid.protocol.ServerProtocolEngine;
import org.apache.qpid.server.model.Broker;
-import org.apache.qpid.server.model.Port;
import org.apache.qpid.server.model.Protocol;
import org.apache.qpid.server.model.Transport;
+import org.apache.qpid.server.model.port.AmqpPort;
import org.apache.qpid.server.plugin.PluggableService;
import org.apache.qpid.server.plugin.ProtocolEngineCreator;
import org.apache.qpid.transport.network.NetworkConnection;
@@ -58,9 +58,9 @@ public class ProtocolEngineCreator_1_0_0
return AMQP_SASL_1_0_0_HEADER;
}
- public ServerProtocolEngine newProtocolEngine(Broker broker,
+ public ServerProtocolEngine newProtocolEngine(Broker<?> broker,
NetworkConnection network,
- Port port,
+ AmqpPort<?> port,
Transport transport,
long id)
{
Modified: qpid/trunk/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ProtocolEngine_1_0_0_SASL.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ProtocolEngine_1_0_0_SASL.java?rev=1633466&r1=1633465&r2=1633466&view=diff
==============================================================================
--- qpid/trunk/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ProtocolEngine_1_0_0_SASL.java (original)
+++ qpid/trunk/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ProtocolEngine_1_0_0_SASL.java Tue Oct 21 21:11:31 2014
@@ -35,6 +35,7 @@ import javax.security.sasl.SaslException
import javax.security.sasl.SaslServer;
import org.apache.log4j.Logger;
+
import org.apache.qpid.amqp_1_0.codec.FrameWriter;
import org.apache.qpid.amqp_1_0.codec.ProtocolHandler;
import org.apache.qpid.amqp_1_0.framing.AMQFrame;
@@ -53,8 +54,8 @@ import org.apache.qpid.common.QpidProper
import org.apache.qpid.common.ServerPropertyNames;
import org.apache.qpid.protocol.ServerProtocolEngine;
import org.apache.qpid.server.model.Broker;
-import org.apache.qpid.server.model.Port;
import org.apache.qpid.server.model.Transport;
+import org.apache.qpid.server.model.port.AmqpPort;
import org.apache.qpid.server.security.SubjectCreator;
import org.apache.qpid.server.security.auth.UsernamePrincipal;
import org.apache.qpid.server.util.ServerScopedRuntimeException;
@@ -67,14 +68,14 @@ public class ProtocolEngine_1_0_0_SASL i
private static final org.apache.log4j.Logger
_logger = org.apache.log4j.Logger.getLogger(ProtocolEngine_1_0_0_SASL.class);
- private final Port _port;
+ private final AmqpPort<?> _port;
private final Transport _transport;
private long _readBytes;
private long _writtenBytes;
private long _lastReadTime;
private long _lastWriteTime;
- private final Broker _broker;
+ private final Broker<?> _broker;
private long _createTime = System.currentTimeMillis();
private ConnectionEndpoint _endpoint;
private long _connectionId;
@@ -134,8 +135,8 @@ public class ProtocolEngine_1_0_0_SASL i
private State _state = State.A;
- public ProtocolEngine_1_0_0_SASL(final NetworkConnection networkDriver, final Broker broker,
- long id, Port port, Transport transport)
+ public ProtocolEngine_1_0_0_SASL(final NetworkConnection networkDriver, final Broker<?> broker,
+ long id, AmqpPort<?> port, Transport transport)
{
_connectionId = id;
_broker = broker;
Modified: qpid/trunk/qpid/java/systests/src/test/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngineFactoryTest.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/systests/src/test/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngineFactoryTest.java?rev=1633466&r1=1633465&r2=1633466&view=diff
==============================================================================
--- qpid/trunk/qpid/java/systests/src/test/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngineFactoryTest.java (original)
+++ qpid/trunk/qpid/java/systests/src/test/java/org/apache/qpid/server/protocol/MultiVersionProtocolEngineFactoryTest.java Tue Oct 21 21:11:31 2014
@@ -157,6 +157,7 @@ public class MultiVersionProtocolEngineF
AmqpPort<?> port = mock(AmqpPort.class);
when(port.canAcceptNewConnection(any(SocketAddress.class))).thenReturn(true);
+ when(port.getContextValue(eq(Integer.class), eq(AmqpPort.PORT_MAX_MESSAGE_SIZE))).thenReturn(AmqpPort.DEFAULT_MAX_MESSAGE_SIZE);
when(port.getContextValue(eq(Long.class),eq(Port.CONNECTION_MAXIMUM_AUTHENTICATION_DELAY))).thenReturn(10000l);
MultiVersionProtocolEngineFactory factory =
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org