Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix and test for: https://issues.apache.org/jira/browse/AMQ-4321 

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1446348 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Timothy A. Bish 2013-02-14 21:42:16 +00:00
parent ecd5c2f738
commit 7809e8067e
2 changed files with 121 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
activemq-client/src/main/java/org/apache/activemq/transport/nio/NIOOutputStream.java
View File

@ -76,6 +76,7 @@ public class NIOOutputStream extends OutputStream implements TimeStampStream {
* @param b - byte to write
* @throws IOException
*/
@Override
public void write(int b) throws IOException {
checkClosed();
if (availableBufferToWrite() < 1) {
@ -92,6 +93,7 @@ public class NIOOutputStream extends OutputStream implements TimeStampStream {
* @param len the length of data to write
* @throws IOException
*/
@Override
public void write(byte b[], int off, int len) throws IOException {
checkClosed();
if (availableBufferToWrite() < len) {
@ -112,6 +114,7 @@ public class NIOOutputStream extends OutputStream implements TimeStampStream {
*
* @throws IOException
*/
@Override
public void flush() throws IOException {
if (count > 0 && out != null) {
byteBuffer.position(0);
@ -126,6 +129,7 @@ public class NIOOutputStream extends OutputStream implements TimeStampStream {
*
* @throws IOException
*/
@Override
public void close() throws IOException {
super.close();
if (engine != null) {
@ -191,7 +195,6 @@ public class NIOOutputStream extends OutputStream implements TimeStampStream {
// Since the write is non-blocking, all the data may not have been
// written.
out.write(plain);
remaining = plain.remaining();
// if the data buffer was larger than the packet buffer we might need to
// wrap more packets until we reach the end of data, but only when plain
@ -202,6 +205,8 @@ public class NIOOutputStream extends OutputStream implements TimeStampStream {
engine.wrap(data, plain);
plain.flip();
}
remaining = plain.remaining();
}
} finally {
writeTimestamp = -1;
@ -212,6 +217,7 @@ public class NIOOutputStream extends OutputStream implements TimeStampStream {
/* (non-Javadoc)
* @see org.apache.activemq.transport.tcp.TimeStampStream#isWriting()
*/
@Override
public boolean isWriting() {
return writeTimestamp > 0;
}
@ -219,6 +225,7 @@ public class NIOOutputStream extends OutputStream implements TimeStampStream {
/* (non-Javadoc)
* @see org.apache.activemq.transport.tcp.TimeStampStream#getWriteTimestamp()
*/
@Override
public long getWriteTimestamp() {
return writeTimestamp;
}

113
activemq-unit-tests/src/test/java/org/apache/activemq/transport/nio/NIOSSLWindowSizeTest.java Normal file
View File

@ -0,0 +1,113 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.transport.nio;
import junit.framework.TestCase;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.TransportConnector;
import javax.jms.BytesMessage;
import javax.jms.Connection;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
@SuppressWarnings("javadoc")
public class NIOSSLWindowSizeTest extends TestCase {
BrokerService broker;
Connection connection;
Session session;
public static final String KEYSTORE_TYPE = "jks";
public static final String PASSWORD = "password";
public static final String SERVER_KEYSTORE = "src/test/resources/server.keystore";
public static final String TRUST_KEYSTORE = "src/test/resources/client.keystore";
public static final int PRODUCER_COUNT = 1;
public static final int CONSUMER_COUNT = 1;
public static final int MESSAGE_COUNT = 1;
public static final int MESSAGE_SIZE = 65536;
byte[] messageData;
@Override
protected void setUp() throws Exception {
System.setProperty("javax.net.ssl.trustStore", TRUST_KEYSTORE);
System.setProperty("javax.net.ssl.trustStorePassword", PASSWORD);
System.setProperty("javax.net.ssl.trustStoreType", KEYSTORE_TYPE);
System.setProperty("javax.net.ssl.keyStore", SERVER_KEYSTORE);
System.setProperty("javax.net.ssl.keyStoreType", KEYSTORE_TYPE);
System.setProperty("javax.net.ssl.keyStorePassword", PASSWORD);
broker = new BrokerService();
broker.setPersistent(false);
broker.setUseJmx(false);
TransportConnector connector = broker.addConnector("nio+ssl://localhost:0?transport.needClientAuth=true");
broker.start();
broker.waitUntilStarted();
messageData = new byte[MESSAGE_SIZE];
for (int i = 0; i < MESSAGE_SIZE; i++)
{
messageData[i] = (byte) (i & 0xff);
}
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("nio+ssl://localhost:" + connector.getConnectUri().getPort());
connection = factory.createConnection();
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
connection.start();
}
@Override
protected void tearDown() throws Exception {
if (session != null) {
session.close();
}
if (connection != null) {
connection.close();
}
if (broker != null) {
broker.stop();
broker.waitUntilStopped();
}
}
public void testLargePayload() throws Exception {
Queue dest = session.createQueue("TEST");
MessageProducer prod = null;
try {
prod = session.createProducer(dest);
BytesMessage msg = session.createBytesMessage();
msg.writeBytes(messageData);
prod.send(msg);
} finally {
prod.close();
}
MessageConsumer cons = null;
try
{
cons = session.createConsumer(dest);
assertNotNull(cons.receive(30000L));
} finally {
cons.close();
}
}
}