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

ARTEMIS-1542 adding example to replicate clustered issue with AMQP and CPP clients

This commit is contained in:
Clebert Suconic 2017-12-11 10:18:45 -05:00
parent fcfa4a5e63
commit 9c4b715aed
9 changed files with 611 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/cluster/impl/ClusterConnectionBridge.java
View File

@ -182,7 +182,7 @@ public class ClusterConnectionBridge extends BridgeImpl {
messageCopy.putExtraBytesProperty(Message.HDR_ROUTE_TO_IDS, queueIds);
messageCopy = super.beforeForward(messageCopy, forwardingAddress);
messageCopy = super.beforeForward(messageCopy, null);
return messageCopy;
}

1
examples/protocols/amqp/pom.xml
View File

@ -48,6 +48,7 @@ under the License.
<id>release</id>
<modules>
<module>proton-cpp</module>
<module>proton-clustered-cpp</module>
<module>queue</module>
<module>proton-ruby</module>
</modules>

18
examples/protocols/amqp/proton-clustered-cpp/compile.sh Executable file
View File

@ -0,0 +1,18 @@
# 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.
# This requires g++ and qpid-cpp-client-devel
g++ src/main/cpp/hello.cpp -o hello -l qpidmessaging -l qpidtypes

167
examples/protocols/amqp/proton-clustered-cpp/pom.xml Normal file
View File

@ -0,0 +1,167 @@
<?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.amqp</groupId>
<artifactId>amqp</artifactId>
<version>2.5.0-SNAPSHOT</version>
</parent>
<artifactId>proton-clustered-cpp</artifactId>
<packaging>jar</packaging>
<name>ActiveMQ Artemis AMQP CPP Example</name>
<properties>
<activemq.basedir>${project.basedir}/../../../..</activemq.basedir>
</properties>
<dependencies>
<dependency>
<groupId>org.apache.activemq</groupId>
<artifactId>artemis-jms-client</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.geronimo.specs</groupId>
<artifactId>geronimo-jms_2.0_spec</artifactId>
</dependency>
<dependency>
<groupId>org.apache.qpid</groupId>
<artifactId>qpid-jms-client</artifactId>
<version>${qpid.jms.version}</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.activemq</groupId>
<artifactId>artemis-maven-plugin</artifactId>
<executions>
<execution>
<id>create0</id>
<goals>
<goal>create</goal>
</goals>
<configuration>
<ignore>${noServer}</ignore>
<instance>${basedir}/target/server0</instance>
<configuration>${basedir}/target/classes/activemq/server0</configuration>
<!-- this makes it easier in certain envs -->
<javaOptions>-Djava.net.preferIPv4Stack=true</javaOptions>
</configuration>
</execution>
<execution>
<id>create1</id>
<goals>
<goal>create</goal>
</goals>
<configuration>
<ignore>${noServer}</ignore>
<instance>${basedir}/target/server1</instance>
<configuration>${basedir}/target/classes/activemq/server1</configuration>
<!-- this makes it easier in certain envs -->
<javaOptions>-Djava.net.preferIPv4Stack=true</javaOptions>
</configuration>
</execution>
<execution>
<id>start0</id>
<goals>
<goal>cli</goal>
</goals>
<configuration>
<ignore>${noServer}</ignore>
<spawn>true</spawn>
<location>${basedir}/target/server0</location>
<testURI>tcp://localhost:61616</testURI>
<args>
<param>run</param>
</args>
<name>server0</name>
</configuration>
</execution>
<execution>
<id>start1</id>
<goals>
<goal>cli</goal>
</goals>
<configuration>
<spawn>true</spawn>
<ignore>${noServer}</ignore>
<location>${basedir}/target/server1</location>
<testURI>tcp://localhost:61617</testURI>
<args>
<param>run</param>
</args>
<name>server1</name>
</configuration>
</execution>
<execution>
<id>runClient</id>
<goals>
<goal>runClient</goal>
</goals>
<configuration>
<clientClass>org.apache.activemq.artemis.jms.example.ProtonCPPExample</clientClass>
</configuration>
</execution>
<execution>
<id>stop1</id>
<goals>
<goal>cli</goal>
</goals>
<configuration>
<ignore>${noServer}</ignore>
<location>${basedir}/target/server1</location>
<args>
<param>stop</param>
</args>
</configuration>
</execution>
<execution>
<id>stop2</id>
<goals>
<goal>cli</goal>
</goals>
<configuration>
<ignore>${noServer}</ignore>
<location>${basedir}/target/server2</location>
<args>
<param>stop</param>
</args>
</configuration>
</execution>
</executions>
<dependencies>
<dependency>
<groupId>org.apache.activemq.examples.amqp</groupId>
<artifactId>protoncpp</artifactId>
<version>${project.version}</version>
</dependency>
</dependencies>
</plugin>
</plugins>
</build>
</project>

