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 2015/09/04 18:32:59 UTC
svn commit: r1701289 - in /qpid/java/trunk:
broker-core/src/main/java/org/apache/qpid/server/model/
broker-core/src/main/java/org/apache/qpid/server/model/adapter/
broker-core/src/main/java/org/apache/qpid/server/model/port/
broker-core/src/main/java/o...
Author: orudyy
Date: Fri Sep 4 16:32:59 2015
New Revision: 1701289
URL: http://svn.apache.org/r1701289
Log:
QPID-6727: Introduce context variable for network buffer size, use it to set max frame size, send and receive buffer sizes and remove port attributes for send and receive buffers sizes
(work by Lorenz Quack <qu...@gmail.com> and Alex Rudyy <or...@apache.org>)
Modified:
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/Broker.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/adapter/BrokerAdapter.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/port/AmqpPort.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/port/AmqpPortImpl.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/store/BrokerStoreUpgraderAndRecoverer.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnection.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionPlainDelegate.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionTLSDelegate.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingNetworkTransport.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/model/adapter/BrokerAdapterTest.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/model/adapter/PortFactoryTest.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/store/BrokerStoreUpgraderAndRecovererTest.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/transport/NetworkConnectionSchedulerTest.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/transport/TCPandSSLTransportTest.java
qpid/java/trunk/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnectionDelegate.java
qpid/java/trunk/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQPConnection_0_8.java
qpid/java/trunk/systests/src/main/java/org/apache/qpid/systest/rest/Asserts.java
qpid/java/trunk/systests/src/test/java/org/apache/qpid/transport/MaxFrameSizeTest.java
Modified: qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/Broker.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/Broker.java?rev=1701289&r1=1701288&r2=1701289&view=diff
==============================================================================
--- qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/Broker.java (original)
+++ qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/Broker.java Fri Sep 4 16:32:59 2015
@@ -61,6 +61,19 @@ public interface Broker<X extends Broker
String QPID_RMI_PORT = "qpid.rmi_port";
String QPID_JMX_PORT = "qpid.jmx_port";
+ String NETWORK_BUFFER_SIZE = "qpid.broker.networkBufferSize";
+ // network buffer should at least hold a SSL/TLS frame which consists of
+ // 5 bytes SSLv3 record header
+ // 256 bytes of IV
+ // 16384 bytes (2^14 bytes) of data
+ // 256 bytes for block cipher padding
+ // 20 bytes for trailer SHA1 hash size
+ // 16384 bytes (2^14 bytes) of extra for large data packets
+ // 33305 bytes in TOTAL
+ int MINIMUM_NETWORK_BUFFER_SIZE = 33305;
+ @ManagedContextDefault(name = NETWORK_BUFFER_SIZE)
+ int DEFAULT_NETWORK_BUFFER_SIZE = 64 * 1024;
+
@ManagedContextDefault(name = "broker.name")
String DEFAULT_BROKER_NAME = "Broker";
@@ -82,13 +95,10 @@ public interface Broker<X extends Broker
@ManagedContextDefault(name = STORE_FILESYSTEM_MAX_USAGE_PERCENT)
int DEFAULT_FILESYSTEM_MAX_USAGE_PERCENT = 90;
- String BROKER_FRAME_SIZE = "qpid.broker_frame_size";
- @ManagedContextDefault(name = BROKER_FRAME_SIZE)
- int DEFAULT_FRAME_SIZE = 65535;
-
@ManagedContextDefault(name = BROKER_FAIL_STARTUP_WITH_ERRORED_CHILD)
boolean DEFAULT_BROKER_FAIL_STARTUP_WITH_ERRORED_CHILD = false;
+
@DerivedAttribute
String getBuildVersion();
@@ -193,4 +203,5 @@ public interface Broker<X extends Broker
void assignTargetSizes();
+ int getNetworkBufferSize();
}
Modified: qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/adapter/BrokerAdapter.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/adapter/BrokerAdapter.java?rev=1701289&r1=1701288&r2=1701289&view=diff
==============================================================================
--- qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/adapter/BrokerAdapter.java (original)
+++ qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/adapter/BrokerAdapter.java Fri Sep 4 16:32:59 2015
@@ -110,6 +110,7 @@ public class BrokerAdapter extends Abstr
private final boolean _virtualHostPropertiesNodeEnabled;
private Collection<BrokerLogger> _brokerLoggersToClose;
+ private int _networkBufferSize = DEFAULT_NETWORK_BUFFER_SIZE;
@ManagedObjectFactoryConstructor
public BrokerAdapter(Map<String, Object> attributes,
@@ -169,6 +170,13 @@ public class BrokerAdapter extends Abstr
{
super.postResolveChildren();
+ Integer networkBufferSize = getContextValue(Integer.class, NETWORK_BUFFER_SIZE);
+ if (networkBufferSize == null || networkBufferSize < MINIMUM_NETWORK_BUFFER_SIZE)
+ {
+ throw new IllegalConfigurationException(NETWORK_BUFFER_SIZE + " is set to unacceptable value '" +
+ networkBufferSize + "'. Must be larger than " + MINIMUM_NETWORK_BUFFER_SIZE + ".");
+ }
+ _networkBufferSize = networkBufferSize;
ch.qos.logback.classic.Logger rootLogger =
(ch.qos.logback.classic.Logger) LoggerFactory.getLogger(Logger.ROOT_LOGGER_NAME);
@@ -883,4 +891,10 @@ public class BrokerAdapter extends Abstr
}
return maxMemory;
}
+
+ @Override
+ public int getNetworkBufferSize()
+ {
+ return _networkBufferSize;
+ }
}
Modified: qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/port/AmqpPort.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/port/AmqpPort.java?rev=1701289&r1=1701288&r2=1701289&view=diff
==============================================================================
--- qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/port/AmqpPort.java (original)
+++ qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/port/AmqpPort.java Fri Sep 4 16:32:59 2015
@@ -42,15 +42,11 @@ import javax.net.ssl.SSLContext;
@ManagedObject( category = false, type = "AMQP")
public interface AmqpPort<X extends AmqpPort<X>> extends ClientAuthCapablePort<X>, PortWithThreadPool
{
- String DEFAULT_AMQP_SEND_BUFFER_SIZE = "262144";
- String DEFAULT_AMQP_RECEIVE_BUFFER_SIZE = "262144";
String DEFAULT_AMQP_TCP_NO_DELAY = "true";
String DEFAULT_AMQP_NEED_CLIENT_AUTH = "false";
String DEFAULT_AMQP_WANT_CLIENT_AUTH = "false";
- String SEND_BUFFER_SIZE = "sendBufferSize";
- String RECEIVE_BUFFER_SIZE = "receiveBufferSize";
String MAX_OPEN_CONNECTIONS = "maxOpenConnections";
@@ -94,12 +90,6 @@ public interface AmqpPort<X extends Amqp
@ManagedAttribute( defaultValue = AmqpPort.DEFAULT_AMQP_TCP_NO_DELAY )
boolean isTcpNoDelay();
- @ManagedAttribute( defaultValue = AmqpPort.DEFAULT_AMQP_SEND_BUFFER_SIZE )
- int getSendBufferSize();
-
- @ManagedAttribute( defaultValue = AmqpPort.DEFAULT_AMQP_RECEIVE_BUFFER_SIZE )
- int getReceiveBufferSize();
-
@Override
@ManagedAttribute( defaultValue = "${" + PORT_AMQP_THREAD_POOL_MAXIMUM + "}")
int getThreadPoolMaximum();
@@ -143,4 +133,6 @@ public interface AmqpPort<X extends Amqp
int decrementConnectionCount();
VirtualHostAlias createVirtualHostAlias(Map<String, Object> attributes);
+
+ int getNetworkBufferSize();
}
Modified: qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/port/AmqpPortImpl.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/port/AmqpPortImpl.java?rev=1701289&r1=1701288&r2=1701289&view=diff
==============================================================================
--- qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/port/AmqpPortImpl.java (original)
+++ qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/port/AmqpPortImpl.java Fri Sep 4 16:32:59 2015
@@ -109,12 +109,6 @@ public class AmqpPortImpl extends Abstra
private boolean _tcpNoDelay;
@ManagedAttributeField
- private int _sendBufferSize;
-
- @ManagedAttributeField
- private int _receiveBufferSize;
-
- @ManagedAttributeField
private String _bindingAddress;
@ManagedAttributeField
@@ -173,18 +167,6 @@ public class AmqpPortImpl extends Abstra
}
@Override
- public int getSendBufferSize()
- {
- return _sendBufferSize;
- }
-
- @Override
- public int getReceiveBufferSize()
- {
- return _receiveBufferSize;
- }
-
- @Override
public int getMaxOpenConnections()
{
return _maxOpenConnections;
@@ -321,6 +303,12 @@ public class AmqpPortImpl extends Abstra
return child;
}
+ @Override
+ public int getNetworkBufferSize()
+ {
+ return _broker.getNetworkBufferSize();
+ }
+
private VirtualHostAlias addVirtualHostAlias(Map<String,Object> attributes)
{
return getObjectFactory().create(VirtualHostAlias.class, attributes, this);
Modified: qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/store/BrokerStoreUpgraderAndRecoverer.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/store/BrokerStoreUpgraderAndRecoverer.java?rev=1701289&r1=1701288&r2=1701289&view=diff
==============================================================================
--- qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/store/BrokerStoreUpgraderAndRecoverer.java (original)
+++ qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/store/BrokerStoreUpgraderAndRecoverer.java Fri Sep 4 16:32:59 2015
@@ -366,6 +366,18 @@ public class BrokerStoreUpgraderAndRecov
getNextUpgrader().configuredObject(record);
}
+ else if (record.getType().equals("Port") && "AMQP".equals(record.getAttributes().get("type")))
+ {
+ Map<String, Object> updatedAttributes = new HashMap<>(record.getAttributes());
+ if (updatedAttributes.containsKey("receiveBufferSize") || updatedAttributes.containsKey("sendBufferSize"))
+ {
+ updatedAttributes.remove("receiveBufferSize");
+ updatedAttributes.remove("sendBufferSize");
+ record = new ConfiguredObjectRecordImpl(record.getId(), record.getType(), updatedAttributes, record.getParents());
+ getUpdateMap().put(record.getId(), record);
+ getNextUpgrader().configuredObject(record);
+ }
+ }
}
private void addLogger(final ConfiguredObjectRecord record, String name, String type)
Modified: qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnection.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnection.java?rev=1701289&r1=1701288&r2=1701289&view=diff
==============================================================================
--- qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnection.java (original)
+++ qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnection.java Fri Sep 4 16:32:59 2015
@@ -22,7 +22,6 @@ package org.apache.qpid.server.transport
import java.io.IOException;
import java.net.SocketAddress;
-import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.security.Principal;
import java.security.cert.Certificate;
@@ -56,7 +55,6 @@ public class NonBlockingConnection imple
private final AtomicBoolean _closed = new AtomicBoolean(false);
private final ProtocolEngine _protocolEngine;
private final Runnable _onTransportEncryptionAction;
- private final int _receiveBufferSize;
private volatile int _maxReadIdle;
private volatile int _maxWriteIdle;
@@ -69,7 +67,6 @@ public class NonBlockingConnection imple
public NonBlockingConnection(SocketChannel socketChannel,
ProtocolEngine protocolEngine,
- int receiveBufferSize,
final Set<TransportEncryption> encryptionSet,
final Runnable onTransportEncryptionAction,
final NetworkConnectionScheduler scheduler,
@@ -81,8 +78,6 @@ public class NonBlockingConnection imple
_protocolEngine = protocolEngine;
_onTransportEncryptionAction = onTransportEncryptionAction;
- _receiveBufferSize = receiveBufferSize;
-
_remoteSocketAddress = _socketChannel.socket().getRemoteSocketAddress().toString();
_port = port;
@@ -111,11 +106,6 @@ public class NonBlockingConnection imple
return _partialRead;
}
- int getReceiveBufferSize()
- {
- return _receiveBufferSize;
- }
-
Ticker getTicker()
{
return _protocolEngine.getAggregateTicker();
@@ -126,11 +116,6 @@ public class NonBlockingConnection imple
return _socketChannel;
}
- public AmqpPort getPort()
- {
- return _port;
- }
-
@Override
public void start()
{
@@ -464,7 +449,7 @@ public class NonBlockingConnection imple
_delegate = new NonBlockingConnectionTLSDelegate(this, _port);
break;
case NONE:
- _delegate = new NonBlockingConnectionPlainDelegate(this);
+ _delegate = new NonBlockingConnectionPlainDelegate(this, _port);
break;
default:
throw new IllegalArgumentException("unknown TransportEncryption " + transportEncryption);
Modified: qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionPlainDelegate.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionPlainDelegate.java?rev=1701289&r1=1701288&r2=1701289&view=diff
==============================================================================
--- qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionPlainDelegate.java (original)
+++ qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionPlainDelegate.java Fri Sep 4 16:32:59 2015
@@ -24,6 +24,7 @@ import java.security.Principal;
import java.security.cert.Certificate;
import java.util.Collection;
+import org.apache.qpid.server.model.port.AmqpPort;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -35,14 +36,15 @@ public class NonBlockingConnectionPlainD
private static final Logger LOGGER = LoggerFactory.getLogger(NonBlockingConnectionPlainDelegate.class);
private final NonBlockingConnection _parent;
+ private final int _networkBufferSize;
private volatile QpidByteBuffer _netInputBuffer;
- public NonBlockingConnectionPlainDelegate(NonBlockingConnection parent)
+ public NonBlockingConnectionPlainDelegate(NonBlockingConnection parent, AmqpPort<?> port)
{
_parent = parent;
- final int receiveBufferSize = parent.getReceiveBufferSize();
- QpidByteBufferUtils.createPool(parent.getPort(), receiveBufferSize);
- _netInputBuffer = QpidByteBuffer.allocateDirect(receiveBufferSize);
+ _networkBufferSize = port.getNetworkBufferSize();
+ QpidByteBufferUtils.createPool(port, _networkBufferSize);
+ _netInputBuffer = QpidByteBuffer.allocateDirect(_networkBufferSize);
}
@Override
@@ -78,9 +80,9 @@ public class NonBlockingConnectionPlainD
else
{
QpidByteBuffer currentBuffer = _netInputBuffer;
- int newBufSize = (currentBuffer.capacity() < _parent.getReceiveBufferSize())
- ? _parent.getReceiveBufferSize()
- : currentBuffer.capacity() + _parent.getReceiveBufferSize();
+ int newBufSize = (currentBuffer.capacity() < _networkBufferSize)
+ ? _networkBufferSize
+ : currentBuffer.capacity() + _networkBufferSize;
_netInputBuffer = QpidByteBuffer.allocateDirect(newBufSize);
_netInputBuffer.put(currentBuffer);
Modified: qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionTLSDelegate.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionTLSDelegate.java?rev=1701289&r1=1701288&r2=1701289&view=diff
==============================================================================
--- qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionTLSDelegate.java (original)
+++ qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnectionTLSDelegate.java Fri Sep 4 16:32:59 2015
@@ -22,6 +22,7 @@ package org.apache.qpid.server.transport
import org.apache.qpid.bytebuffer.QpidByteBuffer;
import org.apache.qpid.server.model.port.AmqpPort;
import org.apache.qpid.server.util.QpidByteBufferUtils;
+import org.apache.qpid.server.util.ServerScopedRuntimeException;
import org.apache.qpid.transport.network.security.ssl.SSLUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -46,6 +47,7 @@ public class NonBlockingConnectionTLSDel
private final SSLEngine _sslEngine;
private final NonBlockingConnection _parent;
private final int _initialApplicationBufferSize;
+ private final int _networkBufferSize;
private SSLEngineResult _status;
private final List<QpidByteBuffer> _encryptedOutput = new ArrayList<>();
private Principal _principal;
@@ -59,24 +61,21 @@ public class NonBlockingConnectionTLSDel
{
_parent = parent;
_sslEngine = createSSLEngine(port);
+ _networkBufferSize = port.getNetworkBufferSize();
-
- final int bufSize = Math.max(parent.getReceiveBufferSize(),
- _sslEngine.getSession().getPacketBufferSize());
- QpidByteBufferUtils.createPool(port, bufSize);
- _netInputBuffer = QpidByteBuffer.allocateDirect(bufSize);
-
- _initialApplicationBufferSize =
- Math.max(_sslEngine.getSession().getApplicationBufferSize() + 50, _parent.getReceiveBufferSize());
-
-
+ final int tlsPacketBufferSize = _sslEngine.getSession().getPacketBufferSize();
+ if (tlsPacketBufferSize > _networkBufferSize)
+ {
+ throw new ServerScopedRuntimeException("TLS implementation packet buffer size (" + tlsPacketBufferSize
+ + ") is greater then broker network buffer size (" + _networkBufferSize + ")");
+ }
+
+ QpidByteBufferUtils.createPool(port, _networkBufferSize);
+ _netInputBuffer = QpidByteBuffer.allocateDirect(_networkBufferSize);
+ _initialApplicationBufferSize = Math.max(_sslEngine.getSession().getApplicationBufferSize() + 50, _networkBufferSize);
QpidByteBufferUtils.createPool(port, _initialApplicationBufferSize);
-
_applicationBuffer = QpidByteBuffer.allocateDirect(_initialApplicationBufferSize);
-
- QpidByteBufferUtils.createPool(port, _sslEngine.getSession().getPacketBufferSize());
-
}
@Override
Modified: qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingNetworkTransport.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingNetworkTransport.java?rev=1701289&r1=1701288&r2=1701289&view=diff
==============================================================================
--- qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingNetworkTransport.java (original)
+++ qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingNetworkTransport.java Fri Sep 4 16:32:59 2015
@@ -141,11 +141,10 @@ public class NonBlockingNetworkTransport
socketChannel.setOption(StandardSocketOptions.TCP_NODELAY, _port.isTcpNoDelay());
socketChannel.socket().setSoTimeout(1000 * HANDSHAKE_TIMEOUT);
- final int sendBufferSize = _port.getSendBufferSize();
- final int receiveBufferSize = _port.getReceiveBufferSize();
+ final int bufferSize = _port.getNetworkBufferSize();
- socketChannel.setOption(StandardSocketOptions.SO_SNDBUF, sendBufferSize);
- socketChannel.setOption(StandardSocketOptions.SO_RCVBUF, receiveBufferSize);
+ socketChannel.setOption(StandardSocketOptions.SO_SNDBUF, bufferSize);
+ socketChannel.setOption(StandardSocketOptions.SO_RCVBUF, bufferSize);
socketChannel.configureBlocking(false);
@@ -157,9 +156,8 @@ public class NonBlockingNetworkTransport
NonBlockingConnection connection =
new NonBlockingConnection(socketChannel,
engine,
- receiveBufferSize,
_encryptionSet,
- new Runnable()
+ new Runnable()
{
@Override
Modified: qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/model/adapter/BrokerAdapterTest.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/model/adapter/BrokerAdapterTest.java?rev=1701289&r1=1701288&r2=1701289&view=diff
==============================================================================
--- qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/model/adapter/BrokerAdapterTest.java (original)
+++ qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/model/adapter/BrokerAdapterTest.java Fri Sep 4 16:32:59 2015
@@ -31,6 +31,7 @@ import org.apache.qpid.server.configurat
import org.apache.qpid.server.logging.EventLogger;
import org.apache.qpid.server.model.Broker;
import org.apache.qpid.server.model.BrokerModel;
+import org.apache.qpid.server.model.State;
import org.apache.qpid.server.model.SystemConfig;
import org.apache.qpid.server.model.VirtualHost;
import org.apache.qpid.server.model.VirtualHostNode;
@@ -100,6 +101,30 @@ public class BrokerAdapterTest extends Q
flowToDiskThreshold);
}
+ public void testNetworkBufferSize()
+ {
+ Map<String, Object> attributes = new HashMap<>();
+ attributes.put(Broker.NAME, "Broker");
+ attributes.put(Broker.MODEL_VERSION, BrokerModel.MODEL_VERSION);
+ attributes.put(Broker.DURABLE, true);
+ attributes.put(Broker.CONTEXT, Collections.singletonMap(Broker.NETWORK_BUFFER_SIZE, Broker.MINIMUM_NETWORK_BUFFER_SIZE + 1));
+
+ // testing successful case
+ BrokerAdapter brokerAdapter = new BrokerAdapter(attributes, _systemConfig);
+ brokerAdapter.open();
+ assertEquals("Broker open should be successful", State.ACTIVE, brokerAdapter.getState());
+ assertEquals("Unexpected buffer size", Broker.MINIMUM_NETWORK_BUFFER_SIZE + 1, brokerAdapter.getNetworkBufferSize());
+ brokerAdapter.close();
+
+ // testing unsuccessful case
+ attributes.put(Broker.CONTEXT, Collections.singletonMap(Broker.NETWORK_BUFFER_SIZE, Broker.MINIMUM_NETWORK_BUFFER_SIZE - 1));
+ brokerAdapter = new BrokerAdapter(attributes, _systemConfig);
+ brokerAdapter.open();
+ assertEquals("Broker open should fail with network buffer size less then minimum", State.ERRORED, brokerAdapter.getState());
+ assertEquals("Unexpected buffer size", Broker.DEFAULT_NETWORK_BUFFER_SIZE, brokerAdapter.getNetworkBufferSize());
+ brokerAdapter.close();
+ }
+
private void doAssignTargetSizeTest(final long[] virtualHostQueueSizes, final long flowToDiskThreshold)
{
Map<String, Object> attributes = new HashMap<>();
Modified: qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/model/adapter/PortFactoryTest.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/model/adapter/PortFactoryTest.java?rev=1701289&r1=1701288&r2=1701289&view=diff
==============================================================================
--- qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/model/adapter/PortFactoryTest.java (original)
+++ qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/model/adapter/PortFactoryTest.java Fri Sep 4 16:32:59 2015
@@ -111,8 +111,6 @@ public class PortFactoryTest extends Qpi
_attributes.put(Port.TRANSPORTS, _tcpStringSet);
_attributes.put(Port.AUTHENTICATION_PROVIDER, _authProviderName);
_attributes.put(Port.TCP_NO_DELAY, "true");
- _attributes.put(AmqpPort.RECEIVE_BUFFER_SIZE, "1");
- _attributes.put(AmqpPort.SEND_BUFFER_SIZE, "2");
_attributes.put(Port.BINDING_ADDRESS, "127.0.0.1");
}
@@ -145,10 +143,6 @@ public class PortFactoryTest extends Qpi
assertTrue(_port instanceof AmqpPort);
assertEquals("Unexpected _port", _portNumber, _port.getPort());
assertEquals("Unexpected transports", Collections.singleton(PortFactory.DEFAULT_TRANSPORT), _port.getTransports());
- assertEquals("Unexpected send buffer size", PortFactory.DEFAULT_AMQP_SEND_BUFFER_SIZE,
- _port.getAttribute(AmqpPort.SEND_BUFFER_SIZE));
- assertEquals("Unexpected receive buffer size", PortFactory.DEFAULT_AMQP_RECEIVE_BUFFER_SIZE,
- _port.getAttribute(AmqpPort.RECEIVE_BUFFER_SIZE));
assertEquals("Unexpected need client auth", PortFactory.DEFAULT_AMQP_NEED_CLIENT_AUTH,
_port.getAttribute(Port.NEED_CLIENT_AUTH));
assertEquals("Unexpected want client auth", PortFactory.DEFAULT_AMQP_WANT_CLIENT_AUTH,
@@ -293,8 +287,6 @@ public class PortFactoryTest extends Qpi
assertEquals(_tcpTransports, _port.getTransports());
}
assertEquals(amqp010ProtocolSet, _port.getProtocols());
- assertEquals("Unexpected send buffer size", 2, _port.getAttribute(AmqpPort.SEND_BUFFER_SIZE));
- assertEquals("Unexpected receive buffer size", 1, _port.getAttribute(AmqpPort.RECEIVE_BUFFER_SIZE));
assertEquals("Unexpected need client auth", needClientAuth, _port.getAttribute(Port.NEED_CLIENT_AUTH));
assertEquals("Unexpected want client auth", wantClientAuth, _port.getAttribute(Port.WANT_CLIENT_AUTH));
assertEquals("Unexpected tcp no delay", true, _port.getAttribute(Port.TCP_NO_DELAY));
Modified: qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/store/BrokerStoreUpgraderAndRecovererTest.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/store/BrokerStoreUpgraderAndRecovererTest.java?rev=1701289&r1=1701288&r2=1701289&view=diff
==============================================================================
--- qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/store/BrokerStoreUpgraderAndRecovererTest.java (original)
+++ qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/store/BrokerStoreUpgraderAndRecovererTest.java Fri Sep 4 16:32:59 2015
@@ -412,7 +412,7 @@ public class BrokerStoreUpgraderAndRecov
List<ConfiguredObjectRecord> records = recoverer.upgrade(dcs);
assertTrue("No virtualhostalias rescords should be returned",
- findRecordByType("VirtualHostAlias", records).isEmpty());
+ findRecordByType("VirtualHostAlias", records).isEmpty());
}
@@ -432,7 +432,7 @@ public class BrokerStoreUpgraderAndRecov
List<ConfiguredObjectRecord> records = recoverer.upgrade(dcs);
assertFalse("VirtualHostAlias rescords should be returned",
- findRecordByType("VirtualHostAlias", records).isEmpty());
+ findRecordByType("VirtualHostAlias", records).isEmpty());
}
@@ -454,9 +454,36 @@ public class BrokerStoreUpgraderAndRecov
List<ConfiguredObjectRecord> records = recoverer.upgrade(dcs);
assertTrue("No virtualhostalias rescords should be returned",
- findRecordByType("VirtualHostAlias", records).isEmpty());
+ findRecordByType("VirtualHostAlias", records).isEmpty());
}
+
+ public void testUpgradeAMQPPortWithNetworkBuffers()
+ {
+ Map<String, Object> portAttributes = new HashMap<>();
+ portAttributes.put("name", getTestName());
+ portAttributes.put("type", "AMQP");
+ portAttributes.put("receiveBufferSize", "1");
+ portAttributes.put("sendBufferSize", "2");
+
+ _brokerRecord.getAttributes().put("modelVersion", "3.0");
+
+ ConfiguredObjectRecord portRecord = new ConfiguredObjectRecordImpl(UUID.randomUUID(), "Port",
+ portAttributes, Collections.singletonMap("Broker", _brokerRecord.getId()));
+ DurableConfigurationStore dcs = new DurableConfigurationStoreStub(_brokerRecord, portRecord);
+
+ BrokerStoreUpgraderAndRecoverer recoverer = new BrokerStoreUpgraderAndRecoverer(_systemConfig);
+ List<ConfiguredObjectRecord> records = recoverer.upgrade(dcs);
+
+ List<ConfiguredObjectRecord> ports = findRecordByType("Port", records);
+ assertEquals("Unexpected port size", 1, ports.size());
+ ConfiguredObjectRecord upgradedRecord = ports.get(0);
+ Map<String, Object> attributes = upgradedRecord.getAttributes();
+ assertFalse("receiveBufferSize is found " + attributes.get("receiveBufferSize"), attributes.containsKey("receiveBufferSize"));
+ assertFalse("sendBufferSize is found " + attributes.get("sendBufferSize"), attributes.containsKey("sendBufferSize"));
+ assertEquals("Unexpected name", getTestName(), attributes.get("name"));
+ }
+
private void upgradeBrokerRecordAndAssertUpgradeResults()
{
DurableConfigurationStore dcs = new DurableConfigurationStoreStub(_brokerRecord);
Modified: qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/transport/NetworkConnectionSchedulerTest.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/transport/NetworkConnectionSchedulerTest.java?rev=1701289&r1=1701288&r2=1701289&view=diff
==============================================================================
--- qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/transport/NetworkConnectionSchedulerTest.java (original)
+++ qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/transport/NetworkConnectionSchedulerTest.java Fri Sep 4 16:32:59 2015
@@ -43,8 +43,7 @@ public class NetworkConnectionSchedulerT
{
AmqpPort port = mock(AmqpPort.class);
when(port.isTcpNoDelay()).thenReturn(true);
- when(port.getSendBufferSize()).thenReturn(1);
- when(port.getReceiveBufferSize()).thenReturn(1);
+ when(port.getNetworkBufferSize()).thenReturn(1);
when(port.getPort()).thenReturn(0);
when(port.getBindingAddress()).thenReturn("*");
when(port.getEnabledCipherSuites()).thenReturn(Collections.emptyList());
Modified: qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/transport/TCPandSSLTransportTest.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/transport/TCPandSSLTransportTest.java?rev=1701289&r1=1701288&r2=1701289&view=diff
==============================================================================
--- qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/transport/TCPandSSLTransportTest.java (original)
+++ qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/transport/TCPandSSLTransportTest.java Fri Sep 4 16:32:59 2015
@@ -106,8 +106,7 @@ public class TCPandSSLTransportTest exte
final AmqpPort<?> port = mock(AmqpPort.class);
when(port.getPort()).thenReturn(0);
- when(port.getSendBufferSize()).thenReturn(64*1024);
- when(port.getReceiveBufferSize()).thenReturn(64*1024);
+ when(port.getNetworkBufferSize()).thenReturn(64*1024);
when(port.canAcceptNewConnection(any(SocketAddress.class))).thenReturn(true);
when(port.getThreadPoolMinimum()).thenReturn(1);
when(port.getThreadPoolMaximum()).thenReturn(1);
Modified: qpid/java/trunk/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnectionDelegate.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnectionDelegate.java?rev=1701289&r1=1701288&r2=1701289&view=diff
==============================================================================
--- qpid/java/trunk/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnectionDelegate.java (original)
+++ qpid/java/trunk/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnectionDelegate.java Fri Sep 4 16:32:59 2015
@@ -83,7 +83,7 @@ public class ServerConnectionDelegate ex
_localFQDN = localFQDN;
_maxNoOfChannels = broker.getConnection_sessionCountLimit();
_subjectCreator = subjectCreator;
- _maximumFrameSize = (int) Math.min(0xffffl, broker.getContextValue(Integer.class, Broker.BROKER_FRAME_SIZE));
+ _maximumFrameSize = Math.min(0xffff, broker.getNetworkBufferSize());
}
private static List<String> getFeatures(Broker<?> broker)
Modified: qpid/java/trunk/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQPConnection_0_8.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQPConnection_0_8.java?rev=1701289&r1=1701288&r2=1701289&view=diff
==============================================================================
--- qpid/java/trunk/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQPConnection_0_8.java (original)
+++ qpid/java/trunk/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQPConnection_0_8.java Fri Sep 4 16:32:59 2015
@@ -1209,7 +1209,7 @@ public class AMQPConnection_0_8
case SUCCESS:
_logger.debug("Connected as: {} ", authResult.getSubject());
- int frameMax = broker.getContextValue(Integer.class, Broker.BROKER_FRAME_SIZE);
+ int frameMax = broker.getNetworkBufferSize();
if (frameMax <= 0)
{
@@ -1316,7 +1316,7 @@ public class AMQPConnection_0_8
_logger.debug("Connected as: {}", authResult.getSubject());
setAuthorizedSubject(authResult.getSubject());
- int frameMax = broker.getContextValue(Integer.class, Broker.BROKER_FRAME_SIZE);
+ int frameMax = broker.getNetworkBufferSize();
if (frameMax <= 0)
{
@@ -1359,7 +1359,7 @@ public class AMQPConnection_0_8
initHeartbeats(heartbeat);
- int brokerFrameMax = getBroker().getContextValue(Integer.class, Broker.BROKER_FRAME_SIZE);
+ int brokerFrameMax = getBroker().getNetworkBufferSize();
if (brokerFrameMax <= 0)
{
brokerFrameMax = Integer.MAX_VALUE;
Modified: qpid/java/trunk/systests/src/main/java/org/apache/qpid/systest/rest/Asserts.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/systests/src/main/java/org/apache/qpid/systest/rest/Asserts.java?rev=1701289&r1=1701288&r2=1701289&view=diff
==============================================================================
--- qpid/java/trunk/systests/src/main/java/org/apache/qpid/systest/rest/Asserts.java (original)
+++ qpid/java/trunk/systests/src/main/java/org/apache/qpid/systest/rest/Asserts.java Fri Sep 4 16:32:59 2015
@@ -312,8 +312,6 @@ public class Asserts
Port.AUTHENTICATION_PROVIDER,
Port.BINDING_ADDRESS,
Port.TCP_NO_DELAY,
- AmqpPort.SEND_BUFFER_SIZE,
- AmqpPort.RECEIVE_BUFFER_SIZE,
Port.NEED_CLIENT_AUTH,
Port.WANT_CLIENT_AUTH,
Port.KEY_STORE,
Modified: qpid/java/trunk/systests/src/test/java/org/apache/qpid/transport/MaxFrameSizeTest.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/systests/src/test/java/org/apache/qpid/transport/MaxFrameSizeTest.java?rev=1701289&r1=1701288&r2=1701289&view=diff
==============================================================================
--- qpid/java/trunk/systests/src/test/java/org/apache/qpid/transport/MaxFrameSizeTest.java (original)
+++ qpid/java/trunk/systests/src/test/java/org/apache/qpid/transport/MaxFrameSizeTest.java Fri Sep 4 16:32:59 2015
@@ -141,7 +141,7 @@ public class MaxFrameSizeTest extends Qp
TestBrokerConfiguration.ENTRY_NAME_AUTHENTICATION_PROVIDER,
"secureOnlyMechanisms",
"[]");
- setTestSystemProperty(Broker.BROKER_FRAME_SIZE, "8192");
+ setTestSystemProperty(Broker.NETWORK_BUFFER_SIZE, String.valueOf(Broker.MINIMUM_NETWORK_BUFFER_SIZE));
super.setUp();
if(isBroker010())
{
@@ -167,7 +167,7 @@ public class MaxFrameSizeTest extends Qp
}
else
{
- doAMQP08test(10000, new ResultEvaluator()
+ doAMQP08test(Broker.MINIMUM_NETWORK_BUFFER_SIZE + 1, new ResultEvaluator()
{
@Override
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org