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-5356 

Upgraded to Jetty 9.2.x to solve JDK 8 compatibility issues.
This required migrating to the new WebSocket API.
This commit is contained in:
Christopher L. Shannon (cshannon) 2015-08-19 16:55:14 +00:00
parent 505916b927
commit f44c3d20ed
47 changed files with 1161 additions and 1155 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
activemq-fileserver/pom.xml
View File

@ -33,12 +33,11 @@
<build>
<plugins>
<plugin>
<groupId>org.mortbay.jetty</groupId>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-maven-plugin</artifactId>
<version>${jetty-version}</version>
<configuration>
<connectors>
<connector implementation="org.eclipse.jetty.server.nio.SelectChannelConnector">
<connector implementation="org.eclipse.jetty.server.ServerConnector">
<port>${jetty.port}</port>
<maxIdleTime>60000</maxIdleTime>
</connector>
@ -63,12 +62,13 @@
<artifactId>slf4j-api</artifactId>
</dependency>
<!-- web container -->
<dependency>
<groupId>org.apache.geronimo.specs</groupId>
<artifactId>geronimo-servlet_3.0_spec</artifactId>
<scope>provided</scope>
</dependency>
<!-- web container -->
<dependency>
<groupId>org.apache.tomcat</groupId>
<artifactId>tomcat-servlet-api</artifactId>
<scope>provided</scope>
</dependency>
<!-- used for testing -->
<dependency>

9
activemq-fileserver/src/main/webapp/WEB-INF/web.xml
View File

@ -15,11 +15,10 @@
See the License for the specific language governing permissions and
limitations under the License.
-->
<!DOCTYPE web-app
PUBLIC "-//Sun Microsystems, Inc.//DTD Web Application 2.3//EN"
"http://java.sun.com/dtd/web-app_2_3.dtd">
<web-app>
<web-app xmlns="http://java.sun.com/xml/ns/javaee"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_3_0.xsd"
version="3.0">
<display-name>RESTful file access application</display-name>

8
activemq-fileserver/src/test/java/org/apache/activemq/util/HttpTestSupport.java
View File