61
examples/protocols/amqp/proton-clustered-cpp/readme.html Normal file
View File

@ -0,0 +1,61 @@
<!--
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements. See the NOTICE file
distributed with this work for additional information
regarding copyright ownership. The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
-->
<html>
<head>
<title>ActiveMQ Artemis QPID cpp 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>AMQP CPP example</h1>
<p>ActiveMQ Artemis is a multi protocol broker. It will inspect the initial handshake of clients to determine what protocol to use.</p>
<p>All you need to do is to connect a client into activemq's configured port and you should be able connect.</p>
<p>To run this example simply run the command <literal>mvn verify -Pexample</literal>, execute the compile.sh script and start the executable called ./hello</p>
<pre>
# first make sure you have the dependencies you need to compile and run the client
# You will have to adapt this step according to your platform. Consult the <a href="http://qpid.apache.org/releases/qpid-0.30/programming/book/">qpid docs</a> for more information.
# There is a list of <a href="http://qpid.apache.org/packages.html">packages</a> you can install as well.
[proton-cpp]$ sudo yum install qpid-cpp-client-devel
# on a first window
[proton-cpp]$ mvn verify -Pexample
# on a second window
# That goes without saying but you will of course need g++ (the C++ compiler) installed
[proton-cpp]$ ./compile.sh
[proton-cpp]$ ./hello
</pre>
<p>You don't need to do anything special to configure the ActiveMQ Artemis server to accept AMQP clients. </p>
<p>Just for the sake of documentation though we are setting the port of ActiveMQ Artemis on this example as 5672 which is the port qpid have by default. </p>
<p>This is totally optional and you don't need to follow this convention. You can use any port you chose including ActiveMQ's 61616 default port</p>
<pre class="prettyprint">
<code>
&lt;acceptor name="proton-acceptor"&gt;tcp://localhost:5672&lt;/acceptor&gt;
</code>
</pre>
<h2>Example step-by-step</h2>
<p> We are using qpid cpp client on this example. There are several libraries you may chose from for AMQP. We have ellect one that we consider simple enough for users.</p>
</body>
</html>

106
examples/protocols/amqp/proton-clustered-cpp/src/main/cpp/hello.cpp Normal file
View File

