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

https://issues.apache.org/jira/browse/AMQ-2583 - stomp+nio+ssl initial implementation 

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1150630 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Bosanac Dejan 2011-07-25 11:08:52 +00:00
parent c78c2449cb
commit 036f2a8d7c
12 changed files with 378 additions and 118 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
activemq-core/src/main/java/org/apache/activemq/transport/nio/NIOOutputStream.java
View File

@ -128,6 +128,9 @@ public class NIOOutputStream extends OutputStream implements TimeStampStream {
*/
public void close() throws IOException {
super.close();
if (engine != null) {
engine.closeOutbound();
}
closed = true;
}
@ -159,6 +162,7 @@ public class NIOOutputStream extends OutputStream implements TimeStampStream {
} else {
plain = data;
}
int remaining = plain.remaining();
int lastRemaining = remaining - 1;
long delay = 1;

119
activemq-core/src/main/java/org/apache/activemq/transport/nio/NIOSSLTransport.java
View File

@ -28,6 +28,7 @@ import javax.net.SocketFactory;
import javax.net.ssl.*;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.EOFException;
import java.io.IOException;
import java.net.Socket;
import java.net.URI;
@ -36,18 +37,18 @@ import java.nio.ByteBuffer;
public class NIOSSLTransport extends NIOTransport {
private boolean needClientAuth;
private boolean wantClientAuth;
private String[] enabledCipherSuites;
protected boolean needClientAuth;
protected boolean wantClientAuth;
protected String[] enabledCipherSuites;
protected SSLContext sslContext;
protected SSLEngine sslEngine;
protected SSLSession sslSession;
boolean handshakeInProgress = false;
SSLEngineResult.Status status = null;
SSLEngineResult.HandshakeStatus handshakeStatus = null;
protected boolean handshakeInProgress = false;
protected SSLEngineResult.Status status = null;
protected SSLEngineResult.HandshakeStatus handshakeStatus = null;
public NIOSSLTransport(WireFormat wireFormat, SocketFactory socketFactory, URI remoteLocation, URI localLocation) throws UnknownHostException, IOException {
super(wireFormat, socketFactory, remoteLocation, localLocation);
@ -90,11 +91,8 @@ public class NIOSSLTransport extends NIOTransport {
outputStream.setEngine(sslEngine);
this.dataOut = new DataOutputStream(outputStream);
this.buffOut = outputStream;
sslEngine.beginHandshake();
handshakeStatus = sslEngine.getHandshakeStatus();
doHandshake();
} catch (Exception e) {
@ -125,8 +123,6 @@ public class NIOSSLTransport extends NIOTransport {
}
}
protected void serviceRead() {
try {
if (handshakeInProgress) {
@ -136,62 +132,75 @@ public class NIOSSLTransport extends NIOTransport {
ByteBuffer plain = ByteBuffer.allocate(sslSession.getApplicationBufferSize());
plain.position(plain.limit());
while (true) {
if (nextFrameSize == -1) {
if (!plain.hasRemaining()) {
plain.clear();
int readCount = secureRead(plain);
if (readCount == 0)
break;
}
nextFrameSize = plain.getInt();
if (wireFormat instanceof OpenWireFormat) {
long maxFrameSize = ((OpenWireFormat)wireFormat).getMaxFrameSize();
if (nextFrameSize > maxFrameSize) {
throw new IOException("Frame size of " + (nextFrameSize / (1024 * 1024)) + " MB larger than max allowed " + (maxFrameSize / (1024 * 1024)) + " MB");
}
}
currentBuffer = ByteBuffer.allocate(nextFrameSize + 4);
currentBuffer.putInt(nextFrameSize);
if (currentBuffer.hasRemaining()) {
if (currentBuffer.remaining() >= plain.remaining()) {
currentBuffer.put(plain);
} else {
byte[] fill = new byte[currentBuffer.remaining()];
plain.get(fill);
currentBuffer.put(fill);
}
}
while(true) {
if (!plain.hasRemaining()) {
if (currentBuffer.hasRemaining()) {
continue;
} else {
currentBuffer.flip();
Object command = wireFormat.unmarshal(new DataInputStream(new NIOInputStream(currentBuffer)));
doConsume((Command) command);
plain.clear();
int readCount = secureRead(plain);
nextFrameSize = -1;
if (readCount == 0)
break;
// channel is closed, cleanup
if (readCount== -1) {
onException(new EOFException());
selection.close();
break;
}
}
}
processCommand(plain);
}
} catch (IOException e) {
onException(e);
} catch (Throwable e) {
onException(IOExceptionSupport.create(e));
}
}
protected void processCommand(ByteBuffer plain) throws Exception {
nextFrameSize = plain.getInt();
if (wireFormat instanceof OpenWireFormat) {
long maxFrameSize = ((OpenWireFormat) wireFormat).getMaxFrameSize();
if (nextFrameSize > maxFrameSize) {
throw new IOException("Frame size of " + (nextFrameSize / (1024 * 1024)) + " MB larger than max allowed " + (maxFrameSize / (1024 * 1024)) + " MB");
}
}
currentBuffer = ByteBuffer.allocate(nextFrameSize + 4);
currentBuffer.putInt(nextFrameSize);
if (currentBuffer.hasRemaining()) {
if (currentBuffer.remaining() >= plain.remaining()) {
currentBuffer.put(plain);
} else {
byte[] fill = new byte[currentBuffer.remaining()];
plain.get(fill);
currentBuffer.put(fill);
}
}
if (currentBuffer.hasRemaining()) {
return;
} else {
currentBuffer.flip();
Object command = wireFormat.unmarshal(new DataInputStream(new NIOInputStream(currentBuffer)));
doConsume((Command) command);
nextFrameSize = -1;
}
}
private int secureRead(ByteBuffer plain) throws Exception {
int bytesRead = channel.read(inputBuffer);
if (bytesRead == -1) {
sslEngine.closeInbound();
if (inputBuffer.position() == 0 ||
status == SSLEngineResult.Status.BUFFER_UNDERFLOW) {
return -1;
protected int secureRead(ByteBuffer plain) throws Exception {
if (!(inputBuffer.position() != 0 && inputBuffer.hasRemaining())) {
int bytesRead = channel.read(inputBuffer);
if (bytesRead == -1) {
sslEngine.closeInbound();
if (inputBuffer.position() == 0 ||
status == SSLEngineResult.Status.BUFFER_UNDERFLOW) {
return -1;
}
}
}
@ -206,12 +215,13 @@ public class NIOSSLTransport extends NIOTransport {
res.bytesProduced() == 0);
if (res.getHandshakeStatus() == SSLEngineResult.HandshakeStatus.FINISHED) {
finishHandshake();
finishHandshake();
}
status = res.getStatus();
handshakeStatus = res.getHandshakeStatus();
//TODO deal with BUFFER_OVERFLOW
if (status == SSLEngineResult.Status.CLOSED) {
@ -253,6 +263,7 @@ public class NIOSSLTransport extends NIOTransport {
protected void doStop(ServiceStopper stopper) throws Exception {
if (channel != null) {
channel.close();
channel = null;
}
super.doStop(stopper);
}

1
activemq-core/src/main/java/org/apache/activemq/transport/nio/NIOTransport.java
View File

@ -161,6 +161,7 @@ public class NIOTransport extends TcpTransport {
protected void doStop(ServiceStopper stopper) throws Exception {
if (selection != null) {
selection.close();
selection = null;
}
super.doStop(stopper);
}

102
activemq-core/src/main/java/org/apache/activemq/transport/stomp/StompCodec.java Normal file
View File

@ -0,0 +1,102 @@
/**
* 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.stomp;
import org.apache.activemq.transport.tcp.TcpTransport;
import org.apache.activemq.util.ByteArrayOutputStream;
import org.apache.activemq.util.DataByteArrayInputStream;
import org.apache.activemq.wireformat.WireFormat;
import java.io.ByteArrayInputStream;
import java.util.HashMap;
public class StompCodec {
TcpTransport transport;
ByteArrayOutputStream currentCommand = new ByteArrayOutputStream();
boolean processedHeaders = false;
String action;
HashMap<String, String> headers;
int contentLength = -1;
int readLength = 0;
int previousByte = -1;
public StompCodec(TcpTransport transport) {
this.transport = transport;
}
public void parse(ByteArrayInputStream input, int readSize) throws Exception {
int i = 0;
int b;
while(i++ < readSize) {
b = input.read();
// skip repeating nulls
if (!processedHeaders && previousByte == 0 && b == 0) {
continue;
}
if (!processedHeaders) {
currentCommand.write(b);
// end of headers section, parse action and header
if (previousByte == '\n' && b == '\n') {
if (transport.getWireFormat() instanceof StompWireFormat) {
DataByteArrayInputStream data = new DataByteArrayInputStream(currentCommand.toByteArray());
action = ((StompWireFormat)transport.getWireFormat()).parseAction(data);
headers = ((StompWireFormat)transport.getWireFormat()).parseHeaders(data);
String contentLengthHeader = headers.get(Stomp.Headers.CONTENT_LENGTH);
if (contentLengthHeader != null) {
contentLength = ((StompWireFormat)transport.getWireFormat()).parseContentLength(contentLengthHeader);
} else {
contentLength = -1;
}
}
processedHeaders = true;
currentCommand.reset();
}
} else {
if (contentLength == -1) {
// end of command reached, unmarshal
if (b == 0) {
processCommand();
} else {
currentCommand.write(b);
}
} else {
// read desired content length
if (readLength++ == contentLength) {
processCommand();
readLength = 0;
} else {
currentCommand.write(b);
}
}
}
previousByte = b;
}
}
protected void processCommand() throws Exception {
StompFrame frame = new StompFrame(action, headers, currentCommand.toByteArray());
transport.doConsume(frame);
processedHeaders = false;
currentCommand.reset();
contentLength = -1;
}
}

61
activemq-core/src/main/java/org/apache/activemq/transport/stomp/StompNIOSSLTransport.java Normal file
View File

@ -0,0 +1,61 @@
/**
* 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.stomp;
import org.apache.activemq.transport.nio.NIOSSLTransport;
import org.apache.activemq.util.IOExceptionSupport;
import org.apache.activemq.wireformat.WireFormat;
import javax.net.SocketFactory;
import java.io.ByteArrayInputStream;
import java.io.EOFException;
import java.io.IOException;
import java.net.Socket;
import java.net.URI;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
public class StompNIOSSLTransport extends NIOSSLTransport {
StompCodec codec;
public StompNIOSSLTransport(WireFormat wireFormat, SocketFactory socketFactory, URI remoteLocation, URI localLocation) throws UnknownHostException, IOException {
super(wireFormat, socketFactory, remoteLocation, localLocation);
}
public StompNIOSSLTransport(WireFormat wireFormat, Socket socket) throws IOException {
super(wireFormat, socket);
}
@Override
protected void initializeStreams() throws IOException {
codec = new StompCodec(this);
super.initializeStreams();
if (inputBuffer.position() != 0 && inputBuffer.hasRemaining()) {
serviceRead();
}
}
@Override
protected void processCommand(ByteBuffer plain) throws Exception {
byte[] fill = new byte[plain.remaining()];
plain.get(fill);
ByteArrayInputStream input = new ByteArrayInputStream(fill);
codec.parse(input, fill.length);
}
}

69
activemq-core/src/main/java/org/apache/activemq/transport/stomp/StompNIOSSLTransportFactory.java Normal file
View File

@ -0,0 +1,69 @@
/**
* 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.stomp;
import org.apache.activemq.broker.SslContext;
import org.apache.activemq.transport.Transport;
import org.apache.activemq.transport.TransportServer;
import org.apache.activemq.transport.tcp.TcpTransport;
import org.apache.activemq.transport.tcp.TcpTransportServer;
import org.apache.activemq.wireformat.WireFormat;
import javax.net.ServerSocketFactory;
import javax.net.SocketFactory;
import javax.net.ssl.SSLContext;
import java.io.IOException;
import java.net.Socket;
import java.net.URI;
import java.net.URISyntaxException;
import java.net.UnknownHostException;
public class StompNIOSSLTransportFactory extends StompNIOTransportFactory {
SSLContext context;
@Override
protected TcpTransportServer createTcpTransportServer(URI location, ServerSocketFactory serverSocketFactory) throws IOException, URISyntaxException {
return new TcpTransportServer(this, location, serverSocketFactory) {
protected Transport createTransport(Socket socket, WireFormat format) throws IOException {
StompNIOSSLTransport transport = new StompNIOSSLTransport(format, socket);
if (context != null) {
transport.setSslContext(context);
}
return transport;
}
};
}
@Override
protected TcpTransport createTcpTransport(WireFormat wf, SocketFactory socketFactory, URI location, URI localLocation) throws UnknownHostException, IOException {
return new StompNIOSSLTransport(wf, socketFactory, location, localLocation);
}
@Override
public TransportServer doBind(URI location) throws IOException {
if (SslContext.getCurrentSslContext() != null) {
try {
context = SslContext.getCurrentSslContext().getSSLContext();
} catch (Exception e) {
throw new IOException(e);
}
}
return super.doBind(location);
}
}

67
activemq-core/src/main/java/org/apache/activemq/transport/stomp/StompNIOTransport.java
View File

@ -54,13 +54,7 @@ public class StompNIOTransport extends TcpTransport {
private SelectorSelection selection;
private ByteBuffer inputBuffer;
ByteArrayOutputStream currentCommand = new ByteArrayOutputStream();
boolean processedHeaders = false;
String action;
HashMap<String, String> headers;
int contentLength = -1;
int readLength = 0;
int previousByte = -1;
StompCodec codec;
public StompNIOTransport(WireFormat wireFormat, SocketFactory socketFactory, URI remoteLocation, URI localLocation) throws UnknownHostException, IOException {
super(wireFormat, socketFactory, remoteLocation, localLocation);
@ -93,6 +87,7 @@ public class StompNIOTransport extends TcpTransport {
NIOOutputStream outPutStream = new NIOOutputStream(channel, 8 * 1024);
this.dataOut = new DataOutputStream(outPutStream);
this.buffOut = outPutStream;
codec = new StompCodec(this);
}
private void serviceRead() {
@ -114,57 +109,9 @@ public class StompNIOTransport extends TcpTransport {
inputBuffer.flip();
int b;
ByteArrayInputStream input = new ByteArrayInputStream(inputBuffer.array());
codec.parse(input, readSize);
int i = 0;
while(i++ < readSize) {
b = input.read();
// skip repeating nulls
if (!processedHeaders && previousByte == 0 && b == 0) {
continue;
}
if (!processedHeaders) {
currentCommand.write(b);
// end of headers section, parse action and header
if (previousByte == '\n' && b == '\n') {
if (wireFormat instanceof StompWireFormat) {
DataByteArrayInputStream data = new DataByteArrayInputStream(currentCommand.toByteArray());
action = ((StompWireFormat)wireFormat).parseAction(data);
headers = ((StompWireFormat)wireFormat).parseHeaders(data);
String contentLengthHeader = headers.get(Stomp.Headers.CONTENT_LENGTH);
if (contentLengthHeader != null) {
contentLength = ((StompWireFormat)wireFormat).parseContentLength(contentLengthHeader);
} else {
contentLength = -1;
}
}
processedHeaders = true;
currentCommand.reset();
}
} else {
if (contentLength == -1) {
// end of command reached, unmarshal
if (b == 0) {
processCommand();
} else {
currentCommand.write(b);
}
} else {
// read desired content length
if (readLength++ == contentLength) {
processCommand();
readLength = 0;
} else {
currentCommand.write(b);
}
}
}
previousByte = b;
}
// clear the buffer
inputBuffer.clear();
@ -176,14 +123,6 @@ public class StompNIOTransport extends TcpTransport {
}
}
private void processCommand() throws Exception {
StompFrame frame = new StompFrame(action, headers, currentCommand.toByteArray());
doConsume(frame);
processedHeaders = false;
currentCommand.reset();
contentLength = -1;
}
protected void doStart() throws Exception {
connect();
selection.setInterestOps(SelectionKey.OP_READ);

4
activemq-core/src/main/java/org/apache/activemq/transport/tcp/TcpTransport.java
View File

@ -687,4 +687,8 @@ public class TcpTransport extends TransportThreadSupport implements Transport, S
this.typeOfServiceChosen = false;
return true;
}
public WireFormat getWireFormat() {
return wireFormat;
}
}

17
activemq-core/src/main/resources/META-INF/services/org/apache/activemq/transport/stomp+nio+ssl Normal file
View File

@ -0,0 +1,17 @@
## ---------------------------------------------------------------------------
## 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.
## ---------------------------------------------------------------------------
class=org.apache.activemq.transport.stomp.StompNIOSSLTransportFactory

49
activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompNIOSSLTest.java Normal file
View File

@ -0,0 +1,49 @@
/**
* 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.stomp;
import javax.net.SocketFactory;
import javax.net.ssl.SSLSocketFactory;
import java.io.IOException;
import java.net.Socket;
import java.net.URI;
public class StompNIOSSLTest extends StompTest {
protected void setUp() throws Exception {
bindAddress = "stomp+nio+ssl://localhost:61613";
confUri = "xbean:org/apache/activemq/transport/stomp/sslstomp-auth-broker.xml";
System.setProperty("javax.net.ssl.trustStore", "src/test/resources/client.keystore");
System.setProperty("javax.net.ssl.trustStorePassword", "password");
System.setProperty("javax.net.ssl.trustStoreType", "jks");
System.setProperty("javax.net.ssl.keyStore", "src/test/resources/server.keystore");
System.setProperty("javax.net.ssl.keyStorePassword", "password");
System.setProperty("javax.net.ssl.keyStoreType", "jks");
super.setUp();
}
@Override
protected void tearDown() throws Exception {
super.tearDown();
}
protected Socket createSocket(URI connectUri) throws IOException {
SocketFactory factory = SSLSocketFactory.getDefault();
return factory.createSocket("127.0.0.1", connectUri.getPort());
}
}

2
activemq-core/src/test/java/org/apache/activemq/transport/stomp/StompTest.java
View File

@ -111,6 +111,7 @@ public class StompTest extends CombinationTestSupport {
}
broker = BrokerFactory.createBroker(new URI(confUri));
broker.start();
broker.waitUntilStarted();
stompConnect();
@ -143,6 +144,7 @@ public class StompTest extends CombinationTestSupport {
// Some tests explicitly disconnect from stomp so can ignore
} finally {
broker.stop();
broker.waitUntilStopped();
}
}

1
activemq-core/src/test/resources/org/apache/activemq/transport/stomp/sslstomp-auth-broker.xml
View File

@ -69,6 +69,7 @@
<transportConnectors>
<transportConnector name="stomp+ssl" uri="stomp+ssl://localhost:61612"/>
<transportConnector name="stomp+nio+ssl" uri="stomp+nio+ssl://localhost:61613"/>
</transportConnectors>