This closes #177
This commit is contained in:
Clebert Suconic
commit
6496be50c7
|
|
@ -20,7 +20,6 @@ import java.util.List;
|
|||
import java.util.Set;
|
||||
|
||||
import org.apache.activemq.artemis.api.core.BaseInterceptor;
|
||||
import org.apache.activemq.artemis.api.core.Interceptor;
|
||||
import org.apache.activemq.artemis.core.protocol.core.CoreRemotingConnection;
|
||||
import org.apache.activemq.artemis.core.security.ActiveMQPrincipal;
|
||||
import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection;
|
||||
|
|
@ -43,17 +42,17 @@ public interface RemotingService {
|
|||
|
||||
ReusableLatch getConnectionCountLatch();
|
||||
|
||||
void addIncomingInterceptor(Interceptor interceptor);
|
||||
void addIncomingInterceptor(BaseInterceptor interceptor);
|
||||
|
||||
List<BaseInterceptor> getIncomingInterceptors();
|
||||
|
||||
void addOutgoingInterceptor(Interceptor interceptor);
|
||||
void addOutgoingInterceptor(BaseInterceptor interceptor);
|
||||
|
||||
List<BaseInterceptor> getOutgoinInterceptors();
|
||||
|
||||
boolean removeIncomingInterceptor(Interceptor interceptor);
|
||||
boolean removeIncomingInterceptor(BaseInterceptor interceptor);
|
||||
|
||||
boolean removeOutgoingInterceptor(Interceptor interceptor);
|
||||
boolean removeOutgoingInterceptor(BaseInterceptor interceptor);
|
||||
|
||||
void stop(boolean criticalError) throws Exception;
|
||||
|
||||
|
|
|
|||
|
|
@ -40,7 +40,6 @@ import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
|
|||
import org.apache.activemq.artemis.api.core.ActiveMQException;
|
||||
import org.apache.activemq.artemis.api.core.ActiveMQInterruptedException;
|
||||
import org.apache.activemq.artemis.api.core.BaseInterceptor;
|
||||
import org.apache.activemq.artemis.api.core.Interceptor;
|
||||
import org.apache.activemq.artemis.api.core.TransportConfiguration;
|
||||
import org.apache.activemq.artemis.core.config.Configuration;
|
||||
import org.apache.activemq.artemis.core.protocol.core.CoreRemotingConnection;
|
||||
|
|
@ -525,7 +524,7 @@ public class RemotingServiceImpl implements RemotingService, ConnectionLifeCycle
|
|||
}
|
||||
|
||||
@Override
|
||||
public void addIncomingInterceptor(final Interceptor interceptor) {
|
||||
public void addIncomingInterceptor(final BaseInterceptor interceptor) {
|
||||
incomingInterceptors.add(interceptor);
|
||||
|
||||
updateProtocols();
|
||||
|
|
@ -537,7 +536,7 @@ public class RemotingServiceImpl implements RemotingService, ConnectionLifeCycle
|
|||
}
|
||||
|
||||
@Override
|
||||
public boolean removeIncomingInterceptor(final Interceptor interceptor) {
|
||||
public boolean removeIncomingInterceptor(final BaseInterceptor interceptor) {
|
||||
if (incomingInterceptors.remove(interceptor)) {
|
||||
updateProtocols();
|
||||
return true;
|
||||
|
|
@ -548,7 +547,7 @@ public class RemotingServiceImpl implements RemotingService, ConnectionLifeCycle
|
|||
}
|
||||
|
||||
@Override
|
||||
public void addOutgoingInterceptor(final Interceptor interceptor) {
|
||||
public void addOutgoingInterceptor(final BaseInterceptor interceptor) {
|
||||
outgoingInterceptors.add(interceptor);
|
||||
updateProtocols();
|
||||
}
|
||||
|
|
@ -559,7 +558,7 @@ public class RemotingServiceImpl implements RemotingService, ConnectionLifeCycle
|
|||
}
|
||||
|
||||
@Override
|
||||
public boolean removeOutgoingInterceptor(final Interceptor interceptor) {
|
||||
public boolean removeOutgoingInterceptor(final BaseInterceptor interceptor) {
|
||||
if (outgoingInterceptors.remove(interceptor)) {
|
||||
updateProtocols();
|
||||
return true;
|
||||
|
|
|
|||
|
|
@ -17,7 +17,6 @@
|
|||
package org.apache.activemq.artemis.core.server;
|
||||
|
||||
import org.apache.activemq.artemis.api.core.BaseInterceptor;
|
||||
import org.apache.activemq.artemis.api.core.Interceptor;
|
||||
import org.apache.activemq.artemis.api.core.Pair;
|
||||
import org.apache.activemq.artemis.core.config.ConnectorServiceConfiguration;
|
||||
import org.apache.activemq.artemis.core.server.cluster.Transformer;
|
||||
|
|
@ -55,7 +54,7 @@ public interface ServiceRegistry {
|
|||
*/
|
||||
Collection<Pair<ConnectorServiceFactory, ConnectorServiceConfiguration>> getConnectorServices(List<ConnectorServiceConfiguration> configs);
|
||||
|
||||
void addIncomingInterceptor(Interceptor interceptor);
|
||||
void addIncomingInterceptor(BaseInterceptor interceptor);
|
||||
|
||||
/**
|
||||
* Get a list of org.apache.activemq.artemis.api.core.BaseInterceptor instances
|
||||
|
|
@ -65,7 +64,7 @@ public interface ServiceRegistry {
|
|||
*/
|
||||
List<BaseInterceptor> getIncomingInterceptors(List<String> classNames);
|
||||
|
||||
void addOutgoingInterceptor(Interceptor interceptor);
|
||||
void addOutgoingInterceptor(BaseInterceptor interceptor);
|
||||
|
||||
/**
|
||||
* Get a list of org.apache.activemq.artemis.api.core.BaseInterceptor instances
|
||||
|
|
|
|||
|
|
@ -17,7 +17,6 @@
|
|||
package org.apache.activemq.artemis.core.server.impl;
|
||||
|
||||
import org.apache.activemq.artemis.api.core.BaseInterceptor;
|
||||
import org.apache.activemq.artemis.api.core.Interceptor;
|
||||
import org.apache.activemq.artemis.api.core.Pair;
|
||||
import org.apache.activemq.artemis.core.config.ConnectorServiceConfiguration;
|
||||
import org.apache.activemq.artemis.core.server.ActiveMQMessageBundle;
|
||||
|
|
@ -118,7 +117,7 @@ public class ServiceRegistryImpl implements ServiceRegistry {
|
|||
}
|
||||
|
||||
@Override
|
||||
public void addIncomingInterceptor(Interceptor interceptor) {
|
||||
public void addIncomingInterceptor(BaseInterceptor interceptor) {
|
||||
incomingInterceptors.add(interceptor);
|
||||
}
|
||||
|
||||
|
|
@ -132,7 +131,7 @@ public class ServiceRegistryImpl implements ServiceRegistry {
|
|||
}
|
||||
|
||||
@Override
|
||||
public void addOutgoingInterceptor(Interceptor interceptor) {
|
||||
public void addOutgoingInterceptor(BaseInterceptor interceptor) {
|
||||
outgoingInterceptors.add(interceptor);
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,92 @@
|
|||
<?xml version='1.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.
|
||||
-->
|
||||
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<parent>
|
||||
<groupId>org.apache.activemq.examples.stomp</groupId>
|
||||
<artifactId>stomp-examples</artifactId>
|
||||
<version>1.1.1-SNAPSHOT</version>
|
||||
</parent>
|
||||
|
||||
<artifactId>stomp-embedded-interceptor</artifactId>
|
||||
<packaging>jar</packaging>
|
||||
<name>ActiveMQ Artemis JMS Stomp Embedded Interceptor Example</name>
|
||||
|
||||
<properties>
|
||||
<activemq.basedir>${project.basedir}/../../../..</activemq.basedir>
|
||||
</properties>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.apache.activemq</groupId>
|
||||
<artifactId>artemis-server</artifactId>
|
||||
<version>${project.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.activemq</groupId>
|
||||
<artifactId>artemis-jms-server</artifactId>
|
||||
<version>${project.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.activemq</groupId>
|
||||
<artifactId>artemis-stomp-protocol</artifactId>
|
||||
<version>${project.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.activemq</groupId>
|
||||
<artifactId>artemis-hornetq-protocol</artifactId>
|
||||
<version>${project.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>io.netty</groupId>
|
||||
<artifactId>netty-all</artifactId>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
<build>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.apache.activemq</groupId>
|
||||
<artifactId>artemis-maven-plugin</artifactId>
|
||||
<executions>
|
||||
<execution>
|
||||
<id>runClient</id>
|
||||
<goals>
|
||||
<goal>runClient</goal>
|
||||
</goals>
|
||||
<configuration>
|
||||
<clientClass>org.apache.activemq.artemis.jms.example.StompEmbeddedWithInterceptorExample</clientClass>
|
||||
</configuration>
|
||||
</execution>
|
||||
</executions>
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.apache.activemq.examples.stomp</groupId>
|
||||
<artifactId>stomp-embedded-interceptor</artifactId>
|
||||
<version>${project.version}</version>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</build>
|
||||
|
||||
</project>
|
||||
|
|
@ -0,0 +1,37 @@
|
|||
<!--
|
||||
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.
|
||||
-->
|
||||
|
||||
<html>
|
||||
<head>
|
||||
<title>ActiveMQ Artemis Stomp 1.1 Example</title>
|
||||
<link rel="stylesheet" type="text/css" href="../../../common/common.css" />
|
||||
<link rel="stylesheet" type="text/css" href="../../../common/prettify.css" />
|
||||
<script type="text/javascript" src="../../../common/prettify.js"></script>
|
||||
</head>
|
||||
<body onload="prettyPrint()">
|
||||
<h1>Stomp 1.2 Example</h1>
|
||||
|
||||
<pre>To run the example, simply type <b>mvn verify -Pexample</b> from this directory.</pre>
|
||||
|
||||
<p>This example shows you how to configure ActiveMQ Artemis to intercept received Stomp messages.</p>
|
||||
<p>The example will start a ActiveMQ Artemis server configured with Stomp and JMS.</p>
|
||||
<p>The client will open a socket to initiate a Stomp 1.2 connection and then send one Stomp message (using TCP directly).
|
||||
The interceptor will print each message received.</p>
|
||||
</body>
|
||||
</html>
|
||||
|
|
@ -0,0 +1,126 @@
|
|||
/*
|
||||
* 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.artemis.jms.example;
|
||||
|
||||
import org.apache.activemq.artemis.api.core.ActiveMQException;
|
||||
import org.apache.activemq.artemis.api.core.TransportConfiguration;
|
||||
import org.apache.activemq.artemis.core.config.Configuration;
|
||||
import org.apache.activemq.artemis.core.config.impl.ConfigurationImpl;
|
||||
import org.apache.activemq.artemis.core.config.impl.SecurityConfiguration;
|
||||
import org.apache.activemq.artemis.core.protocol.stomp.StompConnection;
|
||||
import org.apache.activemq.artemis.core.protocol.stomp.StompFrame;
|
||||
import org.apache.activemq.artemis.core.protocol.stomp.StompFrameInterceptor;
|
||||
import org.apache.activemq.artemis.core.remoting.impl.netty.NettyAcceptorFactory;
|
||||
import org.apache.activemq.artemis.jms.server.embedded.EmbeddedJMS;
|
||||
import org.apache.activemq.artemis.jms.server.config.ConnectionFactoryConfiguration;
|
||||
import org.apache.activemq.artemis.jms.server.config.impl.ConnectionFactoryConfigurationImpl;
|
||||
import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection;
|
||||
import org.apache.activemq.artemis.spi.core.security.ActiveMQSecurityManagerImpl;
|
||||
|
||||
import javax.jms.Connection;
|
||||
import javax.jms.ConnectionFactory;
|
||||
import javax.jms.MessageConsumer;
|
||||
import javax.jms.MessageProducer;
|
||||
import javax.jms.Queue;
|
||||
import javax.jms.Session;
|
||||
import javax.jms.TextMessage;
|
||||
import java.io.OutputStream;
|
||||
import java.net.Socket;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Date;
|
||||
import java.util.List;
|
||||
import java.util.HashSet;
|
||||
|
||||
/**
|
||||
* This example demonstrates how to run an ActiveMQ Artemis embedded with JMS
|
||||
*/
|
||||
public class StompEmbeddedWithInterceptorExample {
|
||||
|
||||
public static class MyStompInterceptor
|
||||
implements StompFrameInterceptor {
|
||||
|
||||
public boolean intercept(StompFrame frame, RemotingConnection remotingConnection)
|
||||
throws ActiveMQException {
|
||||
|
||||
StompConnection connection = (StompConnection) remotingConnection;
|
||||
|
||||
System.out.println("Intercepted frame " + frame + " on connection " + connection);
|
||||
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
private static final String END_OF_FRAME = "\u0000";
|
||||
|
||||
public static void main(final String[] args) throws Exception {
|
||||
EmbeddedJMS jmsServer = new EmbeddedJMS();
|
||||
String brokerConfigPath = StompEmbeddedWithInterceptorExample.class.getResource("/broker.xml").toString();
|
||||
jmsServer.setConfigResourcePath(brokerConfigPath);
|
||||
|
||||
jmsServer.start();
|
||||
jmsServer.getActiveMQServer().getRemotingService().addIncomingInterceptor(new MyStompInterceptor());
|
||||
|
||||
System.out.println("Started Embedded JMS Server");
|
||||
|
||||
// Step 1. Create a TCP socket to connect to the Stomp port
|
||||
Socket socket = new Socket("localhost", 61616);
|
||||
|
||||
// Step 2. Send a CONNECT frame to connect to the server
|
||||
String connectFrame = "CONNECT\n" +
|
||||
"accept-version:1.2\n" +
|
||||
"host:localhost\n" +
|
||||
"login:guest\n" +
|
||||
"passcode:guest\n" +
|
||||
"request-id:1\n" +
|
||||
"\n" +
|
||||
END_OF_FRAME;
|
||||
sendFrame(socket, connectFrame);
|
||||
|
||||
// Step 3. Send a SEND frame (a Stomp message) to the
|
||||
// jms.queue.exampleQueue address with a text body
|
||||
String text = "Hello World from Stomp 1.2 !";
|
||||
String message = "SEND\n" +
|
||||
"destination:jms.queue.exampleQueue\n" +
|
||||
"\n" +
|
||||
text +
|
||||
END_OF_FRAME;
|
||||
sendFrame(socket, message);
|
||||
System.out.println("Sent Stomp message: " + text);
|
||||
|
||||
// Step 4. Send a DISCONNECT frame to disconnect from the server
|
||||
String disconnectFrame = "DISCONNECT\n" +
|
||||
"\n" +
|
||||
END_OF_FRAME;
|
||||
sendFrame(socket, disconnectFrame);
|
||||
|
||||
// Step 5. Slose the TCP socket
|
||||
socket.close();
|
||||
|
||||
Thread.sleep(1000);
|
||||
jmsServer.stop();
|
||||
}
|
||||
|
||||
private static void sendFrame(Socket socket, String data) throws Exception {
|
||||
byte[] bytes = data.getBytes(StandardCharsets.UTF_8);
|
||||
OutputStream outputStream = socket.getOutputStream();
|
||||
for (int i = 0; i < bytes.length; i++) {
|
||||
outputStream.write(bytes[i]);
|
||||
}
|
||||
outputStream.flush();
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,48 @@
|
|||
<?xml version='1.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.
|
||||
-->
|
||||
|
||||
<configuration xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xmlns="urn:activemq"
|
||||
xsi:schemaLocation="urn:activemq /schema/artemis-server.xsd">
|
||||
|
||||
<core xmlns="urn:activemq:core">
|
||||
|
||||
<persistence-enabled>false</persistence-enabled>
|
||||
|
||||
<acceptors>
|
||||
<acceptor name="artemis">tcp://127.0.0.1:61616?protocols=STOMP</acceptor>
|
||||
</acceptors>
|
||||
|
||||
<!-- Other config -->
|
||||
|
||||
<security-settings>
|
||||
<!--security for example queue-->
|
||||
<security-setting match="#">
|
||||
<permission type="createDurableQueue" roles="guest"/>
|
||||
<permission type="deleteDurableQueue" roles="guest"/>
|
||||
<permission type="createNonDurableQueue" roles="guest"/>
|
||||
<permission type="deleteNonDurableQueue" roles="guest"/>
|
||||
<permission type="consume" roles="guest"/>
|
||||
<permission type="send" roles="guest"/>
|
||||
</security-setting>
|
||||
</security-settings>
|
||||
|
||||
</core>
|
||||
</configuration>
|
||||
Loading…
Reference in New Issue