@ -0,0 +1,106 @@
/*
*
* 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.
*
*/
/*
Jiira Issue: https://issues.apache.org/jira/browse/ARTEMIS-1542
Modified example (source was hello.cpp).
This example produces an exception in ActiveMQ Artemis 2.4.0 in a HA setup.
Prerequisites: Two brokers are running in a HA cluster.
This example sends a message to broker1 and tries to read from broker 2.
compile it using: g++ src/main/cpp/send-test.cpp -o send-test -l qpidmessaging -l qpidtypes
*/
#include <qpid/messaging/Connection.h>
#include <qpid/messaging/Message.h>
#include <qpid/messaging/Receiver.h>
#include <qpid/messaging/Sender.h>
#include <qpid/messaging/Session.h>
#include <iostream>
using namespace qpid::messaging;
int main(int argc, char** argv) {
std::string broker = argc > 1 ? argv[1] : "localhost:61616";
std::string broker2 = argc > 1 ? argv[2] : "localhost:61617";
std::string address = argc > 2 ? argv[3] : "test.queue";
std::string exampleQueue = argc > 2 ? argv[3] : "exampleQueue";
// Connection options documented at http://qpid.apache.org/releases/qpid-0.30/programming/book/connections.html#connection-options
std::string connectionOptions = argc > 3 ? argv[3] : "{protocol:amqp1.0}";
try {
// Step 3. Create an amqp qpid 1.0 connection
Connection connection(broker, connectionOptions);
connection.open();
// Step 4. Create a session
Session session = connection.createSession();
// Step 5. Create a sender
Sender sender = session.createSender(address);
Connection connection2(broker2, connectionOptions);
connection2.open();
// Step 4. Create a session
Session session2 = connection2.createSession();
//create a receiver
Receiver receiver = session2.createReceiver(address);
Message message;
message.getContentObject() = "Hello world!";
message.getContentObject().setEncoding("utf8");
message.setContentType("text/plain");
for (int i = 0; i < 10; i++) {
sender.send(message);
}
// receive the simple message
message = receiver.fetch(Duration::SECOND * 1);
std::cout << "Received a message with this following content \"" << message.getContent() << "\"" << std::endl;
// acknowledge the message
session.acknowledge();
// Create a sender towards the example, so the java class will give up waiting
Sender senderExample = session.createSender(exampleQueue);
for (int i = 0; i < 10; i++) {
senderExample.send(message);
}
// close the connection
connection.close();
return 0;
} catch(const std::exception& error) {
std::cerr << error.what() << std::endl;
return 1;
}
}

101
examples/protocols/amqp/proton-clustered-cpp/src/main/java/org/apache/activemq/artemis/jms/example/ProtonCPPExample.java Normal file
View File

@ -0,0 +1,101 @@
/*
* 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 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 javax.naming.InitialContext;
import org.apache.qpid.jms.JmsConnectionFactory;
/**
* This example demonstrates the use of ActiveMQ Artemis "pre-acknowledge" functionality where
* messages are acknowledged before they are delivered to the consumer.
* <p>
* Please see the readme.html for more details.
*/
public class ProtonCPPExample {
public static void main(final String[] args) throws Exception {
Connection connection = null;
InitialContext initialContext = null;
try {
// Create an initial context to perform the JNDI lookup.
initialContext = new InitialContext();
// if you wanted to use Core JMS, use this line instead.
// ConnectionFactory cf = new ActiveMQConnectionFactory();
ConnectionFactory cf = new JmsConnectionFactory("amqp://localhost:61616");
// Create a the JMS objects
connection = cf.createConnection();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
// Perform the look-ups
Queue queue = session.createQueue("exampleQueue");
MessageConsumer messageConsumer = session.createConsumer(queue);
MessageProducer producerAnswer = session.createProducer(queue);
// Start the connection
connection.start();
System.out.println("On a shell script, execute the following:");
System.out.println("./compile.sh");
System.out.println("./hello");
for (int i = 0; i < 10; i++) {
try {
// Step 5. Finally, receive the message
TextMessage messageReceived = (TextMessage) messageConsumer.receive(5000);
if (messageReceived == null) {
System.out.println("No messages");
// We are not going to issue this as an error because
// we also use this example as part of our tests on artemis
// this is not considered an error, just that no messages arrived (i.e. hello wasn't called)
} else {
System.out.println("message received: " + messageReceived.getText());
// Sending message back to client
producerAnswer.send(session.createTextMessage("HELLO from Apache ActiveMQ Artemis " + i + "!!"));
}
} catch (Throwable e) {
e.printStackTrace();
}
}
} finally {
// Step 9. Be sure to close our resources!
if (initialContext != null) {
initialContext.close();
}
if (connection != null) {
connection.close();
}
}
}
}