@ -30,9 +30,7 @@ import junit.framework.TestCase;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.BrokerService;
import org.eclipse.jetty.server.Connector;
import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.server.nio.SelectChannelConnector;
import org.eclipse.jetty.webapp.WebAppContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -51,8 +49,8 @@ public abstract class HttpTestSupport extends TestCase {
protected boolean createBroker = true;
final File homeDir = new File("src/main/webapp/uploads/");
int port;
private int getPort(Object o) throws Exception {
return (Integer)o.getClass().getMethod("getLocalPort").invoke(o);
}
@ -68,7 +66,7 @@ public abstract class HttpTestSupport extends TestCase {
server.setHandler(context);
server.start();
int port = getPort(server.getConnectors()[0]);
port = getPort(server.getConnectors()[0]);
waitForJettySocketToAccept("http://localhost:" + port);
@ -93,6 +91,8 @@ public abstract class HttpTestSupport extends TestCase {
}
}
@Override
protected void tearDown() throws Exception {
server.stop();

4
activemq-fileserver/src/test/java/org/apache/activemq/util/RestFilterTest.java
View File

@ -28,12 +28,12 @@ import org.eclipse.jetty.util.IO;
public class RestFilterTest extends HttpTestSupport {
protected boolean createBroker = false;
public void testFilter() throws Exception {
byte[] fileContents = new byte[] {
'a', 'b', 'c'
};
URL url = new URL("http://localhost:8080/uploads/file.txt");
URL url = new URL("http://localhost:" + port + "/uploads/file.txt");
// 1. upload
HttpURLConnection connection = (HttpURLConnection)url.openConnection();

11
activemq-http/pom.xml
View File

@ -38,6 +38,10 @@
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context</artifactId>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-jms</artifactId>
</dependency>
<dependency>
<groupId>${project.groupId}</groupId>
@ -111,11 +115,10 @@
<dependency>
<groupId>org.eclipse.jetty.websocket</groupId>
<artifactId>websocket-server</artifactId>
<version>${jetty9-version}</version>
<scope>provided</scope>
<optional>true</optional>
</dependency>
<dependency>
</dependency>
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-core</artifactId>
<scope>test</scope>
@ -143,7 +146,7 @@
org.apache.activemq.transport.https*;version=${project.version};-noimport:=;-split-package:=merge-last
</Export-Package>
<Import-Package>
org.eclipse.jetty*;version="[7.5,8.0)";resolution:=optional,
org.eclipse.jetty*;version="[9.0,10.0)";resolution:=optional,
!org.apache.activemq.transport.ws*;version=${project.version},
!org.apache.activemq.transport.xstream;version=${project.version},
!org.apache.activemq.transport.util;version=${project.version},

27
activemq-http/src/main/java/org/apache/activemq/transport/SecureSocketConnectorFactory.java
View File

@ -19,10 +19,10 @@ package org.apache.activemq.transport;
import javax.net.ssl.SSLContext;
import org.apache.activemq.broker.SslContext;
import org.apache.activemq.transport.https.Krb5AndCertsSslSocketConnector;
import org.apache.activemq.util.IntrospectionSupport;
import org.eclipse.jetty.server.Connector;
import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.server.ServerConnector;
import org.eclipse.jetty.util.ssl.SslContextFactory;
public class SecureSocketConnectorFactory extends SocketConnectorFactory {
@ -43,9 +43,9 @@ public class SecureSocketConnectorFactory extends SocketConnectorFactory {
private SslContext context;
private SslContextFactory contextFactory;
public SecureSocketConnectorFactory() {
}
public SecureSocketConnectorFactory(SslContext context) {
this.context = context;
@ -113,21 +113,16 @@ public class SecureSocketConnectorFactory extends SocketConnectorFactory {
factory = contextFactory;
}
if ("KRB".equals(auth) || "BOTH".equals(auth)
&& Server.getVersion().startsWith("8")) {
return new Krb5AndCertsSslSocketConnector(factory, auth);
//return new Krb5AndCertsSslSocketConnector(factory, auth);
return null;
} else {
try {
Class<?> cls = Class.forName("org.eclipse.jetty.server.ssl.SslSelectChannelConnector", true, Server.class.getClassLoader());
return (Connector)cls.getConstructor(SslContextFactory.class).newInstance(factory);
} catch (Throwable t) {
Class<?> c = Class.forName("org.eclipse.jetty.server.ServerConnector", true, Server.class.getClassLoader());
Connector connector = (Connector)c.getConstructor(Server.class, SslContextFactory.class).newInstance(server, factory);
Server.class.getMethod("setStopTimeout", Long.TYPE).invoke(server, 500);
connector.getClass().getMethod("setStopTimeout", Long.TYPE).invoke(connector, 500);
return connector;
}
ServerConnector connector = new ServerConnector(server, factory);
server.setStopTimeout(500);
connector.setStopTimeout(500);
return connector;
}
}
private void setTrustStore(SslContextFactory factory, String trustStore2) throws Exception {
@ -136,7 +131,7 @@ public class SecureSocketConnectorFactory extends SocketConnectorFactory {
}
// Properties
// --------------------------------------------------------------------------------

15
activemq-http/src/main/java/org/apache/activemq/transport/SocketConnectorFactory.java
View File

@ -21,23 +21,16 @@ import java.util.Map;
import org.apache.activemq.util.IntrospectionSupport;
import org.eclipse.jetty.server.Connector;
import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.server.ServerConnector;
public class SocketConnectorFactory {
private Map<String, Object> transportOptions;
public Connector createConnector(Server server) throws Exception {
Connector connector = null;
try {
connector = (Connector)Class.forName("org.eclipse.jetty.server.nio.SelectChannelConnector", true, Server.class.getClassLoader()).newInstance();
} catch (Throwable t) {
Class<?> c = Class.forName("org.eclipse.jetty.server.ServerConnector", true, Server.class.getClassLoader());
connector = (Connector)c.getConstructor(Server.class).newInstance(server);
Server.class.getMethod("setStopTimeout", Long.TYPE).invoke(server, 500);
connector.getClass().getMethod("setStopTimeout", Long.TYPE).invoke(connector, 500);
}
System.out.println(transportOptions);
ServerConnector connector = new ServerConnector(server);
server.setStopTimeout(500);
connector.setStopTimeout(500);
if (transportOptions != null) {
IntrospectionSupport.setProperties(connector, transportOptions, "");
}

7
activemq-http/src/main/java/org/apache/activemq/transport/WebTransportServerSupport.java
View File

@ -38,13 +38,13 @@ abstract public class WebTransportServerSupport extends TransportServerSupport {
private <T> void setConnectorProperty(String name, Class<T> type, T value) throws Exception {
connector.getClass().getMethod("set" + name, type).invoke(connector, value);
}
protected void createServer() {
server = new Server();
try {
server.getClass().getMethod("setStopTimeout", Long.TYPE).invoke(server, 500l);
} catch (Throwable t) {
//ignore, jetty 8.
//ignore, jetty 8.
}
}
public URI bind() throws Exception {
@ -58,9 +58,6 @@ abstract public class WebTransportServerSupport extends TransportServerSupport {
setConnectorProperty("Host", String.class, host);
setConnectorProperty("Port", Integer.TYPE, bindAddress.getPort());
if (Server.getVersion().startsWith("8")) {
connector.setServer(server);
}
server.addConnector(connector);
if (addr.isAnyLocalAddress()) {
host = InetAddressUtil.getLocalHostName();

309
activemq-http/src/main/java/org/apache/activemq/transport/https/Krb5AndCertsSslSocketConnector.java
View File

@ -16,27 +16,6 @@
*/
package org.apache.activemq.transport.https;
import java.io.IOException;
import java.net.ServerSocket;
import java.security.KeyManagementException;
import java.security.NoSuchAlgorithmException;
import java.security.NoSuchProviderException;
import java.security.Principal;
import java.util.Collections;
import java.util.List;
import java.util.Random;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLServerSocket;
import javax.net.ssl.SSLSocket;
import org.eclipse.jetty.http.HttpSchemes;
import org.eclipse.jetty.io.EndPoint;
import org.eclipse.jetty.server.Request;
import org.eclipse.jetty.server.ssl.ServletSSL;
import org.eclipse.jetty.server.ssl.SslSocketConnector;
import org.eclipse.jetty.util.ssl.SslContextFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Extend Jetty's {@link SslSocketConnector} to optionally also provide
@ -44,147 +23,149 @@ import org.slf4j.LoggerFactory;
* that we no longer honor requests to turn off NeedAuthentication when running
* with Kerberos support.
*/
public class Krb5AndCertsSslSocketConnector extends SslSocketConnector {
public static final List<String> KRB5_CIPHER_SUITES = Collections.unmodifiableList(Collections.singletonList("TLS_KRB5_WITH_3DES_EDE_CBC_SHA"));
static {
System.setProperty("https.cipherSuites", KRB5_CIPHER_SUITES.get(0));
}
private static final Logger LOG = LoggerFactory.getLogger(Krb5AndCertsSslSocketConnector.class);
private static final String REMOTE_PRINCIPAL = "remote_principal";
public enum MODE {
KRB, CERTS, BOTH
} // Support Kerberos, certificates or both?
private boolean useKrb;
private boolean useCerts;
public Krb5AndCertsSslSocketConnector() {
// By default, stick to cert based authentication
super();
useKrb = false;
useCerts = true;
setPasswords();
}
public Krb5AndCertsSslSocketConnector(SslContextFactory f, String auth) {
// By default, stick to cert based authentication
super(f);
useKrb = false;
useCerts = true;
setPasswords();
setMode(auth);
}
public static boolean isKrb(String mode) {
return mode == MODE.KRB.toString() || mode == MODE.BOTH.toString();
}
public void setMode(String mode) {
useKrb = mode == MODE.KRB.toString() || mode == MODE.BOTH.toString();
useCerts = mode == MODE.CERTS.toString() || mode == MODE.BOTH.toString();
logIfDebug("useKerb = " + useKrb + ", useCerts = " + useCerts);
}
// If not using Certs, set passwords to random gibberish or else
// Jetty will actually prompt the user for some.
private void setPasswords() {
if (!useCerts) {
Random r = new Random();
System.setProperty("jetty.ssl.password", String.valueOf(r.nextLong()));
System.setProperty("jetty.ssl.keypassword", String.valueOf(r.nextLong()));
}
}
@Override
public SslContextFactory getSslContextFactory() {
final SslContextFactory factory = super.getSslContextFactory();
if (useCerts) {
return factory;
}
try {
SSLContext context = factory.getProvider() == null ? SSLContext.getInstance(factory.getProtocol()) : SSLContext.getInstance(factory.getProtocol(),
factory.getProvider());
context.init(null, null, null);
factory.setSslContext(context);
} catch (NoSuchAlgorithmException e) {
} catch (NoSuchProviderException e) {
} catch (KeyManagementException e) {
}
return factory;
}
/*
* (non-Javadoc)
*
* @see
* org.mortbay.jetty.security.SslSocketConnector#newServerSocket(java.lang
* .String, int, int)
*/
@Override
protected ServerSocket newServerSocket(String host, int port, int backlog) throws IOException {
logIfDebug("Creating new KrbServerSocket for: " + host);
SSLServerSocket ss = null;
if (useCerts) // Get the server socket from the SSL super impl
ss = (SSLServerSocket) super.newServerSocket(host, port, backlog);
else { // Create a default server socket
try {
ss = (SSLServerSocket) super.newServerSocket(host, port, backlog);
} catch (Exception e) {
LOG.warn("Could not create KRB5 Listener", e);
throw new IOException("Could not create KRB5 Listener: " + e.toString());
}
}
// Add Kerberos ciphers to this socket server if needed.
if (useKrb) {
ss.setNeedClientAuth(true);
String[] combined;
if (useCerts) { // combine the cipher suites
String[] certs = ss.getEnabledCipherSuites();
combined = new String[certs.length + KRB5_CIPHER_SUITES.size()];
System.arraycopy(certs, 0, combined, 0, certs.length);
System.arraycopy(KRB5_CIPHER_SUITES.toArray(new String[0]), 0, combined, certs.length, KRB5_CIPHER_SUITES.size());
} else { // Just enable Kerberos auth
combined = KRB5_CIPHER_SUITES.toArray(new String[0]);
}
ss.setEnabledCipherSuites(combined);
}
return ss;
};
@Override
public void customize(EndPoint endpoint, Request request) throws IOException {
if (useKrb) { // Add Kerberos-specific info
SSLSocket sslSocket = (SSLSocket) endpoint.getTransport();
Principal remotePrincipal = sslSocket.getSession().getPeerPrincipal();
logIfDebug("Remote principal = " + remotePrincipal);
request.setScheme(HttpSchemes.HTTPS);
request.setAttribute(REMOTE_PRINCIPAL, remotePrincipal);
if (!useCerts) { // Add extra info that would have been added by
// super
String cipherSuite = sslSocket.getSession().getCipherSuite();
Integer keySize = Integer.valueOf(ServletSSL.deduceKeyLength(cipherSuite));
;
request.setAttribute("javax.servlet.request.cipher_suite", cipherSuite);
request.setAttribute("javax.servlet.request.key_size", keySize);
}
}
if (useCerts)
super.customize(endpoint, request);
}
private void logIfDebug(String s) {
if (LOG.isDebugEnabled())
LOG.debug(s);
}
public class Krb5AndCertsSslSocketConnector {
//
//extends SslSocketConnector {
// public static final List<String> KRB5_CIPHER_SUITES = Collections.unmodifiableList(Collections.singletonList("TLS_KRB5_WITH_3DES_EDE_CBC_SHA"));
// static {
// System.setProperty("https.cipherSuites", KRB5_CIPHER_SUITES.get(0));
// }
//
// private static final Logger LOG = LoggerFactory.getLogger(Krb5AndCertsSslSocketConnector.class);
//
// private static final String REMOTE_PRINCIPAL = "remote_principal";
//
// public enum MODE {
// KRB, CERTS, BOTH
// } // Support Kerberos, certificates or both?
//
// private boolean useKrb;
// private boolean useCerts;
//
// public Krb5AndCertsSslSocketConnector() {
// // By default, stick to cert based authentication
// super();
// useKrb = false;
// useCerts = true;
// setPasswords();
// }
// public Krb5AndCertsSslSocketConnector(SslContextFactory f, String auth) {
// // By default, stick to cert based authentication
// super(f);
// useKrb = false;
// useCerts = true;
// setPasswords();
// setMode(auth);
// }
//
// public static boolean isKrb(String mode) {
// return mode == MODE.KRB.toString() || mode == MODE.BOTH.toString();
// }
//
// public void setMode(String mode) {
// useKrb = mode == MODE.KRB.toString() || mode == MODE.BOTH.toString();
// useCerts = mode == MODE.CERTS.toString() || mode == MODE.BOTH.toString();
// logIfDebug("useKerb = " + useKrb + ", useCerts = " + useCerts);
// }
//
// // If not using Certs, set passwords to random gibberish or else
// // Jetty will actually prompt the user for some.
// private void setPasswords() {
// if (!useCerts) {
// Random r = new Random();
// System.setProperty("jetty.ssl.password", String.valueOf(r.nextLong()));
// System.setProperty("jetty.ssl.keypassword", String.valueOf(r.nextLong()));
// }
// }
//
// @Override
// public SslContextFactory getSslContextFactory() {
// final SslContextFactory factory = super.getSslContextFactory();
//
// if (useCerts) {
// return factory;
// }
//
// try {
// SSLContext context = factory.getProvider() == null ? SSLContext.getInstance(factory.getProtocol()) : SSLContext.getInstance(factory.getProtocol(),
// factory.getProvider());
// context.init(null, null, null);
// factory.setSslContext(context);
// } catch (NoSuchAlgorithmException e) {
// } catch (NoSuchProviderException e) {
// } catch (KeyManagementException e) {
// }
//
// return factory;
// }
//
// /*
// * (non-Javadoc)
// *
// * @see
// * org.mortbay.jetty.security.SslSocketConnector#newServerSocket(java.lang
// * .String, int, int)
// */
// @Override
// protected ServerSocket newServerSocket(String host, int port, int backlog) throws IOException {
// logIfDebug("Creating new KrbServerSocket for: " + host);
// SSLServerSocket ss = null;
//
// if (useCerts) // Get the server socket from the SSL super impl
// ss = (SSLServerSocket) super.newServerSocket(host, port, backlog);
// else { // Create a default server socket
// try {
// ss = (SSLServerSocket) super.newServerSocket(host, port, backlog);
// } catch (Exception e) {
// LOG.warn("Could not create KRB5 Listener", e);
// throw new IOException("Could not create KRB5 Listener: " + e.toString());
// }
// }
//
// // Add Kerberos ciphers to this socket server if needed.
// if (useKrb) {
// ss.setNeedClientAuth(true);
// String[] combined;
// if (useCerts) { // combine the cipher suites
// String[] certs = ss.getEnabledCipherSuites();
// combined = new String[certs.length + KRB5_CIPHER_SUITES.size()];
// System.arraycopy(certs, 0, combined, 0, certs.length);
// System.arraycopy(KRB5_CIPHER_SUITES.toArray(new String[0]), 0, combined, certs.length, KRB5_CIPHER_SUITES.size());
// } else { // Just enable Kerberos auth
// combined = KRB5_CIPHER_SUITES.toArray(new String[0]);
// }
//
// ss.setEnabledCipherSuites(combined);
// }
// return ss;
// };
//
// @Override
// public void customize(EndPoint endpoint, Request request) throws IOException {
// if (useKrb) { // Add Kerberos-specific info
// SSLSocket sslSocket = (SSLSocket) endpoint.getTransport();
// Principal remotePrincipal = sslSocket.getSession().getPeerPrincipal();
// logIfDebug("Remote principal = " + remotePrincipal);
// request.setScheme(HttpSchemes.HTTPS);
// request.setAttribute(REMOTE_PRINCIPAL, remotePrincipal);
//
// if (!useCerts) { // Add extra info that would have been added by
// // super
// String cipherSuite = sslSocket.getSession().getCipherSuite();
// Integer keySize = Integer.valueOf(ServletSSL.deduceKeyLength(cipherSuite));
// ;
//
// request.setAttribute("javax.servlet.request.cipher_suite", cipherSuite);
// request.setAttribute("javax.servlet.request.key_size", keySize);
// }
// }
//
// if (useCerts)
// super.customize(endpoint, request);
// }
//
// private void logIfDebug(String s) {
// if (LOG.isDebugEnabled())
// LOG.debug(s);
// }
}

12
activemq-http/src/main/java/org/apache/activemq/transport/ws/WSTransportServer.java
View File

@ -26,6 +26,7 @@ import javax.servlet.Servlet;
import org.apache.activemq.command.BrokerInfo;
import org.apache.activemq.transport.SocketConnectorFactory;
import org.apache.activemq.transport.WebTransportServerSupport;
import org.apache.activemq.transport.ws.jetty9.WSServlet;
import org.apache.activemq.util.IntrospectionSupport;
import org.apache.activemq.util.ServiceStopper;
import org.eclipse.jetty.server.Connector;
@ -81,7 +82,7 @@ public class WSTransportServer extends WebTransportServerSupport {
// Update the Connect To URI with our actual location in case the configured port
// was set to zero so that we report the actual port we are listening on.
int port = getConnectorLocalPort();
int port = getConnectorLocalPort();
if (port == -1) {
port = boundTo.getPort();
}
@ -98,18 +99,13 @@ public class WSTransportServer extends WebTransportServerSupport {
}
private Servlet createWSServlet() throws Exception {
if (Server.getVersion().startsWith("9")) {
return (Servlet)Class.forName("org.apache.activemq.transport.ws.jetty9.WSServlet", true,
getClass().getClassLoader()).newInstance();
}
return (Servlet)Class.forName("org.apache.activemq.transport.ws.jetty8.WSServlet", true,
getClass().getClassLoader()).newInstance();
return new WSServlet();
}
private int getConnectorLocalPort() throws Exception {
return (Integer)connector.getClass().getMethod("getLocalPort").invoke(connector);
}
@Override
protected void doStop(ServiceStopper stopper) throws Exception {
Server temp = server;

89
activemq-http/src/main/java/org/apache/activemq/transport/ws/jetty8/MQTTSocket.java
View File

@ -1,89 +0,0 @@
/**
* 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.ws.jetty8;
import java.io.IOException;
import org.apache.activemq.transport.ws.AbstractMQTTSocket;
import org.apache.activemq.util.ByteSequence;
import org.apache.activemq.util.IOExceptionSupport;
import org.eclipse.jetty.websocket.WebSocket;
import org.fusesource.mqtt.codec.DISCONNECT;
import org.fusesource.mqtt.codec.MQTTFrame;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class MQTTSocket extends AbstractMQTTSocket implements WebSocket.OnBinaryMessage {
private static final Logger LOG = LoggerFactory.getLogger(MQTTSocket.class);
private Connection outbound;
public MQTTSocket(String remoteAddress) {
super(remoteAddress);
}
@Override
public void sendToMQTT(MQTTFrame command) throws IOException {
ByteSequence bytes = wireFormat.marshal(command);
outbound.sendMessage(bytes.getData(), 0, bytes.getLength());
}
@Override
public void handleStopped() throws IOException {
if (outbound != null && outbound.isOpen()) {
outbound.close();
}
}
//----- WebSocket.OnTextMessage callback handlers ------------------------//
@Override
public void onOpen(Connection connection) {
this.outbound = connection;
}
@Override
public void onMessage(byte[] bytes, int offset, int length) {
if (!transportStartedAtLeastOnce()) {
LOG.debug("Waiting for MQTTSocket to be properly started...");
try {
socketTransportStarted.await();
} catch (InterruptedException e) {
LOG.warn("While waiting for MQTTSocket to be properly started, we got interrupted!! Should be okay, but you could see race conditions...");
}
}
receiveCounter += length;
try {
MQTTFrame frame = (MQTTFrame)wireFormat.unmarshal(new ByteSequence(bytes, offset, length));
getProtocolConverter().onMQTTCommand(frame);
} catch (Exception e) {
onException(IOExceptionSupport.create(e));
}
}
@Override
public void onClose(int closeCode, String message) {
try {
getProtocolConverter().onMQTTCommand(new DISCONNECT().encode());
} catch (Exception e) {
LOG.warn("Failed to close WebSocket", e);
}
}
}

73
activemq-http/src/main/java/org/apache/activemq/transport/ws/jetty8/StompSocket.java
View File

@ -1,73 +0,0 @@
/**
* 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.ws.jetty8;
import java.io.IOException;
import org.apache.activemq.transport.stomp.Stomp;
import org.apache.activemq.transport.stomp.StompFrame;
import org.apache.activemq.transport.ws.AbstractStompSocket;
import org.eclipse.jetty.websocket.WebSocket;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Implements web socket and mediates between servlet and the broker
*/
public class StompSocket extends AbstractStompSocket implements WebSocket.OnTextMessage {
private static final Logger LOG = LoggerFactory.getLogger(StompSocket.class);
private Connection outbound;
public StompSocket(String remoteAddress) {
super(remoteAddress);
}
@Override
public void handleStopped() throws IOException {
if (outbound != null && outbound.isOpen()) {
outbound.close();
}
}
@Override
public void sendToStomp(StompFrame command) throws IOException {
outbound.sendMessage(command.format());
}
//----- WebSocket.OnTextMessage callback handlers ------------------------//
@Override
public void onOpen(Connection connection) {
this.outbound = connection;
}
@Override
public void onClose(int closeCode, String message) {
try {
protocolConverter.onStompCommand(new StompFrame(Stomp.Commands.DISCONNECT));
} catch (Exception e) {
LOG.warn("Failed to close WebSocket", e);
}
}
@Override
public void onMessage(String data) {
processStompFrame(data);
}
}

66
activemq-http/src/main/java/org/apache/activemq/transport/ws/jetty8/WSServlet.java
View File

@ -1,66 +0,0 @@
/**
* 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.ws.jetty8;
import java.io.IOException;
import javax.servlet.ServletException;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import org.apache.activemq.transport.Transport;
import org.apache.activemq.transport.TransportAcceptListener;
import org.apache.activemq.transport.util.HttpTransportUtils;
import org.eclipse.jetty.websocket.WebSocket;
import org.eclipse.jetty.websocket.WebSocketServlet;
/**
* Handle connection upgrade requests and creates web sockets
*/
public class WSServlet extends WebSocketServlet {
private static final long serialVersionUID = -4716657876092884139L;
private TransportAcceptListener listener;
@Override
public void init() throws ServletException {
super.init();
listener = (TransportAcceptListener) getServletContext().getAttribute("acceptListener");
if (listener == null) {
throw new ServletException("No such attribute 'acceptListener' available in the ServletContext");
}
}
@Override
protected void doGet(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
getServletContext().getNamedDispatcher("default").forward(request, response);
}
@Override
public WebSocket doWebSocketConnect(HttpServletRequest request, String protocol) {
WebSocket socket;
if (protocol != null && protocol.startsWith("mqtt")) {
socket = new MQTTSocket(HttpTransportUtils.generateWsRemoteAddress(request));
} else {
socket = new StompSocket(HttpTransportUtils.generateWsRemoteAddress(request));
}
listener.onAccept((Transport) socket);
return socket;
}
}

2
activemq-http/src/main/java/org/apache/activemq/transport/ws/jetty9/MQTTSocket.java
View File

@ -65,6 +65,8 @@ public class MQTTSocket extends AbstractMQTTSocket implements WebSocketListener
}
}
receiveCounter += length;
try {
MQTTFrame frame = (MQTTFrame)wireFormat.unmarshal(new ByteSequence(bytes, offset, length));
getProtocolConverter().onMQTTCommand(frame);

7
activemq-http/src/main/java/org/apache/activemq/transport/ws/jetty9/WSServlet.java
View File

@ -62,7 +62,12 @@ public class WSServlet extends WebSocketServlet {
@Override
public Object createWebSocket(ServletUpgradeRequest req, ServletUpgradeResponse resp) {
WebSocketListener socket;
if (req.getSubProtocols().contains("mqtt")) {
boolean isMqtt = false;
for (String subProtocol : req.getSubProtocols()) {
subProtocol.startsWith("mqtt");
isMqtt = true;
}
if (isMqtt) {
socket = new MQTTSocket(HttpTransportUtils.generateWsRemoteAddress(req.getHttpServletRequest()));
resp.setAcceptedSubProtocol("mqtt");
} else {

59
activemq-http/src/test/java/org/apache/activemq/transport/ws/MQTTWSConnection.java
View File

@ -17,6 +17,7 @@
package org.apache.activemq.transport.ws;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
@ -25,7 +26,9 @@ import java.util.concurrent.TimeUnit;
import org.apache.activemq.transport.mqtt.MQTTWireFormat;
import org.apache.activemq.util.ByteSequence;
import org.eclipse.jetty.websocket.WebSocket;
import org.eclipse.jetty.websocket.api.Session;
import org.eclipse.jetty.websocket.api.WebSocketAdapter;
import org.eclipse.jetty.websocket.api.WebSocketListener;
import org.fusesource.hawtbuf.UTF8Buffer;
import org.fusesource.mqtt.codec.CONNACK;
import org.fusesource.mqtt.codec.CONNECT;
@ -45,13 +48,13 @@ import org.slf4j.LoggerFactory;
/**
* Implements a simple WebSocket based MQTT Client that can be used for unit testing.
*/
public class MQTTWSConnection implements WebSocket, WebSocket.OnBinaryMessage {
public class MQTTWSConnection extends WebSocketAdapter implements WebSocketListener {
private static final Logger LOG = LoggerFactory.getLogger(MQTTWSConnection.class);
private static final MQTTFrame PING_RESP_FRAME = new PINGRESP().encode();
private Connection connection;
private Session connection;
private final CountDownLatch connectLatch = new CountDownLatch(1);
private final MQTTWireFormat wireFormat = new MQTTWireFormat();
@ -60,6 +63,7 @@ public class MQTTWSConnection implements WebSocket, WebSocket.OnBinaryMessage {
private int closeCode = -1;
private String closeMessage;
@Override
public boolean isConnected() {
return connection != null ? connection.isOpen() : false;
}
@ -87,7 +91,7 @@ public class MQTTWSConnection implements WebSocket, WebSocket.OnBinaryMessage {
command.keepAlive((short) 0);
ByteSequence payload = wireFormat.marshal(command.encode());
connection.sendMessage(payload.data, 0, payload.length);
connection.getRemote().sendBytes(ByteBuffer.wrap(payload.data));
MQTTFrame incoming = receive(15, TimeUnit.SECONDS);
if (incoming == null || incoming.messageType() != CONNACK.TYPE) {
@ -102,7 +106,7 @@ public class MQTTWSConnection implements WebSocket, WebSocket.OnBinaryMessage {
DISCONNECT command = new DISCONNECT();
ByteSequence payload = wireFormat.marshal(command.encode());
connection.sendMessage(payload.data, 0, payload.length);
connection.getRemote().sendBytes(ByteBuffer.wrap(payload.data));
}
//---- Send methods ------------------------------------------------------//
@ -110,13 +114,13 @@ public class MQTTWSConnection implements WebSocket, WebSocket.OnBinaryMessage {
public void sendFrame(MQTTFrame frame) throws Exception {
checkConnected();
ByteSequence payload = wireFormat.marshal(frame);
connection.sendMessage(payload.data, 0, payload.length);
connection.getRemote().sendBytes(ByteBuffer.wrap(payload.data));
}
public void keepAlive() throws Exception {
checkConnected();
ByteSequence payload = wireFormat.marshal(new PINGREQ().encode());
connection.sendMessage(payload.data, 0, payload.length);
connection.getRemote().sendBytes(ByteBuffer.wrap(payload.data));
}
//----- Receive methods --------------------------------------------------//
@ -159,7 +163,7 @@ public class MQTTWSConnection implements WebSocket, WebSocket.OnBinaryMessage {
//----- WebSocket callback handlers --------------------------------------//
@Override
public void onMessage(byte[] data, int offset, int length) {
public void onWebSocketBinary(byte[] data, int offset, int length) {
if (data ==null || length <= 0) {
return;
}
@ -228,21 +232,6 @@ public class MQTTWSConnection implements WebSocket, WebSocket.OnBinaryMessage {
}
}
@Override
public void onOpen(Connection connection) {
this.connection = connection;
this.connectLatch.countDown();
}
@Override
public void onClose(int closeCode, String message) {
LOG.trace("MQTT WS Connection closed, code:{} message:{}", closeCode, message);
this.connection = null;
this.closeCode = closeCode;
this.closeMessage = message;
}
//----- Internal implementation ------------------------------------------//
private void checkConnected() throws IOException {
@ -250,4 +239,28 @@ public class MQTTWSConnection implements WebSocket, WebSocket.OnBinaryMessage {
throw new IOException("MQTT WS Connection is closed.");
}
}
/* (non-Javadoc)
* @see org.eclipse.jetty.websocket.api.WebSocketListener#onWebSocketClose(int, java.lang.String)
*/
@Override
public void onWebSocketClose(int statusCode, String reason) {
LOG.trace("MQTT WS Connection closed, code:{} message:{}", statusCode, reason);
this.connection = null;
this.closeCode = statusCode;
this.closeMessage = reason;
}
/* (non-Javadoc)
* @see org.eclipse.jetty.websocket.api.WebSocketListener#onWebSocketConnect(org.eclipse.jetty.websocket.api.Session)
*/
@Override
public void onWebSocketConnect(
org.eclipse.jetty.websocket.api.Session session) {
this.connection = session;
this.connectLatch.countDown();
}
}

23
activemq-http/src/test/java/org/apache/activemq/transport/ws/MQTTWSConnectionTimeoutTest.java
View File

@ -23,8 +23,8 @@ import java.util.Vector;
import java.util.concurrent.TimeUnit;
import org.apache.activemq.util.Wait;
import org.eclipse.jetty.websocket.WebSocketClient;
import org.eclipse.jetty.websocket.WebSocketClientFactory;
import org.eclipse.jetty.websocket.client.ClientUpgradeRequest;
import org.eclipse.jetty.websocket.client.WebSocketClient;
import org.junit.Before;
import org.junit.Test;
@ -39,15 +39,20 @@ public class MQTTWSConnectionTimeoutTest extends WSTransportTestSupport {
@Before
public void setUp() throws Exception {
super.setUp();
WebSocketClientFactory clientFactory = new WebSocketClientFactory();
clientFactory.start();
wsClient = clientFactory.newWebSocketClient();
wsClient.setProtocol("mqttv3.1");
wsMQTTConnection = new MQTTWSConnection();
wsClient.open(wsConnectUri, wsMQTTConnection);
// WebSocketClientFactory clientFactory = new WebSocketClientFactory();
//clientFactory.start();
wsClient = new WebSocketClient();
wsClient.start();
ClientUpgradeRequest request = new ClientUpgradeRequest();
request.setSubProtocols("mqtt");
wsClient.connect(wsMQTTConnection, wsConnectUri, request);
//wsClient.setProtocol("mqttv3.1");
if (!wsMQTTConnection.awaitConnection(30, TimeUnit.SECONDS)) {
throw new IOException("Could not connect to MQTT WS endpoint");
}

23
activemq-http/src/test/java/org/apache/activemq/transport/ws/MQTTWSLinkStealingTest.java
View File

@ -23,8 +23,8 @@ import java.io.IOException;
import java.util.concurrent.TimeUnit;
import org.apache.activemq.util.Wait;
import org.eclipse.jetty.websocket.WebSocketClient;
import org.eclipse.jetty.websocket.WebSocketClientFactory;
import org.eclipse.jetty.websocket.client.ClientUpgradeRequest;
import org.eclipse.jetty.websocket.client.WebSocketClient;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@ -38,21 +38,22 @@ public class MQTTWSLinkStealingTest extends WSTransportTestSupport {
protected WebSocketClient wsClient;
protected MQTTWSConnection wsMQTTConnection;
protected ClientUpgradeRequest request;
@Override
@Before
public void setUp() throws Exception {
super.setUp();
WebSocketClientFactory clientFactory = new WebSocketClientFactory();
clientFactory.start();
wsClient = clientFactory.newWebSocketClient();
wsClient.setProtocol("mqttv3.1");
wsMQTTConnection = new MQTTWSConnection();
wsClient.open(wsConnectUri, wsMQTTConnection);
wsClient = new WebSocketClient();
wsClient.start();
request = new ClientUpgradeRequest();
request.setSubProtocols("mqttv3.1");
wsClient.connect(wsMQTTConnection, wsConnectUri, request);
if (!wsMQTTConnection.awaitConnection(30, TimeUnit.SECONDS)) {
throw new IOException("Could not connect to MQTT WS endpoint");
}
@ -83,12 +84,10 @@ public class MQTTWSLinkStealingTest extends WSTransportTestSupport {
}
}));
WebSocketClientFactory theifFactory = new WebSocketClientFactory();
theifFactory.start();
MQTTWSConnection theif = new MQTTWSConnection();
wsClient.open(wsConnectUri, theif);
wsClient.connect(theif, wsConnectUri, request);
if (!theif.awaitConnection(30, TimeUnit.SECONDS)) {
fail("Could not open new WS connection for link stealing client");
}

17
activemq-http/src/test/java/org/apache/activemq/transport/ws/MQTTWSTransportTest.java
View File

@ -25,8 +25,8 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.activemq.util.Wait;
import org.eclipse.jetty.websocket.WebSocketClient;
import org.eclipse.jetty.websocket.WebSocketClientFactory;
import org.eclipse.jetty.websocket.client.ClientUpgradeRequest;
import org.eclipse.jetty.websocket.client.WebSocketClient;
import org.fusesource.hawtbuf.UTF8Buffer;
import org.fusesource.mqtt.codec.CONNACK;
import org.fusesource.mqtt.codec.CONNECT;
@ -40,21 +40,22 @@ public class MQTTWSTransportTest extends WSTransportTestSupport {
protected WebSocketClient wsClient;
protected MQTTWSConnection wsMQTTConnection;
protected ClientUpgradeRequest request;
@Override
@Before
public void setUp() throws Exception {
super.setUp();
WebSocketClientFactory clientFactory = new WebSocketClientFactory();
clientFactory.start();
wsClient = new WebSocketClient();
wsClient.start();
wsClient = clientFactory.newWebSocketClient();
wsClient.setProtocol("mqttv3.1");
request = new ClientUpgradeRequest();
request.setSubProtocols("mqttv3.1");
wsMQTTConnection = new MQTTWSConnection();
wsClient.open(wsConnectUri, wsMQTTConnection);
wsClient.connect(wsMQTTConnection, wsConnectUri, request);
if (!wsMQTTConnection.awaitConnection(30, TimeUnit.SECONDS)) {
throw new IOException("Could not connect to MQTT WS endpoint");
}
@ -79,7 +80,7 @@ public class MQTTWSTransportTest extends WSTransportTestSupport {
wsMQTTConnection = new MQTTWSConnection();
wsClient.open(wsConnectUri, wsMQTTConnection);
wsClient.connect(wsMQTTConnection, wsConnectUri, request);
if (!wsMQTTConnection.awaitConnection(30, TimeUnit.SECONDS)) {
throw new IOException("Could not connect to MQTT WS endpoint");
}

11
activemq-http/src/test/java/org/apache/activemq/transport/ws/SocketTest.java
View File

@ -18,6 +18,8 @@ package org.apache.activemq.transport.ws;
import static org.junit.Assert.assertEquals;
import org.apache.activemq.transport.ws.jetty9.MQTTSocket;
import org.apache.activemq.transport.ws.jetty9.StompSocket;
import org.junit.Test;
public class SocketTest {
@ -25,8 +27,7 @@ public class SocketTest {
@Test
public void testStompSocketRemoteAddress() {
org.apache.activemq.transport.ws.jetty8.StompSocket stompSocketJetty8 =
new org.apache.activemq.transport.ws.jetty8.StompSocket("ws://localhost:8080");
StompSocket stompSocketJetty8 = new StompSocket("ws://localhost:8080");
assertEquals("ws://localhost:8080", stompSocketJetty8.getRemoteAddress());
@ -39,13 +40,11 @@ public class SocketTest {
@Test
public void testMqttSocketRemoteAddress() {
org.apache.activemq.transport.ws.jetty8.MQTTSocket mqttSocketJetty8 =
new org.apache.activemq.transport.ws.jetty8.MQTTSocket("ws://localhost:8080");
MQTTSocket mqttSocketJetty8 = new MQTTSocket("ws://localhost:8080");
assertEquals("ws://localhost:8080", mqttSocketJetty8.getRemoteAddress());
org.apache.activemq.transport.ws.jetty8.MQTTSocket mqttSocketJetty9 =
new org.apache.activemq.transport.ws.jetty8.MQTTSocket("ws://localhost:8080");
MQTTSocket mqttSocketJetty9 = new MQTTSocket("ws://localhost:8080");
assertEquals("ws://localhost:8080", mqttSocketJetty9.getRemoteAddress());
}

44
activemq-http/src/test/java/org/apache/activemq/transport/ws/StompWSConnection.java
View File

@ -23,18 +23,20 @@ import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;
import org.apache.activemq.transport.stomp.StompFrame;
import org.eclipse.jetty.websocket.WebSocket;
import org.eclipse.jetty.websocket.api.Session;
import org.eclipse.jetty.websocket.api.WebSocketAdapter;
import org.eclipse.jetty.websocket.api.WebSocketListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* STOMP over WS based Connection class
*/
public class StompWSConnection implements WebSocket, WebSocket.OnTextMessage {
public class StompWSConnection extends WebSocketAdapter implements WebSocketListener {
private static final Logger LOG = LoggerFactory.getLogger(StompWSConnection.class);
private Connection connection;
private Session connection;
private final CountDownLatch connectLatch = new CountDownLatch(1);
private final BlockingQueue<String> prefetch = new LinkedBlockingDeque<String>();
@ -42,6 +44,7 @@ public class StompWSConnection implements WebSocket, WebSocket.OnTextMessage {
private int closeCode = -1;
private String closeMessage;
@Override
public boolean isConnected() {
return connection != null ? connection.isOpen() : false;
}
@ -56,17 +59,17 @@ public class StompWSConnection implements WebSocket, WebSocket.OnTextMessage {
public void sendRawFrame(String rawFrame) throws Exception {
checkConnected();
connection.sendMessage(rawFrame);
connection.getRemote().sendString(rawFrame);
}
public void sendFrame(StompFrame frame) throws Exception {
checkConnected();
connection.sendMessage(frame.format());
connection.getRemote().sendString(frame.format());
}
public void keepAlive() throws Exception {
checkConnected();
connection.sendMessage("\n");
connection.getRemote().sendString("\n");
}
//----- Receive methods --------------------------------------------------//
@ -109,7 +112,7 @@ public class StompWSConnection implements WebSocket, WebSocket.OnTextMessage {
//----- WebSocket callback handlers --------------------------------------//
@Override
public void onMessage(String data) {
public void onWebSocketText(String data) {
if (data == null) {
return;
}
@ -122,19 +125,28 @@ public class StompWSConnection implements WebSocket, WebSocket.OnTextMessage {
}
}
@Override
public void onOpen(Connection connection) {
this.connection = connection;
this.connectLatch.countDown();
}
/* (non-Javadoc)
* @see org.eclipse.jetty.websocket.api.WebSocketListener#onWebSocketClose(int, java.lang.String)
*/
@Override
public void onClose(int closeCode, String message) {
LOG.trace("STOMP WS Connection closed, code:{} message:{}", closeCode, message);
public void onWebSocketClose(int statusCode, String reason) {
LOG.trace("STOMP WS Connection closed, code:{} message:{}", statusCode, reason);
this.connection = null;
this.closeCode = closeCode;
this.closeMessage = message;
this.closeCode = statusCode;
this.closeMessage = reason;
}
/* (non-Javadoc)
* @see org.eclipse.jetty.websocket.api.WebSocketListener#onWebSocketConnect(org.eclipse.jetty.websocket.api.Session)
*/
@Override
public void onWebSocketConnect(
org.eclipse.jetty.websocket.api.Session session) {
this.connection = session;
this.connectLatch.countDown();
}
//----- Internal implementation ------------------------------------------//

17
activemq-http/src/test/java/org/apache/activemq/transport/ws/StompWSConnectionTimeoutTest.java
View File

@ -23,8 +23,7 @@ import java.util.Vector;
import java.util.concurrent.TimeUnit;
import org.apache.activemq.util.Wait;
import org.eclipse.jetty.websocket.WebSocketClient;
import org.eclipse.jetty.websocket.WebSocketClientFactory;
import org.eclipse.jetty.websocket.client.WebSocketClient;
import org.junit.Before;
import org.junit.Test;
@ -42,14 +41,16 @@ public class StompWSConnectionTimeoutTest extends WSTransportTestSupport {
@Before
public void setUp() throws Exception {
super.setUp();
WebSocketClientFactory clientFactory = new WebSocketClientFactory();
clientFactory.start();
wsClient = clientFactory.newWebSocketClient();
wsStompConnection = new StompWSConnection();
wsClient.open(wsConnectUri, wsStompConnection);
// WebSocketClientFactory clientFactory = new WebSocketClientFactory();
// clientFactory.start();
wsClient = new WebSocketClient();
wsClient.start();
wsClient.connect(wsStompConnection, wsConnectUri);
if (!wsStompConnection.awaitConnection(30, TimeUnit.SECONDS)) {
throw new IOException("Could not connect to STOMP WS endpoint");
}

13
activemq-http/src/test/java/org/apache/activemq/transport/ws/StompWSTransportTest.java
View File

@ -27,8 +27,7 @@ import java.util.concurrent.TimeUnit;
import org.apache.activemq.transport.stomp.Stomp;
import org.apache.activemq.transport.stomp.StompFrame;
import org.apache.activemq.util.Wait;
import org.eclipse.jetty.websocket.WebSocketClient;
import org.eclipse.jetty.websocket.WebSocketClientFactory;
import org.eclipse.jetty.websocket.client.WebSocketClient;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@ -49,14 +48,12 @@ public class StompWSTransportTest extends WSTransportTestSupport {
@Before
public void setUp() throws Exception {
super.setUp();
WebSocketClientFactory clientFactory = new WebSocketClientFactory();
clientFactory.start();
wsClient = clientFactory.newWebSocketClient();
wsStompConnection = new StompWSConnection();
wsClient.open(wsConnectUri, wsStompConnection);
wsClient = new WebSocketClient();
wsClient.start();
wsClient.connect(wsStompConnection, wsConnectUri);
if (!wsStompConnection.awaitConnection(30, TimeUnit.SECONDS)) {
throw new IOException("Could not connect to STOMP WS endpoint");
}

3
activemq-http/src/test/java/org/apache/activemq/transport/ws/WSTransportTest.java
View File

@ -74,9 +74,6 @@ public class WSTransportTest extends WSTransportTestSupport {
Server server = new Server();
Connector connector = createJettyConnector(server);
if (Server.getVersion().startsWith("8")) {
connector.setServer(server);
}
WebAppContext context = new WebAppContext();
context.setResourceBase("src/test/webapp");

9
activemq-http/src/test/java/org/apache/activemq/transport/ws/WSTransportTestSupport.java
View File

@ -69,11 +69,18 @@ public class WSTransportTestSupport {
LOG.info("========== Finished test: {} ==========", name.getMethodName());
}
// protected String getWSConnectorURI() {
// return "ws://127.0.0.1:" + getProxyPort() +
// "?allowLinkStealing=" + isAllowLinkStealing() +
// "&websocket.maxTextMessageSize=99999&" +
// "transport.maxIdleTime=1001";
// }
protected String getWSConnectorURI() {
return "ws://127.0.0.1:" + getProxyPort() +
"?allowLinkStealing=" + isAllowLinkStealing() +
"&websocket.maxTextMessageSize=99999&" +
"transport.maxIdleTime=1001";
"transport.idleTimeout=1001";
}
protected boolean isAllowLinkStealing() {

7
activemq-http/src/test/java/org/apache/activemq/transport/wss/WSSTransportTest.java
View File

@ -20,6 +20,7 @@ import org.apache.activemq.transport.SecureSocketConnectorFactory;
import org.apache.activemq.transport.ws.WSTransportTest;
import org.eclipse.jetty.server.Connector;
import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.server.ServerConnector;
public class WSSTransportTest extends WSTransportTest {
@Override
@ -29,9 +30,9 @@ public class WSSTransportTest extends WSTransportTest {
sscf.setKeyStorePassword("password");
sscf.setTrustStore("src/test/resources/client.keystore");
sscf.setTrustStorePassword("password");
Connector c = sscf.createConnector(server);
c.getClass().getMethod("setPort", Integer.TYPE).invoke(c, getProxyPort());
ServerConnector c = (ServerConnector) sscf.createConnector(server);
c.setPort(getProxyPort());
return c;
}

13
activemq-http/src/webapp/WEB-INF/web.xml
View File

@ -16,11 +16,10 @@
limitations under the License.
-->
<!DOCTYPE web-app
PUBLIC "-//Sun Microsystems, Inc.//DTD Web Application 2.3//EN"
"http://java.sun.com/dtd/web-app_2_3.dtd">
<web-app>
<web-app xmlns="http://java.sun.com/xml/ns/javaee"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_3_0.xsd"
version="3.0">
<display-name>ActiveMQ Message Broker Web Application</display-name>
<description>
@ -29,9 +28,9 @@
<!-- context config -->
<context-param>
<description>The URL that the embedded broker should listen on in addition to HTTP</description>
<param-name>org.apache.activemq.brokerURL</param-name>
<param-value>tcp://localhost:61616</param-value>
<description>The URL that the embedded broker should listen on in addition to HTTP</description>
<param-value>tcp://localhost:61616</param-value>
</context-param>

2
activemq-osgi/pom.xml
View File

@ -52,7 +52,7 @@
com.fasterxml.jackson*;resolution:=optional,
org.codehaus.jettison*;resolution:=optional,
org.jasypt*;resolution:=optional,
org.eclipse.jetty*;resolution:=optional;version="[8.1,10)",
org.eclipse.jetty*;resolution:=optional;version="[9.0,10)",
org.apache.zookeeper*;resolution:=optional,
org.fusesource.leveldbjni*;resolution:=optional,
org.fusesource.hawtjni*;resolution:=optional,

78
activemq-web-console/pom.xml
View File

@ -32,7 +32,7 @@
<properties>
<jetty.port>8080</jetty.port>
<jetty.maven.groupid>org.mortbay.jetty</jetty.maven.groupid>
<jetty.maven.groupid>org.eclipse.jetty</jetty.maven.groupid>
</properties>
<build>
@ -57,7 +57,7 @@
<version>${jetty-version}</version>
<configuration>
<connectors>
<connector implementation="org.eclipse.jetty.server.nio.SelectChannelConnector">
<connector implementation="org.eclipse.jetty.server.ServerConnector">
<port>${jetty.port}</port>
<maxIdleTime>60000</maxIdleTime>
</connector>
@ -158,7 +158,7 @@
org.slf4j.spi;version="[1.6,2)";resolution:=optional,
org.apache.log4j;version="[1.2.14,2)";resolution:=optional,
org.apache.log4j.spi;version="[1.2.14,2)";resolution:=optional,
org.eclipse.jetty*;resolution:=optional;version="[8.1,10)"
org.eclipse.jetty*;resolution:=optional;version="[9.0,10)"
</Import-Package>
</instructions>
</configuration>
@ -167,18 +167,17 @@
</build>
<dependencies>
<dependency>
<groupId>org.apache.geronimo.specs</groupId>
<artifactId>geronimo-servlet_3.0_spec</artifactId>
<groupId>org.apache.tomcat</groupId>
<artifactId>tomcat-servlet-api</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.geronimo.specs</groupId>
<artifactId>geronimo-jsp_2.1_spec</artifactId>
<scope>provided</scope>
<groupId>org.apache.tomcat</groupId>
<artifactId>tomcat-websocket-api</artifactId>
<scope>provided</scope>
</dependency>
<!-- j2ee jars -->
<dependency>
<groupId>org.apache.geronimo.specs</groupId>
@ -207,8 +206,8 @@
<artifactId>activemq-all</artifactId>
</exclusion>
<exclusion>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-websocket</artifactId>
<groupId>org.eclipse.jetty.websocket</groupId>
<artifactId>websocket-server</artifactId>
</exclusion>
<exclusion>
<groupId>org.eclipse.jetty</groupId>
@ -247,10 +246,10 @@
<artifactId>jetty-all</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.eclipse.jdt</groupId>
<artifactId>core</artifactId>
<version>3.1.1</version>
<dependency>
<groupId>org.eclipse.jetty.orbit</groupId>
<artifactId>org.eclipse.jdt.core</artifactId>
<version>3.8.2.v20130121</version>
<scope>test</scope>
</dependency>
<dependency>
@ -267,15 +266,13 @@
</dependency>
<!-- JSTL support -->
<dependency>
<groupId>javax.servlet</groupId>
<artifactId>jstl</artifactId>
<groupId>org.eclipse.jetty</groupId>
<artifactId>apache-jstl</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>taglibs</groupId>
<artifactId>standard</artifactId>
</dependency>
<!-- used for testing -->
<dependency>
<groupId>junit</groupId>
@ -283,6 +280,41 @@
<scope>test</scope>
</dependency>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>activemq-unit-tests</artifactId>
<scope>test</scope>
<type>test-jar</type>
</dependency>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>activemq-broker</artifactId>
<type>test-jar</type>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>apache-jsp</artifactId>
<scope>provided</scope>
<exclusions>
<exclusion>
<groupId>javax.servlet</groupId>
<artifactId>javax.servlet-api</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.geronimo.specs</groupId>
<artifactId>geronimo-annotation_1.0_spec</artifactId>
</dependency>
<dependency>
<groupId>org.ow2.asm</groupId>
<artifactId>asm</artifactId>
</dependency>
<dependency>
<groupId>org.osgi</groupId>
<artifactId>org.osgi.core</artifactId>

2
activemq-web-console/src/main/webapp/WEB-INF/activemq.xml
View File

@ -30,7 +30,7 @@
<broker brokerName="web-console" useJmx="true" xmlns="http://activemq.apache.org/schema/core">
<persistenceAdapter>
<kahaDB directory="${activemq.data}/kahadb"/>
<kahaDB directory="target/kahadb"/>
</persistenceAdapter>
<transportConnectors>

9
activemq-web-console/src/main/webapp/WEB-INF/web.xml
View File

@ -16,11 +16,10 @@
limitations under the License.
-->
<web-app xmlns="http://java.sun.com/xml/ns/j2ee"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://java.sun.com/xml/ns/j2ee
http://java.sun.com/xml/ns/j2ee/web-app_2_4.xsd"
version="2.4">
<web-app xmlns="http://java.sun.com/xml/ns/javaee"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_3_0.xsd"
version="3.0">
<description>
Apache ActiveMQ Web Console

18
activemq-web-console/src/test/java/org/apache/activemq/web/tool/Main.java
View File

@ -17,15 +17,16 @@
package org.apache.activemq.web.tool;
import org.eclipse.jetty.server.Connector;
import org.apache.activemq.web.config.JspConfigurer;
import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.server.handler.ContextHandlerCollection;
import org.eclipse.jetty.webapp.WebAppContext;
/**
* A simple bootstrap class for starting Jetty in your IDE using the local web
* application.
*
*
*
*
*/
public final class Main {
@ -45,6 +46,8 @@ public final class Main {
String text = args[0];
port = Integer.parseInt(text);
}
System.setProperty("activemq.conf", "/home/clshann/dev/git/apache-activemq/assembly/target/apache-activemq-5.13-SNAPSHOT/conf");
System.out.println("Starting Web Server on port: " + port);
System.setProperty("jetty.port", "" + port);
Server server = new Server(port);
@ -54,11 +57,16 @@ public final class Main {
//System.setProperty("webconsole.jmx.url","service:jmx:rmi:///jndi/rmi://localhost:1099/karaf-root");
WebAppContext context = new WebAppContext();
ContextHandlerCollection handlers = new ContextHandlerCollection();
handlers.setHandlers(new WebAppContext[] {context});
JspConfigurer.configureJetty(server, handlers);
context.setResourceBase(WEBAPP_DIR);
context.setContextPath(WEBAPP_CTX);
context.setServer(server);
server.setHandler(context);
server.setHandler(handlers);
server.start();
System.out.println();
@ -67,4 +75,6 @@ public final class Main {
System.out.println("==============================================================================");
System.out.println();
}
}

5
activemq-web-demo/pom.xml
View File

@ -33,12 +33,11 @@
<build>
<plugins>
<plugin>
<groupId>org.mortbay.jetty</groupId>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-maven-plugin</artifactId>
<version>${jetty-version}</version>
<configuration>
<connectors>
<connector implementation="org.eclipse.jetty.server.nio.SelectChannelConnector">
<connector implementation="org.eclipse.jetty.server.ServerConnector">
<port>${jetty.port}</port>
<maxIdleTime>60000</maxIdleTime>
</connector>

18
activemq-web-demo/src/main/webapp/WEB-INF/web.xml
View File

@ -15,25 +15,25 @@
See the License for the specific language governing permissions and
limitations under the License.
-->
<!DOCTYPE web-app
PUBLIC "-//Sun Microsystems, Inc.//DTD Web Application 2.3//EN"
"http://java.sun.com/dtd/web-app_2_3.dtd">
<web-app>
<web-app xmlns="http://java.sun.com/xml/ns/javaee"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_3_0.xsd"
version="3.0">
<display-name>Apache ActiveMQ Web Demo</display-name>
<!-- context config -->
<context-param>
<description>Whether we should include an embedded broker or not</description>
<param-name>org.apache.activemq.embeddedBroker</param-name>
<param-value>true</param-value>
<description>Whether we should include an embedded broker or not</description>
<param-value>true</param-value>
</context-param>
<!-- filters -->
<filter>
<filter-name>session</filter-name>
<filter-class>org.apache.activemq.web.SessionFilter</filter-class>
<async-supported>true</async-supported>
</filter>
<filter-mapping>
<filter-name>session</filter-name>
@ -52,12 +52,14 @@
<servlet-name>AjaxServlet</servlet-name>
<servlet-class>org.apache.activemq.web.AjaxServlet</servlet-class>
<load-on-startup>1</load-on-startup>
<async-supported>true</async-supported>
</servlet>
<servlet>
<servlet-name>MessageServlet</servlet-name>
<servlet-class>org.apache.activemq.web.MessageServlet</servlet-class>
<load-on-startup>1</load-on-startup>
<async-supported>true</async-supported>
<!--
Uncomment this parameter if you plan to use multiple consumers over REST
<init-param>
@ -71,6 +73,7 @@
<servlet>
<servlet-name>QueueBrowseServlet</servlet-name>
<servlet-class>org.apache.activemq.web.QueueBrowseServlet</servlet-class>
<async-supported>true</async-supported>
</servlet>
<!-- servlets for the portfolio demo -->
@ -78,6 +81,7 @@
<servlet-name>PortfolioPublishServlet</servlet-name>
<servlet-class>org.apache.activemq.web.PortfolioPublishServlet</servlet-class>
<load-on-startup>1</load-on-startup>
<async-supported>true</async-supported>
</servlet>
<!-- servlet mappings -->

447
activemq-web-demo/src/test/java/org/apache/activemq/web/AjaxTest.java
View File

@ -16,9 +16,15 @@
*/
package org.apache.activemq.web;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import java.io.ByteArrayInputStream;
import java.net.SocketTimeoutException;
import java.util.HashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import javax.jms.Message;
import javax.jms.MessageProducer;
@ -27,43 +33,20 @@ import org.apache.activemq.transport.stomp.Stomp;
import org.apache.activemq.transport.stomp.StompConnection;
import org.apache.activemq.transport.stomp.StompFrame;
import org.apache.commons.lang.StringUtils;
import org.eclipse.jetty.client.ContentExchange;
import org.eclipse.jetty.client.HttpClient;
import org.eclipse.jetty.io.Buffer;
import org.eclipse.jetty.io.ByteArrayBuffer;
import org.eclipse.jetty.client.api.Request;
import org.eclipse.jetty.client.api.Result;
import org.eclipse.jetty.client.util.BufferingResponseListener;
import org.eclipse.jetty.client.util.InputStreamContentProvider;
import org.eclipse.jetty.http.HttpHeader;
import org.eclipse.jetty.http.HttpMethod;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static org.junit.Assert.*;
public class AjaxTest extends JettyTestSupport {
private static final Logger LOG = LoggerFactory.getLogger(AjaxTest.class);
private class AjaxTestContentExchange extends ContentExchange {
private HashMap<String,String> headers;
private String responseContent;
AjaxTestContentExchange() {
super(true);
this.headers = new HashMap<String,String>();
this.responseContent = "";
}
protected void onResponseContent( Buffer content ) {
this.responseContent += content.toString();
}
protected void onResponseHeader( Buffer name, Buffer value ) {
headers.put( name.toString(), value.toString() );
}
public String getJsessionId() {
String cookie = headers.get( "Set-Cookie" );
String[] cookie_parts = cookie.split( ";" );
return cookie_parts[0];
}
public String getResponseContent() {
return responseContent;
}
}
public void assertContains( String expected, String actual ) {
assertTrue( "'"+actual+"' does not contain expected fragment '"+expected+"'", actual.indexOf( expected ) != -1 );
@ -79,56 +62,37 @@ public class AjaxTest extends JettyTestSupport {
int port = getPort();
HttpClient httpClient = new HttpClient();
httpClient.setConnectorType(HttpClient.CONNECTOR_SELECT_CHANNEL);
httpClient.start();
// client 1 subscribes to a queue
LOG.debug( "SENDING LISTEN" );
AjaxTestContentExchange contentExchange = new AjaxTestContentExchange();
contentExchange.setMethod( "POST" );
contentExchange.setURL("http://localhost:" + port + "/amq");
contentExchange.setRequestContent( new ByteArrayBuffer("destination=queue://test&type=listen&message=handler") );
contentExchange.setRequestContentType( "application/x-www-form-urlencoded; charset=UTF-8" );
httpClient.send(contentExchange);
contentExchange.waitForDone();
String jsessionid = contentExchange.getJsessionId();
String sessionId = subscribe(httpClient, port, "destination=queue://test&type=listen&message=handler");
// client 1 polls for messages
LOG.debug( "SENDING POLL" );
AjaxTestContentExchange poll = new AjaxTestContentExchange();
poll.setMethod( "GET" );
poll.setURL("http://localhost:" + port + "/amq?timeout=5000");
poll.setRequestHeader( "Cookie", jsessionid );
httpClient.send( poll );
final StringBuffer buf = new StringBuffer();
final CountDownLatch latch =
asyncRequest(httpClient, "http://localhost:" + port + "/amq?timeout=5000", buf, sessionId);
// while client 1 is polling, client 2 sends messages to the queue
LOG.debug( "SENDING MESSAGES" );
contentExchange = new AjaxTestContentExchange();
contentExchange.setMethod( "POST" );
contentExchange.setURL("http://localhost:" + port + "/amq");
contentExchange.setRequestContent( new ByteArrayBuffer(
"destination=queue://test&type=send&message=msg1&"+
"d1=queue://test&t1=send&m1=msg2&"+
"d2=queue://test&t2=send&m2=msg3"
) );
contentExchange.setRequestContentType( "application/x-www-form-urlencoded; charset=UTF-8" );
httpClient.send(contentExchange);
contentExchange.waitForDone();
sendMessages(httpClient, port, ("destination=queue://test&type=send&message=msg1&"+
"d1=queue://test&t1=send&m1=msg2&"+
"d2=queue://test&t2=send&m2=msg3").getBytes());
LOG.debug( "DONE POSTING MESSAGES" );
// wait for poll to finish
poll.waitForDone();
String response = poll.getResponseContent();
latch.await();
String response = buf.toString();
// messages might not all be delivered during the 1st poll. We need to check again.
poll = new AjaxTestContentExchange();
poll.setMethod( "GET" );
poll.setURL("http://localhost:" + port + "/amq?timeout=5000");
poll.setRequestHeader( "Cookie", jsessionid );
httpClient.send( poll );
poll.waitForDone();
final StringBuffer buf2 = new StringBuffer();
final CountDownLatch latch2 =
asyncRequest(httpClient, "http://localhost:" + port + "/amq?timeout=5000", buf2, sessionId);
latch2.await();
String fullResponse = response + poll.getResponseContent();
String fullResponse = response + buf2.toString();
LOG.debug( "full response : " + fullResponse );
assertContains( "<response id='handler' destination='queue://test' >msg1</response>", fullResponse );
assertContains( "<response id='handler' destination='queue://test' >msg2</response>", fullResponse );
@ -144,56 +108,37 @@ public class AjaxTest extends JettyTestSupport {
int port = getPort();
HttpClient httpClient = new HttpClient();
httpClient.setConnectorType(HttpClient.CONNECTOR_SELECT_CHANNEL);
httpClient.start();
// client 1 subscribes to a queue
LOG.debug( "SENDING LISTEN" );
AjaxTestContentExchange contentExchange = new AjaxTestContentExchange();
contentExchange.setMethod( "POST" );
contentExchange.setURL("http://localhost:" + port + "/amq");
contentExchange.setRequestContent( new ByteArrayBuffer("destination=topic://test&type=listen&message=handler") );
contentExchange.setRequestContentType( "application/x-www-form-urlencoded; charset=UTF-8" );
httpClient.send(contentExchange);
contentExchange.waitForDone();
String jsessionid = contentExchange.getJsessionId();
String sessionId = subscribe(httpClient, port, "destination=topic://test&type=listen&message=handler");
// client 1 polls for messages
LOG.debug( "SENDING POLL" );
AjaxTestContentExchange poll = new AjaxTestContentExchange();
poll.setMethod( "GET" );
poll.setURL("http://localhost:" + port + "/amq?timeout=5000");
poll.setRequestHeader( "Cookie", jsessionid );
httpClient.send( poll );
final StringBuffer buf = new StringBuffer();
final CountDownLatch latch =
asyncRequest(httpClient, "http://localhost:" + port + "/amq?timeout=5000", buf, sessionId);
// while client 1 is polling, client 2 sends messages to the queue
LOG.debug( "SENDING MESSAGES" );
contentExchange = new AjaxTestContentExchange();
contentExchange.setMethod( "POST" );
contentExchange.setURL("http://localhost:" + port + "/amq");
contentExchange.setRequestContent( new ByteArrayBuffer(
"destination=topic://test&type=send&message=msg1&"+
"d1=topic://test&t1=send&m1=msg2&"+
"d2=topic://test&t2=send&m2=msg3"
) );
contentExchange.setRequestContentType( "application/x-www-form-urlencoded; charset=UTF-8" );
httpClient.send(contentExchange);
contentExchange.waitForDone();
LOG.debug( "DONE POSTING MESSAGES" );
sendMessages(httpClient, port, ("destination=topic://test&type=send&message=msg1&"+
"d1=topic://test&t1=send&m1=msg2&"+
"d2=topic://test&t2=send&m2=msg3").getBytes());
// wait for poll to finish
poll.waitForDone();
String response = poll.getResponseContent();
latch.await();
String response = buf.toString();
// not all messages might be delivered during the 1st poll. We need to check again.
poll = new AjaxTestContentExchange();
poll.setMethod( "GET" );
poll.setURL("http://localhost:" + port + "/amq?timeout=5000");
poll.setRequestHeader( "Cookie", jsessionid );
httpClient.send( poll );
poll.waitForDone();
// messages might not all be delivered during the 1st poll. We need to
// check again.
final StringBuffer buf2 = new StringBuffer();
final CountDownLatch latch2 = asyncRequest(httpClient,
"http://localhost:" + port + "/amq?timeout=5000", buf2, sessionId);
latch2.await();
String fullResponse = response + buf2.toString();
String fullResponse = response + poll.getResponseContent();
LOG.debug( "full response : " + fullResponse );
assertContains( "<response id='handler' destination='topic://test' >msg1</response>", fullResponse );
@ -215,31 +160,21 @@ public class AjaxTest extends JettyTestSupport {
producer.send( session.createTextMessage("test three") );
HttpClient httpClient = new HttpClient();
httpClient.setConnectorType(HttpClient.CONNECTOR_SELECT_CHANNEL);
httpClient.start();
// client 1 subscribes to queue
LOG.debug( "SENDING LISTEN" );
AjaxTestContentExchange contentExchange = new AjaxTestContentExchange();
contentExchange.setMethod( "POST" );
contentExchange.setURL("http://localhost:" + port + "/amq");
contentExchange.setRequestContent( new ByteArrayBuffer("destination=queue://test&type=listen&message=handler") );
contentExchange.setRequestContentType( "application/x-www-form-urlencoded; charset=UTF-8" );
httpClient.send(contentExchange);
contentExchange.waitForDone();
String jsessionid = contentExchange.getJsessionId();
String sessionId = subscribe(httpClient, port, "destination=queue://test&type=listen&message=handler");
// client 1 polls for messages
LOG.debug( "SENDING POLL" );
AjaxTestContentExchange poll = new AjaxTestContentExchange();
poll.setMethod( "GET" );
poll.setURL("http://localhost:" + port + "/amq?timeout=5000");
poll.setRequestHeader( "Cookie", jsessionid );
httpClient.send( poll );
final StringBuffer buf = new StringBuffer();
final CountDownLatch latch =
asyncRequest(httpClient, "http://localhost:" + port + "/amq?timeout=5000", buf, sessionId);
// wait for poll to finish
poll.waitForDone();
String response = poll.getResponseContent();
latch.await();
String response = buf.toString();
assertContains( "<response id='handler' destination='queue://test' >test one</response>", response );
assertContains( "<response id='handler' destination='queue://test' >test two</response>", response );
@ -255,27 +190,17 @@ public class AjaxTest extends JettyTestSupport {
int port = getPort();
HttpClient httpClient = new HttpClient();
httpClient.setConnectorType(HttpClient.CONNECTOR_SELECT_CHANNEL);
httpClient.start();
// client 1 subscribes to a queue
LOG.debug( "SENDING LISTEN" );
AjaxTestContentExchange contentExchange = new AjaxTestContentExchange();
contentExchange.setMethod( "POST" );
contentExchange.setURL("http://localhost:" + port + "/amq");
contentExchange.setRequestContent( new ByteArrayBuffer("destination=queue://test&type=listen&message=handler") );
contentExchange.setRequestContentType( "application/x-www-form-urlencoded; charset=UTF-8" );
httpClient.send(contentExchange);
contentExchange.waitForDone();
String jsessionid = contentExchange.getJsessionId();
String sessionId = subscribe(httpClient, port, "destination=queue://test&type=listen&message=handler");
// client 1 polls for messages
LOG.debug( "SENDING POLL" );
AjaxTestContentExchange poll = new AjaxTestContentExchange();
poll.setMethod( "GET" );
poll.setURL("http://localhost:" + port + "/amq?timeout=5000");
poll.setRequestHeader( "Cookie", jsessionid );
httpClient.send( poll );
final StringBuffer buf = new StringBuffer();
final CountDownLatch latch =
asyncRequest(httpClient, "http://localhost:" + port + "/amq?timeout=5000", buf, sessionId);
// stomp client queues some messages
StompConnection connection = new StompConnection();
@ -298,18 +223,16 @@ public class AjaxTest extends JettyTestSupport {
connection.disconnect();
// wait for poll to finish
poll.waitForDone();
String response = poll.getResponseContent();
latch.await();
String response = buf.toString();
// not all messages might be delivered during the 1st poll. We need to check again.
poll = new AjaxTestContentExchange();
poll.setMethod( "GET" );
poll.setURL("http://localhost:" + port + "/amq?timeout=5000");
poll.setRequestHeader( "Cookie", jsessionid );
httpClient.send( poll );
poll.waitForDone();
final StringBuffer buf2 = new StringBuffer();
final CountDownLatch latch2 =
asyncRequest(httpClient, "http://localhost:" + port + "/amq?timeout=5000", buf2, sessionId);
latch2.await();
String fullResponse = response + poll.getResponseContent();
String fullResponse = response + buf2.toString();
assertContains( "<response id='handler' destination='queue://test' >message1</response>", fullResponse );
assertContains( "<response id='handler' destination='queue://test' >message2</response>", fullResponse );
@ -327,20 +250,12 @@ public class AjaxTest extends JettyTestSupport {
int port = getPort();
HttpClient httpClient = new HttpClient();
httpClient.setConnectorType(HttpClient.CONNECTOR_SELECT_CHANNEL);
httpClient.start();
AjaxTestContentExchange contentExchange = new AjaxTestContentExchange();
contentExchange.setMethod( "POST" );
contentExchange.setURL("http://localhost:" + port + "/amq");
contentExchange.setRequestContent( new ByteArrayBuffer(
"destination=queue://test&type=send&message=msg1&"+
"d1=queue://test&t1=send&m1=msg2&"+
"d2=queue://test&t2=send&m2=msg3&"+
"d3=queue://test&t3=send&m3=msg4") );
contentExchange.setRequestContentType( "application/x-www-form-urlencoded; charset=UTF-8" );
httpClient.send(contentExchange);
contentExchange.waitForDone();
sendMessages(httpClient, port, ("destination=queue://test&type=send&message=msg1&"+
"d1=queue://test&t1=send&m1=msg2&"+
"d2=queue://test&t2=send&m2=msg3&"+
"d3=queue://test&t3=send&m3=msg4").getBytes());
StompConnection connection = new StompConnection();
connection.open(stompUri.getHost(), stompUri.getPort());
@ -371,6 +286,7 @@ public class AjaxTest extends JettyTestSupport {
LOG.debug( "*** testAjaxClientMayUseSelectors ***" );
int port = getPort();
// send 2 messages to the same queue w/ different 'filter' values.
Message msg = session.createTextMessage("test one");
msg.setStringProperty( "filter", "one" );
@ -380,35 +296,23 @@ public class AjaxTest extends JettyTestSupport {
producer.send( msg );
HttpClient httpClient = new HttpClient();
httpClient.setConnectorType(HttpClient.CONNECTOR_SELECT_CHANNEL);
httpClient.start();
// client subscribes to queue
LOG.debug( "SENDING LISTEN" );
AjaxTestContentExchange contentExchange = new AjaxTestContentExchange();
contentExchange.setMethod( "POST" );
contentExchange.setURL("http://localhost:" + port + "/amq");
contentExchange.setRequestContent( new ByteArrayBuffer("destination=queue://test&type=listen&message=handler") );
contentExchange.setRequestContentType( "application/x-www-form-urlencoded; charset=UTF-8" );
// SELECTOR
contentExchange.setRequestHeader( "selector", "filter='two'" );
httpClient.send(contentExchange);
contentExchange.waitForDone();
String jsessionid = contentExchange.getJsessionId();
String sessionId = subscribe(httpClient, port, "destination=queue://test&type=listen&message=handler", "filter='two'", null);
// client 1 polls for messages
LOG.debug( "SENDING POLL" );
AjaxTestContentExchange poll = new AjaxTestContentExchange();
poll.setMethod( "GET" );
poll.setURL("http://localhost:" + port + "/amq?timeout=5000");
poll.setRequestHeader( "Cookie", jsessionid );
httpClient.send( poll );
poll.waitForDone();
final StringBuffer buf = new StringBuffer();
final CountDownLatch latch =
asyncRequest(httpClient, "http://localhost:" + port + "/amq?timeout=5000", buf, sessionId);
latch.await();
LOG.debug( poll.getResponseContent() );
LOG.debug( buf.toString() );
String expected = "<response id='handler' destination='queue://test' >test two</response>";
assertContains( expected, poll.getResponseContent() );
assertContains( expected, buf.toString() );
httpClient.stop();
}
@ -427,68 +331,39 @@ public class AjaxTest extends JettyTestSupport {
producerB.send( session.createTextMessage("B2") );
HttpClient httpClient = new HttpClient();
httpClient.setConnectorType(HttpClient.CONNECTOR_SELECT_CHANNEL);
httpClient.start();
// clientA subscribes to /queue/testA
LOG.debug( "SENDING LISTEN" );
AjaxTestContentExchange contentExchange = new AjaxTestContentExchange();
contentExchange.setMethod( "POST" );
contentExchange.setURL("http://localhost:" + port + "/amq");
contentExchange.setRequestContent( new ByteArrayBuffer(
"destination=queue://testA&"+
"type=listen&"+
"message=handlerA&"+
"clientId=clientA"
) );
contentExchange.setRequestContentType( "application/x-www-form-urlencoded; charset=UTF-8" );
httpClient.send(contentExchange);
contentExchange.waitForDone();
String jsessionid = contentExchange.getJsessionId();
String sessionId = subscribe(httpClient, port, "destination=queue://testA&type=listen&message=handlerA&clientId=clientA");
// clientB subscribes to /queue/testB using the same JSESSIONID.
contentExchange = new AjaxTestContentExchange();
contentExchange.setMethod( "POST" );
contentExchange.setURL("http://localhost:" + port + "/amq");
contentExchange.setRequestHeader( "Cookie", jsessionid );
contentExchange.setRequestContent( new ByteArrayBuffer(
"destination=queue://testB&"+
"type=listen&"+
"message=handlerB&"+
"clientId=clientB"
) );
contentExchange.setRequestContentType( "application/x-www-form-urlencoded; charset=UTF-8" );
httpClient.send(contentExchange);
contentExchange.waitForDone();
subscribe(httpClient, port, "destination=queue://testB&type=listen&message=handlerB&clientId=clientB", null, sessionId);
// clientA polls for messages
AjaxTestContentExchange poll = new AjaxTestContentExchange();
poll.setMethod( "GET" );
poll.setURL("http://localhost:" + port + "/amq?timeout=5000&clientId=clientA");
poll.setRequestHeader( "Cookie", jsessionid );
httpClient.send( poll );
poll.waitForDone();
final StringBuffer buf = new StringBuffer();
final CountDownLatch latch =
asyncRequest(httpClient, "http://localhost:" + port + "/amq?timeout=5000&clientId=clientA", buf, sessionId);
latch.await();
LOG.debug( "clientA response : " + poll.getResponseContent() );
LOG.debug( "clientA response : " + buf.toString() );
String expected1 = "<response id='handlerA' destination='queue://testA' >A1</response>";
String expected2 = "<response id='handlerA' destination='queue://testA' >A2</response>";
assertContains( expected1, poll.getResponseContent() );
assertContains( expected2, poll.getResponseContent() );
assertContains( expected1, buf.toString() );
assertContains( expected2, buf.toString() );
// clientB polls for messages
poll = new AjaxTestContentExchange();
poll.setMethod( "GET" );
poll.setURL("http://localhost:" + port + "/amq?timeout=5000&clientId=clientB");
poll.setRequestHeader( "Cookie", jsessionid );
httpClient.send( poll );
poll.waitForDone();
final StringBuffer buf2 = new StringBuffer();
final CountDownLatch latch2 =
asyncRequest(httpClient, "http://localhost:" + port + "/amq?timeout=5000&clientId=clientB", buf2, sessionId);
latch2.await();
LOG.debug( "clientB response : " + poll.getResponseContent() );
LOG.debug( "clientB response : " + buf2.toString() );
expected1 = "<response id='handlerB' destination='queue://testB' >B1</response>";
expected2 = "<response id='handlerB' destination='queue://testB' >B2</response>";
assertContains( expected1, poll.getResponseContent() );
assertContains( expected2, poll.getResponseContent() );
assertContains( expected1, buf2.toString() );
assertContains( expected2, buf2.toString() );
httpClient.stop();
}
@ -499,66 +374,39 @@ public class AjaxTest extends JettyTestSupport {
int port = getPort();
HttpClient httpClient = new HttpClient();
httpClient.setConnectorType(HttpClient.CONNECTOR_SELECT_CHANNEL);
httpClient.start();
LOG.debug( "SENDING LISTEN FOR /topic/topicA" );
AjaxTestContentExchange contentExchange = new AjaxTestContentExchange();
contentExchange.setMethod( "POST" );
contentExchange.setURL("http://localhost:" + port + "/amq");
contentExchange.setRequestContent( new ByteArrayBuffer("destination=topic://topicA&type=listen&message=handlerA") );
contentExchange.setRequestContentType( "application/x-www-form-urlencoded; charset=UTF-8" );
httpClient.send(contentExchange);
contentExchange.waitForDone();
String jsessionid = contentExchange.getJsessionId();
String sessionId = subscribe(httpClient, port, "destination=topic://topicA&type=listen&message=handlerA");
LOG.debug( "SENDING LISTEN FOR /topic/topicB" );
contentExchange = new AjaxTestContentExchange();
contentExchange.setMethod( "POST" );
contentExchange.setURL("http://localhost:" + port + "/amq");
contentExchange.setRequestContent( new ByteArrayBuffer("destination=topic://topicB&type=listen&message=handlerB") );
contentExchange.setRequestContentType( "application/x-www-form-urlencoded; charset=UTF-8" );
contentExchange.setRequestHeader( "Cookie", jsessionid );
httpClient.send(contentExchange);
contentExchange.waitForDone();
subscribe(httpClient, port, "destination=topic://topicB&type=listen&message=handlerB", null, sessionId);
// client 1 polls for messages
LOG.debug( "SENDING POLL" );
AjaxTestContentExchange poll = new AjaxTestContentExchange();
poll.setMethod( "GET" );
poll.setURL("http://localhost:" + port + "/amq?timeout=5000");
poll.setRequestHeader( "Cookie", jsessionid );
httpClient.send( poll );
final StringBuffer buf = new StringBuffer();
final CountDownLatch latch =
asyncRequest(httpClient, "http://localhost:" + port + "/amq?timeout=5000", buf, sessionId);
// while client 1 is polling, client 2 sends messages to the topics
LOG.debug( "SENDING MESSAGES" );
contentExchange = new AjaxTestContentExchange();
contentExchange.setMethod( "POST" );
contentExchange.setURL("http://localhost:" + port + "/amq");
contentExchange.setRequestContent( new ByteArrayBuffer(
"destination=topic://topicA&type=send&message=A1&"+
"d1=topic://topicB&t1=send&m1=B1&"+
"d2=topic://topicA&t2=send&m2=A2&"+
"d3=topic://topicB&t3=send&m3=B2"
) );
contentExchange.setRequestContentType( "application/x-www-form-urlencoded; charset=UTF-8" );
httpClient.send(contentExchange);
contentExchange.waitForDone();
sendMessages(httpClient, port, ("destination=topic://topicA&type=send&message=A1&"+
"d1=topic://topicB&t1=send&m1=B1&"+
"d2=topic://topicA&t2=send&m2=A2&"+
"d3=topic://topicB&t3=send&m3=B2").getBytes());
LOG.debug( "DONE POSTING MESSAGES" );
// wait for poll to finish
poll.waitForDone();
String response = poll.getResponseContent();
latch.await();
String response = buf.toString();
// not all messages might be delivered during the 1st poll. We need to check again.
poll = new AjaxTestContentExchange();
poll.setMethod( "GET" );
poll.setURL("http://localhost:" + port + "/amq?timeout=5000");
poll.setRequestHeader( "Cookie", jsessionid );
httpClient.send( poll );
poll.waitForDone();
final StringBuffer buf2 = new StringBuffer();
final CountDownLatch latch2 =
asyncRequest(httpClient, "http://localhost:" + port + "/amq?timeout=5000", buf2, sessionId);
latch2.await();
String fullResponse = response + poll.getResponseContent();
String fullResponse = response + buf2.toString();
LOG.debug( "full response " + fullResponse );
assertContains( "<response id='handlerA' destination='topic://topicA' >A1</response>", fullResponse );
assertContains( "<response id='handlerB' destination='topic://topicB' >B1</response>", fullResponse );
@ -568,4 +416,87 @@ public class AjaxTest extends JettyTestSupport {
httpClient.stop();
}
protected void sendMessages(HttpClient httpClient, int port, byte[] content) throws InterruptedException {
final CountDownLatch latch = new CountDownLatch(1);
final StringBuffer buf = new StringBuffer();
httpClient
.newRequest("http://localhost:" + port + "/amq")
.header("Content-Type", "application/x-www-form-urlencoded; charset=UTF-8")
.content(
new InputStreamContentProvider(new ByteArrayInputStream(content)))
.method(HttpMethod.POST).send(new BufferingResponseListener() {
@Override
public void onComplete(Result result) {
buf.append(getContentAsString());
latch.countDown();
}
});
latch.await();
}
protected String subscribe(HttpClient httpClient, int port, String content) throws InterruptedException {
return this.subscribe(httpClient, port, content, null, null);
}
protected String subscribe(HttpClient httpClient, int port, String content, String selector, String session) throws InterruptedException {
final CountDownLatch latch = new CountDownLatch(1);
final StringBuffer buf = new StringBuffer();
final StringBuffer sessionId = new StringBuffer();
Request request = httpClient.newRequest("http://localhost:" + port + "/amq");
if (selector != null) {
request.header("selector", selector);
}
if (session != null) {
request.header(HttpHeader.COOKIE, session);
}
request.header("Content-Type","application/x-www-form-urlencoded; charset=UTF-8")
.content(new InputStreamContentProvider(new ByteArrayInputStream(content.getBytes())))
.method(HttpMethod.POST).send(new BufferingResponseListener() {
@Override
public void onComplete(Result result) {
buf.append(getContentAsString());
String cookie = result.getResponse().getHeaders().get(HttpHeader.SET_COOKIE);
if (cookie != null) {
String[] cookieParts = cookie.split(";");
sessionId.append(cookieParts[0]);
}
latch.countDown();
}
});
latch.await();
return sessionId.toString();
}
protected CountDownLatch asyncRequest(final HttpClient httpClient, final String url, final StringBuffer buffer,
final String sessionId) {
final CountDownLatch latch = new CountDownLatch(1);
Request request = httpClient.newRequest(url);
if (sessionId != null) {
request.header(HttpHeader.COOKIE, sessionId);
}
request.send(new BufferingResponseListener() {
@Override
public void onComplete(Result result) {
buffer.append(getContentAsString());
latch.countDown();
}
});
return latch;
}
protected CountDownLatch asyncRequest(final HttpClient httpClient, final String url, final StringBuffer buffer,
final AtomicInteger status) {
final CountDownLatch latch = new CountDownLatch(1);
httpClient.newRequest(url).send(new BufferingResponseListener() {
@Override
public void onComplete(Result result) {
status.getAndSet(result.getResponse().getStatus());
buffer.append(getContentAsString());
latch.countDown();
}
});
return latch;
}
}

7
activemq-web-demo/src/test/java/org/apache/activemq/web/JettyServer.java
View File

@ -21,7 +21,7 @@ import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.demo.DefaultQueueSender;
import org.eclipse.jetty.server.Connector;
import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.server.nio.SelectChannelConnector;
import org.eclipse.jetty.server.ServerConnector;
import org.eclipse.jetty.webapp.WebAppContext;
/**
@ -46,7 +46,7 @@ public final class JettyServer {
BrokerService broker = new BrokerService();
broker.setPersistent(false);
broker.setUseJmx(true);
broker.addConnector("tcp://localhost:0");
broker.addConnector("tcp://localhost:61616");
broker.addConnector("stomp://localhost:0");
broker.start();
@ -63,9 +63,8 @@ public final class JettyServer {
}
System.out.println("Starting Web Server on port: " + port);
Server server = new Server();
SelectChannelConnector connector = new SelectChannelConnector();
ServerConnector connector = new ServerConnector(server);
connector.setPort(port);
connector.setServer(server);
WebAppContext context = new WebAppContext();
context.setResourceBase(WEBAPP_DIR);

10
activemq-web-demo/src/test/java/org/apache/activemq/web/JettyTestSupport.java
View File

@ -16,6 +16,8 @@
*/
package org.apache.activemq.web;
import static org.junit.Assert.assertTrue;
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
@ -33,15 +35,13 @@ import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.util.Wait;
import org.eclipse.jetty.server.Connector;
import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.server.nio.SelectChannelConnector;
import org.eclipse.jetty.server.ServerConnector;
import org.eclipse.jetty.webapp.WebAppContext;
import org.junit.After;
import org.junit.Before;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static org.junit.Assert.assertTrue;
public class JettyTestSupport {
private static final Logger LOG = LoggerFactory.getLogger(JettyTestSupport.class);
@ -74,9 +74,8 @@ public class JettyTestSupport {
int port = getPort();
server = new Server();
SelectChannelConnector connector = new SelectChannelConnector();
ServerConnector connector = new ServerConnector(server);
connector.setPort(port);
connector.setServer(server);
WebAppContext context = new WebAppContext();
context.setResourceBase("src/main/webapp");
@ -129,6 +128,7 @@ public class JettyTestSupport {
final URL url = new URL(bindLocation);
assertTrue("Jetty endpoint is available", Wait.waitFor(new Wait.Condition() {
@Override
public boolean isSatisified() throws Exception {
boolean canConnect = false;
try {

75
activemq-web-demo/src/test/java/org/apache/activemq/web/RestPersistentTest.java
View File

@ -16,16 +16,24 @@
*/
package org.apache.activemq.web;
import java.io.ByteArrayInputStream;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import java.io.ByteArrayInputStream;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
import org.eclipse.jetty.client.ContentExchange;
import org.eclipse.jetty.client.HttpClient;
import org.eclipse.jetty.client.api.Request;
import org.eclipse.jetty.client.api.Result;
import org.eclipse.jetty.client.util.BufferingResponseListener;
import org.eclipse.jetty.client.util.InputStreamContentProvider;
import org.eclipse.jetty.http.HttpMethod;
import org.eclipse.jetty.http.HttpStatus;
import org.junit.Ignore;
import org.junit.Test;
import static org.junit.Assert.*;
public class RestPersistentTest extends JettyTestSupport {
@Override
@ -58,7 +66,6 @@ public class RestPersistentTest extends JettyTestSupport {
HttpClient httpClient = new HttpClient();
httpClient.start();
httpClient.setConnectorType(HttpClient.CONNECTOR_SELECT_CHANNEL);
//post first message
// TODO: a problem with GET before POST
@ -69,36 +76,56 @@ public class RestPersistentTest extends JettyTestSupport {
private void postMessage(HttpClient httpClient, String url, String properties, String message) throws Exception
{
ContentExchange contentExchange = new ContentExchange();
contentExchange.setMethod("POST");
contentExchange.setURL(url+"&"+properties);
//contentExchange.setRequestHeader("accept", "text/xml");
contentExchange.setRequestHeader("Content-Type","text/xml");
contentExchange.setRequestContentSource(new ByteArrayInputStream(message.getBytes("UTF-8")));
final CountDownLatch latch = new CountDownLatch(1);
final StringBuffer buf = new StringBuffer();
final AtomicInteger status = new AtomicInteger();
httpClient.newRequest(url+"&"+properties)
.header("Content-Type","text/xml")
.content(new InputStreamContentProvider(new ByteArrayInputStream(message.getBytes("UTF-8"))))
.method(HttpMethod.POST).send(new BufferingResponseListener() {
@Override
public void onComplete(Result result) {
status.getAndSet(result.getResponse().getStatus());
buf.append(getContentAsString());
latch.countDown();
}
});
httpClient.send(contentExchange);
contentExchange.waitForDone();
assertTrue("success status", HttpStatus.isSuccess(contentExchange.getResponseStatus()));
latch.await();
assertTrue("success status", HttpStatus.isSuccess(status.get()));
}
private void getMessage(HttpClient httpClient, String url, String selector, String expectedMessage) throws Exception
{
ContentExchange contentExchange = new ContentExchange(true);
contentExchange.setURL(url);
contentExchange.setRequestHeader("accept", "text/xml");
contentExchange.setRequestHeader("Content-Type","text/xml");
final CountDownLatch latch = new CountDownLatch(1);
final StringBuffer buf = new StringBuffer();
final AtomicInteger status = new AtomicInteger();
Request request = httpClient.newRequest(url)
.header("accept", "text/xml")
.header("Content-Type","text/xml");
if(selector!=null)
{
contentExchange.setRequestHeader("selector", selector);
request.header("selector", selector);
}
httpClient.send(contentExchange);
contentExchange.waitForDone();
assertTrue("success status", HttpStatus.isSuccess(contentExchange.getResponseStatus()));
request.method(HttpMethod.GET).send(new BufferingResponseListener() {
@Override
public void onComplete(Result result) {
status.getAndSet(result.getResponse().getStatus());
buf.append(getContentAsString());
latch.countDown();
}
});
latch.await();
assertTrue("success status", HttpStatus.isSuccess(status.get()));
if(expectedMessage!=null)
{
assertNotNull(contentExchange.getResponseContent());
assertEquals(expectedMessage, contentExchange.getResponseContent().trim());
assertNotNull(buf.toString());
assertEquals(expectedMessage, buf.toString().trim());
}
}
}

236
activemq-web-demo/src/test/java/org/apache/activemq/web/RestTest.java
View File

@ -16,22 +16,28 @@
*/
package org.apache.activemq.web;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
import javax.jms.TextMessage;
import javax.management.ObjectName;
import org.apache.commons.lang.RandomStringUtils;
import org.eclipse.jetty.client.ContentExchange;
import org.eclipse.jetty.client.HttpClient;
import org.eclipse.jetty.client.api.Result;
import org.eclipse.jetty.client.util.BufferingResponseListener;
import org.eclipse.jetty.http.HttpFields;
import org.eclipse.jetty.http.HttpMethod;
import org.eclipse.jetty.http.HttpStatus;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static org.junit.Assert.*;
public class RestTest extends JettyTestSupport {
private static final Logger LOG = LoggerFactory.getLogger(RestTest.class);
@ -44,12 +50,13 @@ public class RestTest extends JettyTestSupport {
HttpClient httpClient = new HttpClient();
httpClient.start();
ContentExchange contentExchange = new ContentExchange();
httpClient.setConnectorType(HttpClient.CONNECTOR_SELECT_CHANNEL);
contentExchange.setURL("http://localhost:" + port + "/message/test?readTimeout=1000&type=queue");
httpClient.send(contentExchange);
contentExchange.waitForDone();
assertEquals("test", contentExchange.getResponseContent());
final StringBuffer buf = new StringBuffer();
final CountDownLatch latch =
asyncRequest(httpClient, "http://localhost:" + port + "/message/test?readTimeout=1000&type=queue", buf);
latch.await();
assertEquals("test", buf.toString());
}
@Test(timeout = 60 * 1000)
@ -58,18 +65,17 @@ public class RestTest extends JettyTestSupport {
HttpClient httpClient = new HttpClient();
httpClient.start();
ContentExchange contentExchange = new ContentExchange();
httpClient.setConnectorType(HttpClient.CONNECTOR_SELECT_CHANNEL);
contentExchange.setURL("http://localhost:" + port + "/message/test?readTimeout=5000&type=queue");
httpClient.send(contentExchange);
Thread.sleep(1000);
final StringBuffer buf = new StringBuffer();
final CountDownLatch latch =
asyncRequest(httpClient, "http://localhost:" + port + "/message/test?readTimeout=5000&type=queue", buf);
producer.send(session.createTextMessage("test"));
LOG.info("message sent");
contentExchange.waitForDone();
assertEquals("test", contentExchange.getResponseContent());
latch.await();
assertEquals("test", buf.toString());
}
@Test(timeout = 60 * 1000)
@ -88,22 +94,28 @@ public class RestTest extends JettyTestSupport {
HttpClient httpClient = new HttpClient();
httpClient.start();
ContentExchange contentExchange = new ContentExchange();
httpClient.setConnectorType(HttpClient.CONNECTOR_SELECT_CHANNEL);
contentExchange.setURL("http://localhost:" + port + "/message/test?readTimeout=1000&type=queue");
contentExchange.setRequestHeader("selector", "test=2");
httpClient.send(contentExchange);
contentExchange.waitForDone();
assertEquals("test2", contentExchange.getResponseContent());
final StringBuffer buf = new StringBuffer();
final CountDownLatch latch = new CountDownLatch(1);
httpClient.newRequest("http://localhost:" + port + "/message/test?readTimeout=1000&type=queue")
.header("selector", "test=2").send(new BufferingResponseListener() {
@Override
public void onComplete(Result result) {
buf.append(getContentAsString());
latch.countDown();
}
});
latch.await();
assertEquals("test2", buf.toString());
}
// test for https://issues.apache.org/activemq/browse/AMQ-2827
@Test(timeout = 15 * 1000)
public void testCorrelation() throws Exception {
int port = getPort();
HttpClient httpClient = new HttpClient();
httpClient.setConnectorType(HttpClient.CONNECTOR_SELECT_CHANNEL);
httpClient.start();
for (int i = 0; i < 200; i++) {
@ -116,13 +128,14 @@ public class RestTest extends JettyTestSupport {
LOG.info("Sending: " + correlId);
producer.send(message);
ContentExchange contentExchange = new ContentExchange();
contentExchange.setURL("http://localhost:" + port + "/message/test?readTimeout=1000&type=queue&clientId=test");
httpClient.send(contentExchange);
contentExchange.waitForDone();
LOG.info("Received: [" + contentExchange.getResponseStatus() + "] " + contentExchange.getResponseContent());
assertEquals(200, contentExchange.getResponseStatus());
assertEquals(correlId, contentExchange.getResponseContent());
final StringBuffer buf = new StringBuffer();
final CountDownLatch latch =
asyncRequest(httpClient, "http://localhost:" + port + "/message/test?readTimeout=1000&type=queue&clientId=test", buf);
latch.await();
LOG.info("Received: " + buf.toString());
// assertEquals(200, contentExchange.getResponseStatus());
assertEquals(correlId, buf.toString());
}
httpClient.stop();
}
@ -134,19 +147,26 @@ public class RestTest extends JettyTestSupport {
producer.send(session.createTextMessage("test"));
HttpClient httpClient = new HttpClient();
httpClient.start();
ContentExchange contentExchange = new ContentExchange();
httpClient.setConnectorType(HttpClient.CONNECTOR_SELECT_CHANNEL);
contentExchange.setURL("http://localhost:" + port + "/message/test?readTimeout=1000&type=queue&clientId=test");
httpClient.send(contentExchange);
contentExchange.waitForDone();
LOG.info("Received: [" + contentExchange.getResponseStatus() + "] " + contentExchange.getResponseContent());
contentExchange = new ContentExchange();
contentExchange.setMethod("POST");
contentExchange.setURL("http://localhost:" + port + "/message/test?clientId=test&action=unsubscribe");
httpClient.send(contentExchange);
contentExchange.waitForDone();
final StringBuffer buf = new StringBuffer();
final CountDownLatch latch =
asyncRequest(httpClient, "http://localhost:" + port + "/message/test?readTimeout=1000&type=queue&clientId=test", buf);
latch.await();
LOG.info("Received: " + buf.toString());
final StringBuffer buf2 = new StringBuffer();
final CountDownLatch latch2 = new CountDownLatch(1);
httpClient.newRequest("http://localhost:" + port + "/message/test?clientId=test&action=unsubscribe")
.method(HttpMethod.POST).send(new BufferingResponseListener() {
@Override
public void onComplete(Result result) {
buf2.append(getContentAsString());
latch2.countDown();
}
});
latch2.await();
httpClient.stop();
ObjectName query = new ObjectName("org.apache.activemq:BrokerName=localhost,Type=Subscription,destinationType=Queue,destinationName=test,*");
@ -160,20 +180,31 @@ public class RestTest extends JettyTestSupport {
HttpClient httpClient = new HttpClient();
httpClient.start();
ContentExchange contentExchange = new ContentExchange();
httpClient.setConnectorType(HttpClient.CONNECTOR_SELECT_CHANNEL);
contentExchange.setMethod("POST");
contentExchange.setURL("http://localhost:" + port + "/message/testPost?type=queue");
httpClient.send(contentExchange);
contentExchange.waitForDone();
assertTrue("success status", HttpStatus.isSuccess(contentExchange.getResponseStatus()));
final CountDownLatch latch = new CountDownLatch(1);
final StringBuffer buf = new StringBuffer();
final AtomicInteger status = new AtomicInteger();
httpClient.newRequest("http://localhost:" + port + "/message/testPost?type=queue")
.method(HttpMethod.POST).send(new BufferingResponseListener() {
@Override
public void onComplete(Result result) {
status.getAndSet(result.getResponse().getStatus());
buf.append(getContentAsString());
latch.countDown();
}
});
ContentExchange contentExchange2 = new ContentExchange();
contentExchange2.setURL("http://localhost:" + port + "/message/testPost?readTimeout=1000&type=Queue");
httpClient.send(contentExchange2);
contentExchange2.waitForDone();
assertTrue("success status", HttpStatus.isSuccess(contentExchange2.getResponseStatus()));
latch.await();
assertTrue("success status", HttpStatus.isSuccess(status.get()));
final StringBuffer buf2 = new StringBuffer();
final AtomicInteger status2 = new AtomicInteger();
final CountDownLatch latch2 =
asyncRequest(httpClient, "http://localhost:" + port + "/message/testPost?readTimeout=1000&type=Queue", buf2, status2);
latch2.await();
assertTrue("success status", HttpStatus.isSuccess(status2.get()));
}
// test for https://issues.apache.org/activemq/browse/AMQ-3857
@ -183,22 +214,42 @@ public class RestTest extends JettyTestSupport {
HttpClient httpClient = new HttpClient();
httpClient.start();
ContentExchange contentExchange = new ContentExchange();
httpClient.setConnectorType(HttpClient.CONNECTOR_SELECT_CHANNEL);
contentExchange.setMethod("POST");
contentExchange.setURL("http://localhost:" + port + "/message/testPost?type=queue&property=value");
httpClient.send(contentExchange);
contentExchange.waitForDone();
assertTrue("success status", HttpStatus.isSuccess(contentExchange.getResponseStatus()));
final CountDownLatch latch = new CountDownLatch(1);
final StringBuffer buf = new StringBuffer();
final AtomicInteger status = new AtomicInteger();
httpClient.newRequest("http://localhost:" + port + "/message/testPost?type=queue&property=value")
.method(HttpMethod.POST).send(new BufferingResponseListener() {
@Override
public void onComplete(Result result) {
status.getAndSet(result.getResponse().getStatus());
buf.append(getContentAsString());
latch.countDown();
}
});
ContentExchange contentExchange2 = new ContentExchange(true);
contentExchange2.setURL("http://localhost:" + port + "/message/testPost?readTimeout=1000&type=Queue");
httpClient.send(contentExchange2);
contentExchange2.waitForDone();
assertTrue("success status", HttpStatus.isSuccess(contentExchange2.getResponseStatus()));
latch.await();
assertTrue("success status", HttpStatus.isSuccess(status.get()));
HttpFields fields = contentExchange2.getResponseFields();
final CountDownLatch latch2 = new CountDownLatch(1);
final StringBuffer buf2 = new StringBuffer();
final AtomicInteger status2 = new AtomicInteger();
final HttpFields responseFields = new HttpFields();
httpClient.newRequest("http://localhost:" + port + "/message/testPost?readTimeout=1000&type=Queue")
.method(HttpMethod.GET).send(new BufferingResponseListener() {
@Override
public void onComplete(Result result) {
responseFields.add(result.getResponse().getHeaders());
status2.getAndSet(result.getResponse().getStatus());
buf2.append(getContentAsString());
latch2.countDown();
}
});
latch2.await();
assertTrue("success status", HttpStatus.isSuccess(status2.get()));
HttpFields fields = responseFields;
assertNotNull("Headers Exist", fields);
assertEquals("header value", "value", fields.getStringField("property"));
}
@ -210,14 +261,49 @@ public class RestTest extends JettyTestSupport {
HttpClient httpClient = new HttpClient();
httpClient.start();
ContentExchange contentExchange = new ContentExchange();
httpClient.setConnectorType(HttpClient.CONNECTOR_SELECT_CHANNEL);
contentExchange.setMethod("POST");
contentExchange.setURL("http://localhost:" + port + "/message/testPost?type=queue");
contentExchange.setRequestHeader("Authorization", "Basic YWRtaW46YWRtaW4=");
httpClient.send(contentExchange);
contentExchange.waitForDone();
assertTrue("success status", HttpStatus.isSuccess(contentExchange.getResponseStatus()));
final CountDownLatch latch = new CountDownLatch(1);
final StringBuffer buf = new StringBuffer();
final AtomicInteger status = new AtomicInteger();
httpClient.newRequest("http://localhost:" + port + "/message/testPost?type=queue")
.header("Authorization", "Basic YWRtaW46YWRtaW4=")
.method(HttpMethod.POST).send(new BufferingResponseListener() {
@Override
public void onComplete(Result result) {
status.getAndSet(result.getResponse().getStatus());
buf.append(getContentAsString());
latch.countDown();
}
});
latch.await();
assertTrue("success status", HttpStatus.isSuccess(status.get()));
}
protected CountDownLatch asyncRequest(final HttpClient httpClient, final String url, final StringBuffer buffer) {
final CountDownLatch latch = new CountDownLatch(1);
httpClient.newRequest(url).send(new BufferingResponseListener() {
@Override
public void onComplete(Result result) {
buffer.append(getContentAsString());
latch.countDown();
}
});
return latch;
}
protected CountDownLatch asyncRequest(final HttpClient httpClient, final String url, final StringBuffer buffer,
final AtomicInteger status) {
final CountDownLatch latch = new CountDownLatch(1);
httpClient.newRequest(url).send(new BufferingResponseListener() {
@Override
public void onComplete(Result result) {
status.getAndSet(result.getResponse().getStatus());
buffer.append(getContentAsString());
latch.countDown();
}
});
return latch;
}
}

17
activemq-web/pom.xml
View File

@ -68,20 +68,19 @@
</dependency>
<!-- web container -->
<dependency>
<groupId>org.apache.geronimo.specs</groupId>
<artifactId>geronimo-servlet_3.0_spec</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.tomcat</groupId>
<artifactId>tomcat-servlet-api</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.eclipse.jetty.aggregate</groupId>
<artifactId>jetty-all</artifactId>
<scope>test</scope>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-websocket</artifactId>
<version>${jetty8-version}</version>
<groupId>org.eclipse.jetty.websocket</groupId>
<artifactId>websocket-server</artifactId>
</dependency>
<dependency>
<groupId>org.eclipse.jetty</groupId>

50
activemq-web/src/main/java/org/apache/activemq/web/config/JspConfigurer.java Normal file
View File

@ -0,0 +1,50 @@
/**
* 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.web.config;
import org.eclipse.jetty.server.Handler;
import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.server.handler.HandlerCollection;
import org.eclipse.jetty.webapp.Configuration;
import org.eclipse.jetty.webapp.WebAppContext;
/**
*
*
*/
public class JspConfigurer {
public static void configureJetty(Server server, HandlerCollection collection) {
Configuration.ClassList classlist = Configuration.ClassList
.setServerDefault( server );
classlist.addBefore(
"org.eclipse.jetty.webapp.JettyWebXmlConfiguration",
"org.eclipse.jetty.annotations.AnnotationConfiguration" );
// Set the ContainerIncludeJarPattern so that jetty examines these
// container-path jars for tlds, web-fragments etc.
// If you omit the jar that contains the jstl .tlds, the jsp engine will
// scan for them instead.
for (Handler handler: collection.getHandlers()) {
if (handler instanceof WebAppContext){
((WebAppContext) handler).setAttribute(
"org.eclipse.jetty.server.webapp.ContainerIncludeJarPattern",
".*/[^/]*servlet-api-[^/]*\\.jar$|.*/javax.servlet.jsp.jstl-.*\\.jar$|.*/[^/]*taglibs.*\\.jar$" );
}
}
}
}

41
assembly/pom.xml
View File

@ -256,10 +256,6 @@
<groupId>org.apache.geronimo.specs</groupId>
<artifactId>geronimo-jms_1.1_spec</artifactId>
</dependency>
<dependency>
<groupId>org.apache.geronimo.specs</groupId>
<artifactId>geronimo-jsp_2.1_spec</artifactId>
</dependency>
<dependency>
<groupId>org.apache.geronimo.specs</groupId>
<artifactId>geronimo-j2ee-connector_1.5_spec</artifactId>
@ -272,10 +268,14 @@
<groupId>org.apache.geronimo.specs</groupId>
<artifactId>geronimo-j2ee-management_1.1_spec</artifactId>
</dependency>
<dependency>
<groupId>org.apache.geronimo.specs</groupId>
<artifactId>geronimo-servlet_3.0_spec</artifactId>
</dependency>
<dependency>
<groupId>org.apache.tomcat</groupId>
<artifactId>tomcat-servlet-api</artifactId>
</dependency>
<dependency>
<groupId>org.apache.tomcat</groupId>
<artifactId>tomcat-websocket-api</artifactId>
</dependency>
<dependency>
<groupId>org.apache.geronimo.specs</groupId>
<artifactId>geronimo-jta_1.0.1B_spec</artifactId>
@ -357,17 +357,22 @@
<artifactId>jcl-over-slf4j</artifactId>
</dependency>
<dependency>
<groupId>javax.servlet</groupId>
<artifactId>jstl</artifactId>
</dependency>
<groupId>org.apache.geronimo.specs</groupId>
<artifactId>geronimo-annotation_1.0_spec</artifactId>
</dependency>
<dependency>
<groupId>taglibs</groupId>
<artifactId>standard</artifactId>
</dependency>
<dependency>
<groupId>org.mortbay.jetty</groupId>
<artifactId>jsp-2.1-glassfish</artifactId>
</dependency>
<groupId>org.ow2.asm</groupId>
<artifactId>asm</artifactId>
</dependency>
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>apache-jsp</artifactId>
</dependency>
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>apache-jstl</artifactId>
</dependency>
<dependency>
<groupId>org.jasypt</groupId>
<artifactId>jasypt</artifactId>

22
assembly/src/main/descriptors/common-bin.xml
View File

@ -247,12 +247,24 @@
<include>org.springframework:spring-web</include>
<include>org.springframework:spring-webmvc</include>
<include>org.eclipse.jetty.aggregate:jetty-all</include>
<include>org.apache.geronimo.specs:geronimo-servlet_3.0_spec</include>
<include>org.apache.tomcat:tomcat-servlet-api</include>
<include>org.apache.tomcat:tomcat-websocket-api</include>
<!-- JSP support -->
<include>org.mortbay.jetty:jsp-2.1-glassfish</include>
<include>org.mortbay.jetty:jsp-api-2.1-glassfish</include>
<include>org.eclipse.jdt:core</include>
<!-- Jetty JSP api-->
<include>org.eclipse.jetty:apache-jsp</include>
<!-- Jetty JSP impl-->
<include>org.mortbay.jasper:apache-jsp</include>
<!-- jstl and el api -->
<include>org.eclipse.jetty:apache-jstl</include>
<include>org.apache.taglibs:taglibs-standard-spec</include>
<!-- jstl and el impl -->
<include>org.mortbay.jasper:apache-el</include>
<include>org.apache.taglibs:taglibs-standard-impl</include>
<include>org.apache.geronimo.specs:geronimo-annotation_1.0_spec</include>
<include>org.ow2.asm:asm</include>
<include>org.eclipse.jetty.orbit:org.eclipse.jdt.core</include>
<!-- Atom/RSS support -->
<include>rome:rome</include>

142
assembly/src/release/conf/jetty.xml
View File

@ -45,6 +45,40 @@
<property name="constraint" ref="adminSecurityConstraint" />
<property name="pathSpec" value="*.action" />
</bean>
<bean id="secHandlerCollection" class="org.eclipse.jetty.server.handler.HandlerCollection">
<property name="handlers">
<list>
<bean class="org.eclipse.jetty.webapp.WebAppContext">
<property name="contextPath" value="/admin" />
<property name="resourceBase" value="${activemq.home}/webapps/admin" />
<property name="logUrlOnStart" value="true" />
</bean>
<!-- Enable embedded file server for Blob messages -->
<!-- <bean class="org.eclipse.jetty.webapp.WebAppContext"> <property name="contextPath"
value="/fileserver" /> <property name="resourceBase" value="${activemq.home}/webapps/fileserver"
/> <property name="logUrlOnStart" value="true" /> <property name="parentLoaderPriority"
value="true" /> </bean> -->
<bean class="org.eclipse.jetty.webapp.WebAppContext">
<property name="contextPath" value="/api" />
<property name="resourceBase" value="${activemq.home}/webapps/api" />
<property name="logUrlOnStart" value="true" />
</bean>
<bean class="org.eclipse.jetty.server.handler.ResourceHandler">
<property name="directoriesListed" value="false" />
<property name="welcomeFiles">
<list>
<value>index.html</value>
</list>
</property>
<property name="resourceBase" value="${activemq.home}/webapps/" />
</bean>
<bean id="defaultHandler" class="org.eclipse.jetty.server.handler.DefaultHandler">
<property name="serveIcon" value="false" />
</bean>
</list>
</property>
</bean>
<bean id="securityHandler" class="org.eclipse.jetty.security.ConstraintSecurityHandler">
<property name="loginService" ref="securityLoginService" />
<property name="authenticator">
@ -56,45 +90,7 @@
<ref bean="securityConstraintMapping" />
</list>
</property>
<property name="handler">
<bean id="sec" class="org.eclipse.jetty.server.handler.HandlerCollection">
<property name="handlers">
<list>
<bean class="org.eclipse.jetty.webapp.WebAppContext">
<property name="contextPath" value="/admin" />
<property name="resourceBase" value="${activemq.home}/webapps/admin" />
<property name="logUrlOnStart" value="true" />
</bean>
<!-- Enable embedded file server for Blob messages -->
<!--
<bean class="org.eclipse.jetty.webapp.WebAppContext">
<property name="contextPath" value="/fileserver" />
<property name="resourceBase" value="${activemq.home}/webapps/fileserver" />
<property name="logUrlOnStart" value="true" />
<property name="parentLoaderPriority" value="true" />
</bean>
-->
<bean class="org.eclipse.jetty.webapp.WebAppContext">
<property name="contextPath" value="/api" />
<property name="resourceBase" value="${activemq.home}/webapps/api" />
<property name="logUrlOnStart" value="true" />
</bean>
<bean class="org.eclipse.jetty.server.handler.ResourceHandler">
<property name="directoriesListed" value="false" />
<property name="welcomeFiles">
<list>
<value>index.html</value>
</list>
</property>
<property name="resourceBase" value="${activemq.home}/webapps/" />
</bean>
<bean id="defaultHandler" class="org.eclipse.jetty.server.handler.DefaultHandler">
<property name="serveIcon" value="false" />
</bean>
</list>
</property>
</bean>
</property>
<property name="handler" ref="secHandlerCollection" />
</bean>
<bean id="contexts" class="org.eclipse.jetty.server.handler.ContextHandlerCollection">
@ -106,29 +102,9 @@
<property name="port" value="8161"/>
</bean>
<bean id="Server" depends-on="jettyPort" class="org.eclipse.jetty.server.Server" init-method="start"
<bean id="Server" depends-on="jettyPort" class="org.eclipse.jetty.server.Server"
destroy-method="stop">
<property name="connectors">
<list>
<bean id="Connector" class="org.eclipse.jetty.server.nio.SelectChannelConnector">
<!-- see the jettyPort bean -->
<property name="host" value="#{systemProperties['jetty.host']}" />
<property name="port" value="#{systemProperties['jetty.port']}" />
</bean>
<!--
Enable this connector if you wish to use https with web console
-->
<!--
<bean id="SecureConnector" class="org.eclipse.jetty.server.ssl.SslSelectChannelConnector">
<property name="port" value="8162" />
<property name="keystore" value="file:${activemq.conf}/broker.ks" />
<property name="password" value="password" />
</bean>
-->
</list>
</property>
<property name="handler">
<bean id="handlers" class="org.eclipse.jetty.server.handler.HandlerCollection">
<property name="handlers">
@ -142,4 +118,50 @@
</bean>
<bean id="invokeConnectors" class="org.springframework.beans.factory.config.MethodInvokingFactoryBean">
<property name="targetObject" ref="Server" />
<property name="targetMethod" value="setConnectors" />
<property name="arguments">
<list>
<bean id="Connector" class="org.eclipse.jetty.server.ServerConnector">
<constructor-arg ref="Server" />
<!-- see the jettyPort bean -->
<property name="host" value="#{systemProperties['jetty.host']}" />
<property name="port" value="#{systemProperties['jetty.port']}" />
</bean>
<!--
Enable this connector if you wish to use https with web console
-->
<!-- bean id="SecureConnector" class="org.eclipse.jetty.server.ServerConnector">
<constructor-arg ref="Server" />
<constructor-arg>
<bean id="handlers" class="org.eclipse.jetty.util.ssl.SslContextFactory">
<property name="keyStorePath" value="${activemq.conf}/broker.ks" />
<property name="keyStorePassword" value="password" />
</bean>
</constructor-arg>
<property name="port" value="8162" />
</bean -->
</list>
</property>
</bean>
<bean id="configureJetty" class="org.springframework.beans.factory.config.MethodInvokingFactoryBean">
<property name="staticMethod" value="org.apache.activemq.web.config.JspConfigurer.configureJetty" />
<property name="arguments">
<list>
<ref bean="Server" />
<ref bean="secHandlerCollection" />
</list>
</property>
</bean>
<bean id="invokeStart" class="org.springframework.beans.factory.config.MethodInvokingFactoryBean"
depends-on="configureJetty, invokeConnectors">
<property name="targetObject" ref="Server" />
<property name="targetMethod" value="start" />
</bean>
</beans>

153
assembly/src/release/examples/conf/jetty-demo.xml
View File

@ -45,6 +45,47 @@
<property name="constraint" ref="adminSecurityConstraint" />
<property name="pathSpec" value="*.action" />
</bean>
<bean id="secHandlerCollection" class="org.eclipse.jetty.server.handler.HandlerCollection">
<property name="handlers">
<list>
<bean class="org.eclipse.jetty.webapp.WebAppContext">
<property name="contextPath" value="/admin" />
<property name="resourceBase" value="${activemq.home}/webapps/admin" />
<property name="logUrlOnStart" value="true" />
</bean>
<bean class="org.eclipse.jetty.webapp.WebAppContext">
<property name="contextPath" value="/fileserver" />
<property name="resourceBase" value="${activemq.home}/webapps/fileserver" />
<property name="logUrlOnStart" value="true" />
<property name="parentLoaderPriority" value="true" />
</bean>
<bean class="org.eclipse.jetty.webapp.WebAppContext">
<property name="contextPath" value="/demo" />
<property name="resourceBase" value="${activemq.home}/webapps-demo/demo" />
<property name="logUrlOnStart" value="true" />
</bean>
<bean class="org.eclipse.jetty.webapp.WebAppContext">
<property name="contextPath" value="/api" />
<property name="resourceBase" value="${activemq.home}/webapps/api" />
<property name="logUrlOnStart" value="true" />
</bean>
<bean class="org.eclipse.jetty.server.handler.ResourceHandler">
<property name="directoriesListed" value="false" />
<property name="welcomeFiles">
<list>
<value>index.html</value>
</list>
</property>
<property name="resourceBase" value="${activemq.home}/webapps/" />
</bean>
<bean id="defaultHandler" class="org.eclipse.jetty.server.handler.DefaultHandler">
<property name="serveIcon" value="false" />
</bean>
</list>
</property>
</bean>
<bean id="securityHandler" class="org.eclipse.jetty.security.ConstraintSecurityHandler">
<property name="loginService" ref="securityLoginService" />
<property name="authenticator">
@ -56,49 +97,10 @@
<ref bean="securityConstraintMapping" />
</list>
</property>
<property name="handler">
<bean id="sec" class="org.eclipse.jetty.server.handler.HandlerCollection">
<property name="handlers">
<list>
<bean class="org.eclipse.jetty.webapp.WebAppContext">
<property name="contextPath" value="/admin" />
<property name="resourceBase" value="${activemq.home}/webapps/admin" />
<property name="logUrlOnStart" value="true" />
</bean>
<bean class="org.eclipse.jetty.webapp.WebAppContext">
<property name="contextPath" value="/fileserver" />
<property name="resourceBase" value="${activemq.home}/webapps/fileserver" />
<property name="logUrlOnStart" value="true" />
<property name="parentLoaderPriority" value="true" />
</bean>
<bean class="org.eclipse.jetty.webapp.WebAppContext">
<property name="contextPath" value="/demo" />
<property name="resourceBase" value="${activemq.home}/webapps-demo/demo" />
<property name="logUrlOnStart" value="true" />
</bean>
<bean class="org.eclipse.jetty.webapp.WebAppContext">
<property name="contextPath" value="/api" />
<property name="resourceBase" value="${activemq.home}/webapps/api" />
<property name="logUrlOnStart" value="true" />
</bean>
<bean class="org.eclipse.jetty.server.handler.ResourceHandler">
<property name="directoriesListed" value="false" />
<property name="welcomeFiles">
<list>
<value>index.html</value>
</list>
</property>
<property name="resourceBase" value="${activemq.home}/webapps/" />
</bean>
<bean id="defaultHandler" class="org.eclipse.jetty.server.handler.DefaultHandler">
<property name="serveIcon" value="false" />
</bean>
</list>
</property>
</bean>
</property>
<property name="handler" ref="secHandlerCollection" />
</bean>
<bean id="contexts" class="org.eclipse.jetty.server.handler.ContextHandlerCollection">
</bean>
@ -106,29 +108,10 @@
<!-- the default port number for the web console -->
<property name="port" value="8161"/>
</bean>
<bean id="Server" depends-on="jettyPort" class="org.eclipse.jetty.server.Server" init-method="start"
<bean id="Server" depends-on="jettyPort" class="org.eclipse.jetty.server.Server"
destroy-method="stop">
<property name="connectors">
<list>
<bean id="Connector" class="org.eclipse.jetty.server.nio.SelectChannelConnector">
<!-- see the jettyPort bean -->
<property name="port" value="#{systemProperties['jetty.port']}" />
</bean>
<!--
Enable this connector if you wish to use https with web console
-->
<!--
<bean id="SecureConnector" class="org.eclipse.jetty.server.ssl.SslSelectChannelConnector">
<property name="port" value="8162" />
<property name="keystore" value="file:${activemq.conf}/broker.ks" />
<property name="password" value="password" />
</bean>
-->
</list>
</property>
<property name="handler">
<bean id="handlers" class="org.eclipse.jetty.server.handler.HandlerCollection">
<property name="handlers">
@ -142,4 +125,50 @@
</bean>
<bean id="invokeConnectors" class="org.springframework.beans.factory.config.MethodInvokingFactoryBean">
<property name="targetObject" ref="Server" />
<property name="targetMethod" value="setConnectors" />
<property name="arguments">
<list>
<bean id="Connector" class="org.eclipse.jetty.server.ServerConnector">
<constructor-arg ref="Server" />
<!-- see the jettyPort bean -->
<property name="port" value="#{systemProperties['jetty.port']}" />
</bean>
<!--
Enable this connector if you wish to use https with web console
-->
<!-- bean id="SecureConnector" class="org.eclipse.jetty.server.ServerConnector">
<constructor-arg ref="Server" />
<constructor-arg>
<bean id="handlers" class="org.eclipse.jetty.util.ssl.SslContextFactory">
<property name="keyStorePath" value="${activemq.conf}/broker.ks" />
<property name="keyStorePassword" value="password" />
</bean>
</constructor-arg>
<property name="port" value="8162" />
</bean -->
</list>
</property>
</bean>
<bean id="configureJetty" class="org.springframework.beans.factory.config.MethodInvokingFactoryBean">
<property name="staticMethod" value="org.apache.activemq.web.config.JspConfigurer.configureJetty" />
<property name="arguments">
<list>
<ref bean="Server" />
<ref bean="secHandlerCollection" />
</list>
</property>
</bean>
<bean id="invokeStart" class="org.springframework.beans.factory.config.MethodInvokingFactoryBean"
depends-on="configureJetty, invokeConnectors">
<property name="targetObject" ref="Server" />
<property name="targetMethod" value="start" />
</bean>
</beans>

14
assembly/src/release/webapps/api/WEB-INF/web.xml
View File

@ -15,11 +15,10 @@
See the License for the specific language governing permissions and
limitations under the License.
-->
<!DOCTYPE web-app
PUBLIC "-//Sun Microsystems, Inc.//DTD Web Application 2.3//EN"
"http://java.sun.com/dtd/web-app_2_3.dtd">
<web-app>
<web-app xmlns="http://java.sun.com/xml/ns/javaee"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_3_0.xsd"
version="3.0">
<display-name>Apache ActiveMQ REST API</display-name>
@ -27,6 +26,7 @@
<servlet-name>MessageServlet</servlet-name>
<servlet-class>org.apache.activemq.web.MessageServlet</servlet-class>
<load-on-startup>1</load-on-startup>
<async-supported>true</async-supported>
<!--
Uncomment this parameter if you plan to use multiple consumers over REST
<init-param>
@ -38,8 +38,7 @@
<servlet>
<servlet-name>jolokia-agent</servlet-name>
<servlet-class>org.jolokia.http.AgentServlet</servlet-class>
<load-on-startup>1</load-on-startup>
<servlet-class>org.jolokia.http.AgentServlet</servlet-class>
<init-param>
<param-name>discoveryEnabled</param-name>
<param-value>false</param-value>
@ -52,6 +51,7 @@
<param-name>agentDescription</param-name>
<param-value>Apache ActiveMQ</param-value>
</init-param>
<load-on-startup>1</load-on-startup>
</servlet>
<servlet-mapping>

74
pom.xml
View File

@ -75,12 +75,10 @@
<jasypt-version>1.9.2</jasypt-version>
<jaxb-bundle-version>2.2.11_1</jaxb-bundle-version>
<jdom-version>1.0</jdom-version>
<jetty9-version>9.2.6.v20141205</jetty9-version>
<jetty8-version>8.1.16.v20140903</jetty8-version>
<jetty-version>${jetty8-version}</jetty-version>
<jetty9-version>9.2.13.v20150730</jetty9-version>
<jetty-version>${jetty9-version}</jetty-version>
<jmdns-version>3.4.1</jmdns-version>
<jsp-version>2.1.v20100127</jsp-version>
<jstl-version>1.1.2</jstl-version>
<tomcat-api-version>8.0.24</tomcat-api-version>
<jettison-version>1.3.7</jettison-version>
<jmock-version>2.5.1</jmock-version>
<jolokia-version>1.3.1</jolokia-version>
@ -539,17 +537,27 @@
<artifactId>geronimo-j2ee-connector_1.5_spec</artifactId>
<version>2.0.0</version>
</dependency>
<dependency>
<groupId>org.ow2.asm</groupId>
<artifactId>asm</artifactId>
<version>5.0.3</version>
</dependency>
<!-- Servlet 3.1 and JSP -->
<dependency>
<groupId>org.apache.geronimo.specs</groupId>
<artifactId>geronimo-jsp_2.1_spec</artifactId>
<version>1.0.1</version>
<groupId>org.apache.tomcat</groupId>
<artifactId>tomcat-jsp-api</artifactId>
<version>${tomcat-api-version}</version>
</dependency>
<dependency>
<groupId>org.apache.tomcat</groupId>
<artifactId>tomcat-servlet-api</artifactId>
<version>${tomcat-api-version}</version>
</dependency>
<dependency>
<groupId>org.apache.geronimo.specs</groupId>
<artifactId>geronimo-servlet_3.0_spec</artifactId>
<version>1.0</version>
<groupId>org.apache.tomcat</groupId>
<artifactId>tomcat-websocket-api</artifactId>
<version>${tomcat-api-version}</version>
</dependency>
<dependency>
@ -923,19 +931,39 @@
<artifactId>xpp3</artifactId>
<version>${xpp3-version}</version>
</dependency>
<dependency>
<groupId>org.mortbay.jetty</groupId>
<artifactId>jsp-2.1-glassfish</artifactId>
<version>${jsp-version}</version>
<groupId>org.eclipse.jetty</groupId>
<artifactId>apache-jsp</artifactId>
<version>${jetty-version}</version>
</dependency>
<dependency>
<groupId>org.eclipse.jetty.aggregate</groupId>
<artifactId>jetty-all</artifactId>
<groupId>org.eclipse.jetty</groupId>
<artifactId>apache-jstl</artifactId>
<version>${jetty-version}</version>
</dependency>
<dependency>
<groupId>org.eclipse.jetty.aggregate</groupId>
<artifactId>jetty-all</artifactId>
<version>${jetty-version}</version>
<exclusions>
<exclusion>
<groupId>javax.servlet</groupId>
<artifactId>javax.servlet-api</artifactId>
</exclusion>
<exclusion>
<groupId>javax.websocket</groupId>
<artifactId>javax.websocket-api</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.eclipse.jetty.websocket</groupId>
<artifactId>websocket-server</artifactId>
<version>${jetty-version}</version>
</dependency>
<dependency>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpclient</artifactId>
@ -996,7 +1024,7 @@
<artifactId>jettison</artifactId>
<version>${jettison-version}</version>
</dependency>
<dependency>
<groupId>annogen</groupId>
<artifactId>annogen</artifactId>
@ -1058,7 +1086,7 @@
<artifactId>jcl-over-slf4j</artifactId>
<version>${slf4j-version}</version>
</dependency>
<dependency>
<!-- >dependency>
<groupId>javax.servlet</groupId>
<artifactId>jstl</artifactId>
<version>${jstl-version}</version>
@ -1067,7 +1095,7 @@
<groupId>taglibs</groupId>
<artifactId>standard</artifactId>
<version>1.1.2</version>
</dependency>
</dependency -->
<dependency>
<groupId>org.apache.geronimo.components</groupId>
@ -1191,7 +1219,7 @@
<version>${apache-rat-plugin-version}</version>
</plugin>
<plugin>
<groupId>org.mortbay.jetty</groupId>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-maven-plugin</artifactId>
<version>${jetty-version}</version>
</plugin>