78
examples/protocols/amqp/proton-clustered-cpp/src/main/resources/activemq/server0/broker.xml Normal file
View File

@ -0,0 +1,78 @@
<?xml version="1.0" encoding="UTF-8" standalone="no"?>
<!--
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="urn:activemq" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="urn:activemq /schema/artemis-server.xsd">
<core xmlns="urn:activemq:core">
<bindings-directory>./data/bindings</bindings-directory>
<journal-directory>./data/journal</journal-directory>
<large-messages-directory>./data/largemessages</large-messages-directory>
<paging-directory>./data/paging</paging-directory>
<!-- Connectors -->
<connectors>
<connector name="netty-connector">tcp://localhost:61616</connector>
</connectors>
<!-- Acceptors -->
<acceptors>
<acceptor name="netty-acceptor">tcp://localhost:61616</acceptor>
</acceptors>
<cluster-connections>
<cluster-connection-uri
address="static://(tcp://localhost:61617)?connectorName=netty-connector;retryInterval=500;messageLoadBalancingType=STRICT;maxHops=1"
name="my-cluster"/>
</cluster-connections>
<!-- Other config -->
<security-settings>
<!--security for example queue-->
<security-setting match="#">
<permission roles="guest" type="createDurableQueue"/>
<permission roles="guest" type="deleteDurableQueue"/>
<permission roles="guest" type="createNonDurableQueue"/>
<permission roles="guest" type="deleteNonDurableQueue"/>
<permission roles="guest" type="consume"/>
<permission roles="guest" type="send"/>
</security-setting>
</security-settings>
<addresses>
<address name="test.queue">
<anycast>
<queue name="test.queue"/>
</anycast>
</address>
<address name="exampleQueue">
<anycast>
<queue name="exampleQueue"/>
</anycast>
</address>
</addresses>
</core>
</configuration>

78
examples/protocols/amqp/proton-clustered-cpp/src/main/resources/activemq/server1/broker.xml Normal file
View File

@ -0,0 +1,78 @@
<?xml version="1.0" encoding="UTF-8" standalone="no"?>
<!--
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="urn:activemq" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="urn:activemq /schema/artemis-server.xsd">
<core xmlns="urn:activemq:core">
<bindings-directory>./data/bindings</bindings-directory>
<journal-directory>./data/journal</journal-directory>
<large-messages-directory>./data/largemessages</large-messages-directory>
<paging-directory>./data/paging</paging-directory>
<!-- Connectors -->
<connectors>
<connector name="netty-connector">tcp://localhost:61617</connector>
</connectors>
<!-- Acceptors -->
<acceptors>
<acceptor name="netty-acceptor">tcp://localhost:61617</acceptor>
</acceptors>
<!-- Clustering configuration -->
<cluster-connections>
<cluster-connection-uri
address="static://(tcp://localhost:61616)?connectorName=netty-connector;retryInterval=500;messageLoadBalancingType=STRICT;maxHops=1"
name="my-cluster"/>
</cluster-connections>
<!-- Other config -->
<security-settings>
<!--security for example queue-->
<security-setting match="#">
<permission roles="guest" type="createDurableQueue"/>
<permission roles="guest" type="deleteDurableQueue"/>
<permission roles="guest" type="createNonDurableQueue"/>
<permission roles="guest" type="deleteNonDurableQueue"/>
<permission roles="guest" type="consume"/>
<permission roles="guest" type="send"/>
</security-setting>
</security-settings>
<addresses>
<address name="test.queue">
<anycast>
<queue name="test.queue"/>
</anycast>
</address>
<address name="exampleQueue">
<anycast>
<queue name="exampleQueue"/>
</anycast>
</address>
</addresses>
</core>
</configuration>