Merge pull request #217 from jbonofre/KARAF-6706
[KARAF-6706] Add SNMP collector
diff --git a/assembly/src/main/feature/feature.xml b/assembly/src/main/feature/feature.xml
index 29fc430..85002d9 100644
--- a/assembly/src/main/feature/feature.xml
+++ b/assembly/src/main/feature/feature.xml
@@ -403,6 +403,11 @@
<feature>decanter-common</feature>
<feature>http</feature>
<bundle>mvn:commons-lang/commons-lang/2.6</bundle>
+ <bundle dependency="true">mvn:org.ow2.asm/asm/7.3.1</bundle>
+ <bundle dependency="true">mvn:org.ow2.asm/asm-util/7.3.1</bundle>
+ <bundle dependency="true">mvn:org.ow2.asm/asm-tree/7.3.1</bundle>
+ <bundle dependency="true">mvn:org.ow2.asm/asm-analysis/7.3.1</bundle>
+ <bundle dependency="true">mvn:org.ow2.asm/asm-commons/7.3.1</bundle>
<bundle>mvn:org.apache.karaf.decanter.appender/org.apache.karaf.decanter.appender.orientdb/${project.version}</bundle>
</feature>
@@ -467,6 +472,16 @@
<feature>decanter-processor-aggregate-core</feature>
</feature>
+ <feature name="decanter-processor-groupby-core" version="${project.version}" description="Karaf Decanter GroupBy Processor core">
+ <feature>decanter-common</feature>
+ <bundle>mvn:org.apache.karaf.decanter.processor/org.apache.karaf.decanter.processor.groupby/${project.version}</bundle>
+ </feature>
+
+ <feature name="decanter-processor-groupby" version="${project.version}" description="Karaf Decanter GroupBy Processor">
+ <configfile finalname="/etc/org.apache.karaf.decanter.processor.groupby.cfg">mvn:org.apache.karaf.decanter.processor/org.apache.karaf.decanter.processor.groupby/${project.version}/cfg</configfile>
+ <feature>decanter-processor-groupby-core</feature>
+ </feature>
+
<feature name="decanter-alerting-core" version="${project.version}" description="Karaf Decanter Alerting core">
<feature>decanter-common</feature>
<bundle>mvn:org.apache.karaf.decanter.alerting/org.apache.karaf.decanter.alerting.service/${project.version}</bundle>
diff --git a/collector/jetty/src/main/java/org/apache/karaf/decanter/collector/jetty/DecanterCollectorJettyHandler.java b/collector/jetty/src/main/java/org/apache/karaf/decanter/collector/jetty/DecanterCollectorJettyHandler.java
index ad9d1d7..4573bbf 100644
--- a/collector/jetty/src/main/java/org/apache/karaf/decanter/collector/jetty/DecanterCollectorJettyHandler.java
+++ b/collector/jetty/src/main/java/org/apache/karaf/decanter/collector/jetty/DecanterCollectorJettyHandler.java
@@ -30,9 +30,7 @@
import javax.servlet.ServletException;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
-import java.io.BufferedReader;
import java.io.IOException;
-import java.io.Reader;
import java.util.Dictionary;
import java.util.Enumeration;
import java.util.HashMap;
diff --git a/collector/mqtt/src/main/java/org/apache/karaf/decanter/collector/mqtt/MqttCollector.java b/collector/mqtt/src/main/java/org/apache/karaf/decanter/collector/mqtt/MqttCollector.java
index 1217e85..074318f 100644
--- a/collector/mqtt/src/main/java/org/apache/karaf/decanter/collector/mqtt/MqttCollector.java
+++ b/collector/mqtt/src/main/java/org/apache/karaf/decanter/collector/mqtt/MqttCollector.java
@@ -59,7 +59,6 @@
private String dispatcherTopic;
private boolean consuming = false;
-
@Activate
public void activate(ComponentContext componentContext) throws Exception {
properties = componentContext.getProperties();
@@ -78,8 +77,10 @@
if (password != null) {
options.setPassword(password.toCharArray());
}
- client.connect(options);
- client.subscribe(topic);
+ options.setConnectionTimeout(60);
+ options.setAutomaticReconnect(true);
+ options.setKeepAliveInterval(10);
+ options.setExecutorServiceTimeout(30);
client.setCallback(new MqttCallback() {
@Override
public void connectionLost(Throwable cause) {
@@ -113,12 +114,18 @@
// nothing to do
}
});
+ client.connect(options);
+ client.subscribe(topic);
}
@Deactivate
- public void deactivate() throws Exception {
+ public void deactivate() {
if (client != null) {
- client.disconnect();
+ try {
+ client.disconnect();
+ } catch (Exception e) {
+ // no-op
+ }
}
}
diff --git a/itest/NOTICE b/itest/NOTICE
new file mode 100644
index 0000000..4e4af9e
--- /dev/null
+++ b/itest/NOTICE
@@ -0,0 +1,57 @@
+Apache Karaf Decanter
+Copyright 2015-2019 The Apache Software Foundation
+
+I. Included Software
+
+This product includes software developed at
+The Apache Software Foundation (http://www.apache.org/).
+Licensed under the Apache License 2.0.
+
+This product includes software developed at
+Elastic (https://www.elastic.co/).
+Licensed under the Apache License 2.0.
+
+This product includes software developed at
+OrientDB (http://orientdb.com).
+Licensed under the Apache License 2.0.
+
+II. Used Software
+
+This product uses software developed at
+The OSGi Alliance (http://www.osgi.org/).
+Copyright (c) OSGi Alliance (2000, 2010).
+Licensed under the Apache License 2.0.
+
+This product uses software developed at
+OPS4J (http://www.ops4j.org/).
+Licensed under the Apache License 2.0.
+
+This product uses software developed at
+SLF4J (http://www.slf4j.org/).
+Licensed under the MIT License.
+
+This product uses software developed at
+JUnit (http://www.junit.org/).
+Licensed under the Eclipse Public License 1.0.
+
+This product uses software developed at
+Redis (http://www.redis.io).
+Licensed under the BSD license.
+
+This product uses software developed at
+Dropwizard (http://www.dropwizard.io).
+Licensed under the Apache License 2.0.
+
+This product uses software developed at
+searchbox.io (https://github.com/searchbox-io)
+Licensed under the Apache License 2.0.
+
+This product uses software developed at
+MongoDB (https://www.mongodb.com/)
+Licensed under the Apache License 2.0.
+
+III. License Summary
+- Apache License 2.0
+- MIT License
+- Eclipse Public License 1.0
+- BSD License
diff --git a/itest/pom.xml b/itest/pom.xml
index 39bb31b..df35e26 100644
--- a/itest/pom.xml
+++ b/itest/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.karaf</groupId>
<artifactId>decanter</artifactId>
- <version>2.3.0-SNAPSHOT</version>
+ <version>2.6.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
@@ -89,6 +89,13 @@
<scope>runtime</scope>
</dependency>
+ <!-- dropwizard -->
+ <dependency>
+ <groupId>io.dropwizard.metrics</groupId>
+ <artifactId>metrics-core</artifactId>
+ <version>4.1.16</version>
+ </dependency>
+
<!-- camel -->
<dependency>
<groupId>org.apache.camel</groupId>
diff --git a/itest/src/test/java/org/apache/karaf/decanter/itests/appender/CamelAppenderTest.java b/itest/src/test/java/org/apache/karaf/decanter/itests/appender/CamelAppenderTest.java
index 2c2cd1d..4c2ff09 100644
--- a/itest/src/test/java/org/apache/karaf/decanter/itests/appender/CamelAppenderTest.java
+++ b/itest/src/test/java/org/apache/karaf/decanter/itests/appender/CamelAppenderTest.java
@@ -56,7 +56,9 @@
@Test
public void test() throws Exception {
final List<Exchange> exchanges = new ArrayList<>();
+
// create route
+ System.out.println("Creating Camel consumer route ...");
RouteBuilder routeBuilder = new RouteBuilder() {
@Override
public void configure() throws Exception {
@@ -73,24 +75,42 @@
camelContext.addRoutes(routeBuilder);
camelContext.start();
- Thread.sleep(1000);
+ while (!camelContext.isStarted()) {
+ Thread.sleep(200);
+ }
// install decanter
+ System.out.println("Installing Decanter Appender Camel ...");
System.out.println(executeCommand("feature:repo-add decanter " + System.getProperty("decanter.version")));
System.out.println(executeCommand("feature:install decanter-appender-camel", new RolePrincipal("admin")));
- Thread.sleep(2000);
+ String configList = executeCommand("config:list '(service.pid=org.apache.karaf.decanter.appender.camel)'");
+ while (!configList.contains("service.pid")) {
+ Thread.sleep(500);
+ configList = executeCommand("config:list '(service.pid=org.apache.karaf.decanter.appender.camel)'");
+ }
// send event
+ System.out.println("Sending event ...");
EventAdmin eventAdmin = getOsgiService(EventAdmin.class);
HashMap<String, String> data = new HashMap<>();
data.put("foo", "bar");
Event event = new Event("decanter/collect/test", data);
eventAdmin.sendEvent(event);
+ System.out.println("Waiting event ...");
+ while (exchanges.size() < 1) {
+ Thread.sleep(200);
+ }
+
Assert.assertEquals(1, exchanges.size());
HashMap<String, Object> received = exchanges.get(0).getIn().getBody(HashMap.class);
+
+ for (String key : received.keySet()) {
+ System.out.println(key + " = " + received.get(key));
+ }
+
Assert.assertEquals("decanter/collect/test", received.get("event.topics"));
Assert.assertEquals("bar", received.get("foo"));
}
diff --git a/itest/src/test/java/org/apache/karaf/decanter/itests/appender/FileAppenderTest.java b/itest/src/test/java/org/apache/karaf/decanter/itests/appender/FileAppenderTest.java
index d32272b..dab1ef9 100644
--- a/itest/src/test/java/org/apache/karaf/decanter/itests/appender/FileAppenderTest.java
+++ b/itest/src/test/java/org/apache/karaf/decanter/itests/appender/FileAppenderTest.java
@@ -27,6 +27,10 @@
import org.ops4j.pax.exam.karaf.options.KarafDistributionOption;
import org.ops4j.pax.exam.spi.reactors.ExamReactorStrategy;
import org.ops4j.pax.exam.spi.reactors.PerClass;
+import org.osgi.framework.Bundle;
+import org.osgi.service.component.runtime.ServiceComponentRuntime;
+import org.osgi.service.component.runtime.dto.ComponentConfigurationDTO;
+import org.osgi.service.component.runtime.dto.ComponentDescriptionDTO;
import org.osgi.service.event.Event;
import org.osgi.service.event.EventAdmin;
@@ -48,12 +52,14 @@
return Stream.of(super.config(), options).flatMap(Stream::of).toArray(Option[]::new);
}
- @Test
+ @Test(timeout = 60000)
public void test() throws Exception {
// install decanter
System.out.println(executeCommand("feature:repo-add decanter " + System.getProperty("decanter.version")));
System.out.println(executeCommand("feature:install decanter-appender-file", new RolePrincipal("admin")));
+ Thread.sleep(2000);
+
// send event
EventAdmin eventAdmin = getOsgiService(EventAdmin.class);
HashMap<String, String> data = new HashMap<>();
@@ -61,6 +67,8 @@
Event event = new Event("decanter/collect/test", data);
eventAdmin.sendEvent(event);
+ Thread.sleep(2000);
+
// read file
File file = new File(System.getProperty("karaf.data"), "decanter");
StringBuilder builder = new StringBuilder();
@@ -68,7 +76,13 @@
builder.append(reader.readLine()).append("\n");
}
- Assert.assertTrue(builder.toString().contains("foo=bar"));
+ System.out.println(builder.toString());
+
+ if (builder.toString().contains("foo=bar")) {
+ Assert.assertTrue(builder.toString().contains("foo=bar"));
+ } else {
+ Assert.assertTrue(builder.toString().contains("\"foo\":\"bar\""));
+ }
}
}
diff --git a/itest/src/test/java/org/apache/karaf/decanter/itests/appender/JdbcAppenderTest.java b/itest/src/test/java/org/apache/karaf/decanter/itests/appender/JdbcAppenderTest.java
index 15864b9..7245790 100644
--- a/itest/src/test/java/org/apache/karaf/decanter/itests/appender/JdbcAppenderTest.java
+++ b/itest/src/test/java/org/apache/karaf/decanter/itests/appender/JdbcAppenderTest.java
@@ -55,18 +55,25 @@
return Stream.of(super.config(), options).flatMap(Stream::of).toArray(Option[]::new);
}
- @Test
+ @Test(timeout = 60000)
public void test() throws Exception {
// install database
System.out.println(executeCommand("feature:install jdbc", new RolePrincipal("admin")));
System.out.println(executeCommand("feature:install pax-jdbc-derby", new RolePrincipal("admin")));
- System.out.println(executeCommand("jdbc:ds-list"));
+ String dsList = executeCommand("jdbc:ds-list");
+ while (!dsList.contains("jdbc/decanter")) {
+ Thread.sleep(200);
+ dsList = executeCommand("jdbc:ds-list");
+ }
+ System.out.println(dsList);
// install decanter
System.out.println(executeCommand("feature:repo-add decanter " + System.getProperty("decanter.version")));
System.out.println(executeCommand("feature:install decanter-appender-jdbc", new RolePrincipal("admin")));
+ Thread.sleep(2000);
+
// send event
EventAdmin eventAdmin = getOsgiService(EventAdmin.class);
HashMap<String, String> data = new HashMap<>();
@@ -74,15 +81,23 @@
Event event = new Event("decanter/collect/test", data);
eventAdmin.sendEvent(event);
+ Thread.sleep(2000);
+
// check database content
DataSource dataSource = getOsgiService(DataSource.class);
try (Connection connection = dataSource.getConnection()) {
try (Statement statement = connection.createStatement()) {
try (ResultSet resultSet = statement.executeQuery("select * from decanter")) {
resultSet.next();
- String json = resultSet.getString(2);
- Assert.assertTrue(json.contains("\"foo\":\"bar\""));
- Assert.assertTrue(json.contains("\"event_topics\":\"decanter/collect/test\""));
+ String result = resultSet.getString(2);
+ System.out.println(result);
+ if (result.contains("foo=bar")) {
+ Assert.assertTrue(result.contains("foo=bar"));
+ Assert.assertTrue(result.contains("event.topics=decanter/collect/test"));
+ } else {
+ Assert.assertTrue(result.contains("\"foo\":\"bar\""));
+ Assert.assertTrue(result.contains("\"event_topics\":\"decanter/collect/test\""));
+ }
}
}
}
diff --git a/itest/src/test/java/org/apache/karaf/decanter/itests/appender/JmsAppenderTest.java b/itest/src/test/java/org/apache/karaf/decanter/itests/appender/JmsAppenderTest.java
index b5af80b..37ec430 100644
--- a/itest/src/test/java/org/apache/karaf/decanter/itests/appender/JmsAppenderTest.java
+++ b/itest/src/test/java/org/apache/karaf/decanter/itests/appender/JmsAppenderTest.java
@@ -64,7 +64,7 @@
return Stream.of(super.config(), options).flatMap(Stream::of).toArray(Option[]::new);
}
- @Test
+ @Test(timeout = 60000)
public void test() throws Exception {
// install jms
System.out.println(executeCommand("feature:install jms", new RolePrincipal("admin")));
@@ -82,6 +82,8 @@
System.out.println(executeCommand("feature:repo-add decanter " + System.getProperty("decanter.version")));
System.out.println(executeCommand("feature:install decanter-appender-jms", new RolePrincipal("admin")));
+ Thread.sleep(2000);
+
// send event
EventAdmin eventAdmin = getOsgiService(EventAdmin.class);
HashMap<String, String> data = new HashMap<>();
@@ -89,12 +91,18 @@
Event event = new Event("decanter/collect/test", data);
eventAdmin.sendEvent(event);
+ Thread.sleep(2000);
+
// browse
String browse = executeCommand("jms:browse jms/decanter decanter");
System.out.println(browse);
- Assert.assertTrue(browse.contains("\"foo\":\"bar\""));
+ if (browse.contains("foo=bar")) {
+ Assert.assertTrue(browse.contains("foo=bar"));
+ } else {
+ Assert.assertTrue(browse.contains("\"foo\":\"bar\""));
+ }
}
}
diff --git a/itest/src/test/java/org/apache/karaf/decanter/itests/appender/LogAppenderTest.java b/itest/src/test/java/org/apache/karaf/decanter/itests/appender/LogAppenderTest.java
index c3c25c3..7d919ed 100644
--- a/itest/src/test/java/org/apache/karaf/decanter/itests/appender/LogAppenderTest.java
+++ b/itest/src/test/java/org/apache/karaf/decanter/itests/appender/LogAppenderTest.java
@@ -45,12 +45,14 @@
return Stream.of(super.config(), options).flatMap(Stream::of).toArray(Option[]::new);
}
- @Test
+ @Test(timeout = 60000)
public void test() throws Exception {
// install decanter
System.out.println(executeCommand("feature:repo-add decanter " + System.getProperty("decanter.version")));
System.out.println(executeCommand("feature:install decanter-appender-log", new RolePrincipal("admin")));
+ Thread.sleep(2000);
+
// send event synchronously for test
EventAdmin eventAdmin = getOsgiService(EventAdmin.class);
HashMap<String, String> data = new HashMap<>();
@@ -58,11 +60,20 @@
Event event = new Event("decanter/collect/test", data);
eventAdmin.sendEvent(event);
+ Thread.sleep(2000);
+
// get log
String log = executeCommand("log:display");
- Assert.assertTrue(log.contains("foo=bar"));
- Assert.assertTrue(log.contains("decanter/collect/test"));
+ System.out.println(log);
+
+ if (log.contains("foo=bar")) {
+ Assert.assertTrue(log.contains("foo=bar"));
+ Assert.assertTrue(log.contains("decanter/collect/test"));
+ } else {
+ Assert.assertTrue(log.contains("\"foo\":\"bar\""));
+ Assert.assertTrue(log.contains("decanter/collect/test"));
+ }
}
}
diff --git a/itest/src/test/java/org/apache/karaf/decanter/itests/appender/PrometheusAppenderTest.java b/itest/src/test/java/org/apache/karaf/decanter/itests/appender/PrometheusAppenderTest.java
new file mode 100644
index 0000000..b070e23
--- /dev/null
+++ b/itest/src/test/java/org/apache/karaf/decanter/itests/appender/PrometheusAppenderTest.java
@@ -0,0 +1,99 @@
+/*
+ * 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.karaf.decanter.itests.appender;
+
+import org.apache.karaf.itests.KarafTestSupport;
+import org.apache.karaf.jaas.boot.principal.RolePrincipal;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.ops4j.pax.exam.Configuration;
+import org.ops4j.pax.exam.Option;
+import org.ops4j.pax.exam.junit.PaxExam;
+import org.ops4j.pax.exam.karaf.options.KarafDistributionOption;
+import org.ops4j.pax.exam.spi.reactors.ExamReactorStrategy;
+import org.ops4j.pax.exam.spi.reactors.PerClass;
+import org.osgi.service.event.Event;
+import org.osgi.service.event.EventAdmin;
+
+import java.io.BufferedReader;
+import java.io.InputStreamReader;
+import java.net.HttpURLConnection;
+import java.net.URL;
+import java.util.HashMap;
+import java.util.stream.Stream;
+
+@RunWith(PaxExam.class)
+@ExamReactorStrategy(PerClass.class)
+public class PrometheusAppenderTest extends KarafTestSupport {
+
+ @Configuration
+ public Option[] config() {
+ Option[] options = new Option[]{
+ KarafDistributionOption.editConfigurationFilePut("etc/system.properties", "decanter.version", System.getProperty("decanter.version"))
+ };
+ return Stream.of(super.config(), options).flatMap(Stream::of).toArray(Option[]::new);
+ }
+
+ @Test(timeout = 60000)
+ public void test() throws Exception {
+ System.out.println("Installing Decanter Appender Prometheus ...");
+ System.out.println(executeCommand("feature:repo-add decanter " + System.getProperty("decanter.version"), new RolePrincipal("admin")));
+ System.out.println(executeCommand("feature:install decanter-appender-prometheus", new RolePrincipal("admin")));
+ String configList = executeCommand("config:list '(service.pid=org.apache.karaf.decanter.appender.prometheus)'");
+ while (!configList.contains("service.pid")) {
+ Thread.sleep(500);
+ configList = executeCommand("config:list '(service.pid=org.apache.karaf.decanter.appender.prometheus)'");
+ }
+ String httpList = executeCommand("http:list");
+ while (!httpList.contains("Deployed")) {
+ Thread.sleep(500);
+ httpList = executeCommand("http:list");
+ }
+
+ System.out.println("Sending test event ...");
+ EventAdmin dispatcher = getOsgiService(EventAdmin.class);
+ HashMap<String, Object> data = new HashMap<>();
+ data.put("Test", 0);
+ dispatcher.sendEvent(new Event("decanter/collect/test", data));
+
+ boolean found = false;
+ StringBuilder builder = new StringBuilder();
+ while (!found) {
+ URL url = new URL("http://localhost:" + getHttpPort() + "/decanter/prometheus");
+ HttpURLConnection httpURLConnection = (HttpURLConnection) url.openConnection();
+ httpURLConnection.setRequestMethod("GET");
+ httpURLConnection.setDoInput(true);
+ if (httpURLConnection.getResponseCode() == 200) {
+ found = true;
+ try (BufferedReader reader = new BufferedReader(new InputStreamReader(httpURLConnection.getInputStream()))) {
+ String line;
+ while ((line = reader.readLine()) != null) {
+ builder.append(line).append("\n");
+ }
+ }
+ }
+ }
+
+ System.out.println("");
+ System.out.println(builder.toString());
+ System.out.println("");
+
+ Assert.assertTrue(builder.toString().contains("Test 0.0"));
+ }
+
+}
diff --git a/itest/src/test/java/org/apache/karaf/decanter/itests/appender/RestAppenderTest.java b/itest/src/test/java/org/apache/karaf/decanter/itests/appender/RestAppenderTest.java
index 285de32..346d552 100644
--- a/itest/src/test/java/org/apache/karaf/decanter/itests/appender/RestAppenderTest.java
+++ b/itest/src/test/java/org/apache/karaf/decanter/itests/appender/RestAppenderTest.java
@@ -28,6 +28,7 @@
import org.ops4j.pax.exam.karaf.options.KarafDistributionOption;
import org.ops4j.pax.exam.spi.reactors.ExamReactorStrategy;
import org.ops4j.pax.exam.spi.reactors.PerClass;
+import org.osgi.framework.Bundle;
import org.osgi.service.event.Event;
import org.osgi.service.event.EventAdmin;
import org.osgi.service.http.HttpService;
@@ -65,7 +66,7 @@
return Stream.of(super.config(), options).flatMap(Stream::of).toArray(Option[]::new);
}
- @Test
+ @Test(timeout = 60000)
public void test() throws Exception {
List<String> calls = new ArrayList<>();
// register servlet
@@ -88,7 +89,13 @@
}
}, null, null);
- System.out.println(executeCommand("http:list"));
+ System.out.println("Waiting testing REST service ...");
+ String httpList = executeCommand("http:list");
+ while (!httpList.contains("Deployed")) {
+ Thread.sleep(200);
+ httpList = executeCommand("http:list");
+ }
+ System.out.println(httpList);
// configure appender
File file = new File(System.getProperty("karaf.etc"), "org.apache.karaf.decanter.appender.rest.cfg");
@@ -97,6 +104,14 @@
writer.write("marshaller.target=(dataFormat=json)");
}
+ System.out.println("Waiting org.apache.karaf.decanter.appender.rest configuration ...");
+ String configList = executeCommand("config:list '(service.pid=org.apache.karaf.decanter.appender.rest)'");
+ while (!configList.contains("service.pid")) {
+ Thread.sleep(500);
+ configList = executeCommand("config:list '(service.pid=org.apache.karaf.decanter.appender.rest)'");
+ }
+ System.out.println(configList);
+
// install decanter
System.out.println(executeCommand("feature:repo-add decanter " + System.getProperty("decanter.version")));
System.out.println(executeCommand("feature:install decanter-appender-rest", new RolePrincipal("admin")));
@@ -110,6 +125,10 @@
Event event = new Event("decanter/collect/test", data);
eventAdmin.sendEvent(event);
+ while (calls.size() != 1) {
+ Thread.sleep(200);
+ }
+
Assert.assertEquals(1, calls.size());
Assert.assertTrue(calls.get(0).contains("\"foo\":\"bar\""));
diff --git a/itest/src/test/java/org/apache/karaf/decanter/itests/appender/SocketAppenderTest.java b/itest/src/test/java/org/apache/karaf/decanter/itests/appender/SocketAppenderTest.java
index 4211831..80545cc 100644
--- a/itest/src/test/java/org/apache/karaf/decanter/itests/appender/SocketAppenderTest.java
+++ b/itest/src/test/java/org/apache/karaf/decanter/itests/appender/SocketAppenderTest.java
@@ -51,11 +51,12 @@
return Stream.of(super.config(), options).flatMap(Stream::of).toArray(Option[]::new);
}
- @Test
+ @Test(timeout = 60000)
public void test() throws Exception {
List<String> received = new ArrayList<>();
// create server socket
+ System.out.println("Starting test socket listener ...");
ServerSocket serverSocket = new ServerSocket(34343);
Runnable runnable = new Runnable() {
@Override
@@ -79,18 +80,33 @@
thread.start();
// install decanter
+ System.out.println("Installing Decanter Socket Appender ...");
System.out.println(executeCommand("feature:repo-add decanter " + System.getProperty("decanter.version")));
System.out.println(executeCommand("feature:install decanter-appender-socket", new RolePrincipal("admin")));
- Thread.sleep(2000);
+ System.out.println("Waiting org.apache.karaf.decanter.appender.socket configuration");
+ String configList = executeCommand("config:list '(service.pid=org.apache.karaf.decanter.appender.socket)'");
+ while (!configList.contains("service.pid")) {
+ Thread.sleep(500);
+ configList = executeCommand("config:list '(service.pid=org.apache.karaf.decanter.appender.socket)'");
+ }
+ System.out.println(configList);
// send event
+ System.out.println("Sending Decanter test event");
EventAdmin eventAdmin = getOsgiService(EventAdmin.class);
HashMap<String, String> data = new HashMap<>();
data.put("foo", "bar");
Event event = new Event("decanter/collect/test", data);
eventAdmin.sendEvent(event);
+ System.out.println("Waiting events ...");
+ while (received.size() != 1) {
+ Thread.sleep(200);
+ }
+
+ System.out.println(received.get(0));
+
Assert.assertEquals(1, received.size());
Assert.assertTrue(received.get(0).contains("\"foo\":\"bar\""));
diff --git a/itest/src/test/java/org/apache/karaf/decanter/itests/appender/WebsocketAppenderTest.java b/itest/src/test/java/org/apache/karaf/decanter/itests/appender/WebsocketAppenderTest.java
index 569731b..9618643 100644
--- a/itest/src/test/java/org/apache/karaf/decanter/itests/appender/WebsocketAppenderTest.java
+++ b/itest/src/test/java/org/apache/karaf/decanter/itests/appender/WebsocketAppenderTest.java
@@ -65,9 +65,17 @@
@Test
public void test() throws Exception {
// install decanter
+ System.out.println("Installing Decanter WebSocket Appender ...");
System.out.println(executeCommand("feature:repo-add decanter " + System.getProperty("decanter.version")));
System.out.println(executeCommand("feature:install decanter-appender-websocket-servlet", new RolePrincipal("admin")));
+ String configList = executeCommand("config:list '(service.pid=org.apache.karaf.decanter.appender.websocket.servlet)'");
+ while (!configList.contains("service.pid")) {
+ Thread.sleep(500);
+ configList = executeCommand("config:list '(service.pid=org.apache.karaf.decanter.appender.websocket.servlet)'");
+ }
+
+ System.out.println("Waiting websocket servlet deployed ...");
String httpList = executeCommand("http:list");
while (!httpList.contains("Deployed")) {
Thread.sleep(500);
@@ -76,6 +84,7 @@
System.out.println(httpList);
// websocket
+ System.out.println("Creating testing websocket client ...");
WebSocketClient client = new WebSocketClient();
DecanterSocket decanterSocket = new DecanterSocket();
client.start();
@@ -92,6 +101,11 @@
decanterSocket.awaitClose(20, TimeUnit.SECONDS);
+ System.out.println("Waiting event ...");
+ while (decanterSocket.messages.size() != 1) {
+ Thread.sleep(200);
+ }
+
Assert.assertEquals(1, decanterSocket.messages.size());
Assert.assertTrue(decanterSocket.messages.get(0).contains("\"foo\":\"bar\""));
diff --git a/itest/src/test/java/org/apache/karaf/decanter/itests/collector/CamelCollectorTest.java b/itest/src/test/java/org/apache/karaf/decanter/itests/collector/CamelCollectorTest.java
index 3d66704..2ac30a0 100644
--- a/itest/src/test/java/org/apache/karaf/decanter/itests/collector/CamelCollectorTest.java
+++ b/itest/src/test/java/org/apache/karaf/decanter/itests/collector/CamelCollectorTest.java
@@ -103,14 +103,13 @@
ProducerTemplate producerTemplate = camelContext.createProducerTemplate();
producerTemplate.sendBodyAndHeader("direct:test", "This is a test", "testHeader", "testValue");
- // TODO two events ?
Assert.assertEquals(2, received.size());
Assert.assertEquals("decanter/collect/camel/tracer", received.get(0).getTopic());
Assert.assertEquals("context-test", received.get(0).getProperty("camelContextName"));
Assert.assertEquals("InOnly", received.get(0).getProperty("exchangePattern"));
Assert.assertEquals("camelTracer", received.get(0).getProperty("type"));
- Assert.assertEquals("log://foo", received.get(0).getProperty("toNode"));
+ Assert.assertEquals("log:foo", received.get(0).getProperty("to1.label"));
Assert.assertEquals("route-test", received.get(0).getProperty("routeId"));
Assert.assertEquals("direct://test", received.get(0).getProperty("fromEndpointUri"));
Assert.assertEquals("root", received.get(0).getProperty("karafName"));
@@ -138,7 +137,7 @@
// create route with notifier
EventAdmin eventAdmin = getOsgiService(EventAdmin.class);
- DecanterEventNotifier notifier = new DecanterEventNotifier();
+ DecanterEventNotifier notifier = new DecanterEventNotifier();
notifier.setEventAdmin(eventAdmin);
RouteBuilder builder = new RouteBuilder() {
diff --git a/itest/src/test/java/org/apache/karaf/decanter/itests/collector/EventAdminCollectorTest.java b/itest/src/test/java/org/apache/karaf/decanter/itests/collector/EventAdminCollectorTest.java
index ea51ac3..0101efa 100644
--- a/itest/src/test/java/org/apache/karaf/decanter/itests/collector/EventAdminCollectorTest.java
+++ b/itest/src/test/java/org/apache/karaf/decanter/itests/collector/EventAdminCollectorTest.java
@@ -52,14 +52,15 @@
return Stream.of(super.config(), options).flatMap(Stream::of).toArray(Option[]::new);
}
- @Test
+ @Test(timeout = 60000)
public void test() throws Exception {
// install decanter
System.out.println(executeCommand("feature:repo-add decanter " + System.getProperty("decanter.version")));
System.out.println(executeCommand("feature:install decanter-collector-eventadmin", new RolePrincipal("admin")));
- // just gives time for the factory to populate
- Thread.sleep(1000);
+ // waiting for collector service
+ System.out.println("Waiting Decanter Collector EventAdmin");
+ getOsgiService(EventAdmin.class);
// add a event handler
List<Event> received = new ArrayList();
@@ -80,10 +81,26 @@
Event testEvent = new Event("org/apache/karaf/test", data);
eventAdmin.sendEvent(testEvent);
+ System.out.println("Waiting Decanter events ...");
+ while (received.size() < 1) {
+ Thread.sleep(100);
+ }
+
+ System.out.println("");
+
+ for (int i = 0; i < received.size(); i++) {
+ for (String property : received.get(i).getPropertyNames()) {
+ System.out.println(property + " = " + received.get(i).getProperty(property));
+ }
+ System.out.println("===========");
+ }
+
+ System.out.println("");
+
Assert.assertTrue(received.size() >= 1);
for (Event event : received) {
- if (event.getTopic().equals("decanter/collect/eventadmin/org/apache/karaf/test")) {
+ if (event.getProperty("event.topics").equals("decanter/collect/eventadmin/org/apache/karaf/test")) {
Assert.assertEquals("bar", event.getProperty("foo"));
Assert.assertEquals("root", event.getProperty("karafName"));
Assert.assertEquals("eventadmin", event.getProperty("type"));
@@ -91,7 +108,7 @@
}
}
- throw new IllegalStateException("No karaf eventadmin event received");
+ throw new IllegalStateException("No decanter eventadmin event received");
}
}
diff --git a/itest/src/test/java/org/apache/karaf/decanter/itests/collector/FileCollectorTest.java b/itest/src/test/java/org/apache/karaf/decanter/itests/collector/FileCollectorTest.java
index 2ed9857..40b0be1 100644
--- a/itest/src/test/java/org/apache/karaf/decanter/itests/collector/FileCollectorTest.java
+++ b/itest/src/test/java/org/apache/karaf/decanter/itests/collector/FileCollectorTest.java
@@ -53,16 +53,21 @@
return Stream.of(super.config(), options).flatMap(Stream::of).toArray(Option[]::new);
}
- @Test(timeout = 60000)
+ @Test(timeout = 30000)
public void test() throws Exception {
// install decanter
+ System.out.println("Installing Decanter Collector File ...");
System.out.println(executeCommand("feature:repo-add decanter " + System.getProperty("decanter.version")));
System.out.println(executeCommand("feature:install decanter-collector-file", new RolePrincipal("admin")));
- // wait for the factory
- Thread.sleep(1000);
+ String configList = executeCommand("config:list '(service.factoryPid=org.apache.karaf.decanter.collector.file)'");
+ while (!configList.contains("service.pid")) {
+ Thread.sleep(500);
+ configList = executeCommand("config:list '(service.factoryPid=org.apache.karaf.decanter.collector.file)'");
+ }
// add a event handler
+ System.out.println("Adding test event handler ...");
List<Event> received = new ArrayList();
EventHandler eventHandler = new EventHandler() {
@Override
@@ -75,17 +80,29 @@
bundleContext.registerService(EventHandler.class, eventHandler, serviceProperties);
// append data in the file
+ System.out.println("Writing data in test.log file ...");
try (BufferedWriter writer = new BufferedWriter(new FileWriter(new File("test.log")))) {
writer.write("This is a test\n");
writer.write("Another test\n");
writer.flush();
}
- System.out.println("Waiting message ...");
- while (received.size() == 0) {
+ System.out.println("Waiting events ...");
+ while (received.size() < 2) {
Thread.sleep(500);
}
+ System.out.println("");
+
+ for (int i = 0; i < received.size(); i++) {
+ for (String property : received.get(i).getPropertyNames()) {
+ System.out.println(property + " = " + received.get(i).getProperty(property));
+ }
+ System.out.println("========");
+ }
+
+ System.out.println("");
+
Assert.assertEquals(2, received.size());
// first line
diff --git a/itest/src/test/java/org/apache/karaf/decanter/itests/collector/JdbcCollectorTest.java b/itest/src/test/java/org/apache/karaf/decanter/itests/collector/JdbcCollectorTest.java
index e855e7e..eb8ad58 100644
--- a/itest/src/test/java/org/apache/karaf/decanter/itests/collector/JdbcCollectorTest.java
+++ b/itest/src/test/java/org/apache/karaf/decanter/itests/collector/JdbcCollectorTest.java
@@ -61,7 +61,7 @@
return Stream.of(super.config(), options).flatMap(Stream::of).toArray(Option[]::new);
}
- @Test
+ @Test(timeout = 120000)
public void test() throws Exception {
// install database
System.out.println(executeCommand("feature:install jdbc", new RolePrincipal("admin")));
@@ -88,7 +88,12 @@
Thread.sleep(1000);
// list scheduler jobs
- System.out.println(executeCommand("scheduler:list"));
+ String schedulerList = executeCommand("scheduler:list");
+ while (!schedulerList.contains("decanter-collector-jdbc")) {
+ Thread.sleep(200);
+ schedulerList = executeCommand("scheduler:list");
+ }
+ System.out.println(schedulerList);
// add a event handler
List<Event> received = new ArrayList();
@@ -108,6 +113,17 @@
Thread.sleep(500);
}
+ System.out.println("");
+
+ for (int i = 0; i < received.size(); i++) {
+ for (String property : received.get(i).getPropertyNames()) {
+ System.out.println(property + " = " + received.get(i).getProperty(property));
+ }
+ System.out.println("========");
+ }
+
+ System.out.println("");
+
Assert.assertTrue(received.size() >= 1);
Assert.assertEquals("decanter/collect/jdbc", received.get(0).getTopic());
diff --git a/itest/src/test/java/org/apache/karaf/decanter/itests/collector/JettyCollectorTest.java b/itest/src/test/java/org/apache/karaf/decanter/itests/collector/JettyCollectorTest.java
new file mode 100644
index 0000000..3cb1fa1
--- /dev/null
+++ b/itest/src/test/java/org/apache/karaf/decanter/itests/collector/JettyCollectorTest.java
@@ -0,0 +1,137 @@
+/*
+ * 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.karaf.decanter.itests.collector;
+
+import org.apache.karaf.itests.KarafTestSupport;
+import org.apache.karaf.jaas.boot.principal.RolePrincipal;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.ops4j.pax.exam.Configuration;
+import org.ops4j.pax.exam.MavenUtils;
+import org.ops4j.pax.exam.Option;
+import org.ops4j.pax.exam.junit.PaxExam;
+import org.ops4j.pax.exam.karaf.options.KarafDistributionOption;
+import org.ops4j.pax.exam.spi.reactors.ExamReactorStrategy;
+import org.ops4j.pax.exam.spi.reactors.PerClass;
+import org.osgi.service.event.Event;
+import org.osgi.service.event.EventAdmin;
+import org.osgi.service.event.EventConstants;
+import org.osgi.service.event.EventHandler;
+import org.osgi.service.http.HttpService;
+
+import javax.inject.Inject;
+import javax.servlet.ServletException;
+import javax.servlet.http.HttpServlet;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.io.OutputStreamWriter;
+import java.io.PrintWriter;
+import java.net.HttpURLConnection;
+import java.net.URL;
+import java.util.ArrayList;
+import java.util.Hashtable;
+import java.util.List;
+import java.util.stream.Stream;
+
+@RunWith(PaxExam.class)
+@ExamReactorStrategy(PerClass.class)
+public class JettyCollectorTest extends KarafTestSupport {
+
+ @Inject
+ HttpService httpService;
+
+ @Configuration
+ public Option[] config() {
+ String karafVersion = MavenUtils.getArtifactVersion("org.apache.karaf", "apache-karaf");
+ Option[] options = new Option[]{
+ KarafDistributionOption.editConfigurationFilePut("etc/system.properties", "decanter.version", System.getProperty("decanter.version")),
+ KarafDistributionOption.features("mvn:org.apache.karaf.features/standard/" + karafVersion + "/xml/features", "http")
+ };
+ return Stream.of(super.config(), options).flatMap(Stream::of).toArray(Option[]::new);
+ }
+
+ @Test(timeout = 60000)
+ public void test() throws Exception {
+ System.out.println("Registering test servlet ...");
+ httpService.registerServlet("/test", new HttpServlet() {
+ @Override
+ public void doGet(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
+ try (BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(response.getOutputStream()))) {
+ writer.write("{\"test\":\"test\"}");
+ writer.flush();
+ }
+ }
+ }, null, null);
+ String httpList = executeCommand("http:list");
+ while (!httpList.contains("Deployed")) {
+ Thread.sleep(200);
+ httpList = executeCommand("http:list");
+ }
+ System.out.println(httpList);
+
+ System.out.println("Installing Decanter Jetty Collector ...");
+ System.out.println(executeCommand("feature:repo-add decanter " + System.getProperty("decanter.version"), new RolePrincipal("admin")));
+ System.out.println(executeCommand("feature:install decanter-collector-jetty", new RolePrincipal("admin")));
+
+ getOsgiService("org.eclipse.jetty.server.Handler");
+
+ System.out.println("Registering event handler ...");
+ List<Event> received = new ArrayList();
+ EventHandler eventHandler = new EventHandler() {
+ @Override
+ public void handleEvent(Event event) {
+ received.add(event);
+ }
+ };
+ Hashtable serviceProperties = new Hashtable();
+ serviceProperties.put(EventConstants.EVENT_TOPIC, "decanter/collect/*");
+ bundleContext.registerService(EventHandler.class, eventHandler, serviceProperties);
+
+ System.out.println("Calling servlet ...");
+ // send data to rest servlet collector
+ URL url = new URL("http://localhost:" + getHttpPort() + "/test");
+ HttpURLConnection connection = (HttpURLConnection) url.openConnection();
+ connection.setRequestMethod("GET");
+ connection.setRequestProperty("Content-Type", "application/json");
+ Assert.assertEquals(200, connection.getResponseCode());
+ System.out.println(connection.getResponseMessage());
+
+ System.out.println("Waiting events ...");
+ while (received.size() < 1) {
+ Thread.sleep(200);
+ }
+
+ Assert.assertEquals(1, received.size());
+
+ for (int i = 0; i < received.size(); i++) {
+ for (String property : received.get(i).getPropertyNames()) {
+ System.out.println(property + " = " + received.get(i).getProperty(property));
+ }
+ System.out.println("========");
+ }
+
+ Assert.assertEquals("decanter/collect/jetty", received.get(0).getProperty("event.topics"));
+ Assert.assertEquals("GET", received.get(0).getProperty("request.method"));
+ Assert.assertEquals("/test", received.get(0).getProperty("request.pathInfo"));
+ Assert.assertEquals("/test", received.get(0).getProperty("request.requestURI"));
+ Assert.assertEquals(200, received.get(0).getProperty("response.status"));
+ }
+
+}
diff --git a/itest/src/test/java/org/apache/karaf/decanter/itests/collector/JmsCollectorTest.java b/itest/src/test/java/org/apache/karaf/decanter/itests/collector/JmsCollectorTest.java
index 77aa1f2..4b90d81 100644
--- a/itest/src/test/java/org/apache/karaf/decanter/itests/collector/JmsCollectorTest.java
+++ b/itest/src/test/java/org/apache/karaf/decanter/itests/collector/JmsCollectorTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.karaf.decanter.itests.collector;
+import org.apache.karaf.features.FeaturesService;
import org.apache.karaf.itests.KarafTestSupport;
import org.junit.Assert;
import org.junit.Test;
@@ -25,6 +26,7 @@
import org.ops4j.pax.exam.Option;
import org.ops4j.pax.exam.junit.PaxExam;
import org.ops4j.pax.exam.karaf.options.KarafDistributionOption;
+import org.ops4j.pax.exam.karaf.options.KarafFeaturesOption;
import org.ops4j.pax.exam.spi.reactors.ExamReactorStrategy;
import org.ops4j.pax.exam.spi.reactors.PerClass;
import org.osgi.service.event.Event;
@@ -32,6 +34,7 @@
import org.osgi.service.event.EventHandler;
import java.util.ArrayList;
+import java.util.EnumSet;
import java.util.Hashtable;
import java.util.List;
import java.util.stream.Stream;
@@ -59,11 +62,12 @@
return Stream.of(super.config(), options).flatMap(Stream::of).toArray(Option[]::new);
}
- @Test
+ @Test(timeout = 60000)
public void test() throws Exception {
// install activemq
- featureService.installFeature("aries-blueprint");
- featureService.installFeature("activemq-broker-noweb");
+ featureService.installFeature("shell-compat");
+ featureService.installFeature("activemq-broker-noweb", EnumSet.of(FeaturesService.Option.NoAutoRefreshBundles));
+ Thread.sleep(2000);
// install jms
featureService.installFeature("jms");
@@ -71,7 +75,12 @@
// create connection factory
System.out.println(executeCommand("jms:create decanter"));
- Thread.sleep(2000);
+ String jmsConnectionFactory = executeCommand("jms:connectionfactories");
+ while (!jmsConnectionFactory.contains("jms/decanter")) {
+ Thread.sleep(200);
+ jmsConnectionFactory = executeCommand("jms:connectionfactories");
+ }
+
System.out.println(executeCommand("jms:connectionfactories"));
System.out.println(executeCommand("jms:info jms/decanter"));
@@ -93,14 +102,29 @@
// send message to JMS queue
System.out.println(executeCommand("jms:send jms/decanter decanter '{\"foo\":\"bar\"}'"));
- while (received.size() == 0) {
- Thread.sleep(500);
+ while (received.size() < 1) {
+ Thread.sleep(200);
}
+ System.out.println("");
+
+ for (int i = 0; i < received.size(); i++) {
+ for (String property : received.get(i).getPropertyNames()) {
+ System.out.println(property + " = " + received.get(i).getProperty(property));
+ }
+ System.out.println("=========");
+ }
+
+ System.out.println("");
+
Assert.assertEquals(1, received.size());
Assert.assertEquals("decanter/collect/jms/decanter", received.get(0).getTopic());
- Assert.assertEquals("bar", received.get(0).getProperty("foo"));
+ if (received.get(0).getProperty("payload") != null) {
+ Assert.assertTrue(((String) received.get(0).getProperty("payload")).contains("{\"foo\":\"bar\"}"));
+ } else {
+ Assert.assertEquals("bar", received.get(0).getProperty("foo"));
+ }
Assert.assertEquals("jms", received.get(0).getProperty("type"));
Assert.assertEquals("root", received.get(0).getProperty("karafName"));
}
diff --git a/itest/src/test/java/org/apache/karaf/decanter/itests/collector/JmxCollectorTest.java b/itest/src/test/java/org/apache/karaf/decanter/itests/collector/JmxCollectorTest.java
index bcd87c6..1168ba0 100644
--- a/itest/src/test/java/org/apache/karaf/decanter/itests/collector/JmxCollectorTest.java
+++ b/itest/src/test/java/org/apache/karaf/decanter/itests/collector/JmxCollectorTest.java
@@ -48,7 +48,7 @@
return Stream.of(super.config(), options).flatMap(Stream::of).toArray(Option[]::new);
}
- @Test
+ @Test(timeout = 120000)
public void test() throws Exception {
// install decanter
System.out.println(executeCommand("feature:repo-add decanter " + System.getProperty("decanter.version")));
@@ -72,6 +72,17 @@
Thread.sleep(500);
}
+ System.out.println("");
+
+ for (int i = 0; i < received.size(); i++) {
+ for (String property : received.get(i).getPropertyNames()) {
+ System.out.println(property + " = " + received.get(i).getProperty(property));
+ }
+ System.out.println("========");
+ }
+
+ System.out.println("");
+
Assert.assertTrue(received.size() >= 1);
Event event = received.get(0);
diff --git a/itest/src/test/java/org/apache/karaf/decanter/itests/collector/Log4jSocketCollectorTest.java b/itest/src/test/java/org/apache/karaf/decanter/itests/collector/Log4jSocketCollectorTest.java
index 071171c..68029ba 100644
--- a/itest/src/test/java/org/apache/karaf/decanter/itests/collector/Log4jSocketCollectorTest.java
+++ b/itest/src/test/java/org/apache/karaf/decanter/itests/collector/Log4jSocketCollectorTest.java
@@ -35,6 +35,7 @@
import org.osgi.service.event.EventHandler;
import java.io.ObjectOutputStream;
+import java.net.ConnectException;
import java.net.Socket;
import java.util.ArrayList;
import java.util.HashMap;
@@ -54,14 +55,12 @@
return Stream.of(super.config(), options).flatMap(Stream::of).toArray(Option[]::new);
}
- @Test
+ @Test(timeout = 60000)
public void test() throws Exception {
// install decanter
System.out.println(executeCommand("feature:repo-add decanter " + System.getProperty("decanter.version")));
System.out.println(executeCommand("feature:install decanter-collector-log-socket", new RolePrincipal("admin")));
- Thread.sleep(2000);
-
// create event handler
List<Event> received = new ArrayList();
EventHandler eventHandler = new EventHandler() {
@@ -75,7 +74,15 @@
bundleContext.registerService(EventHandler.class, eventHandler, serviceProperties);
LoggingEvent loggingEvent = new LoggingEvent("test", Category.getInstance("logger"), System.currentTimeMillis(), Level.toLevel("INFO"), "Test", "thread", null, "test", null, new HashMap());
- Socket socket = new Socket("localhost", 4560);
+ Socket socket = null;
+ while (socket == null) {
+ Thread.sleep(100);
+ try {
+ socket = new Socket("localhost", 4560);
+ } catch (ConnectException connectException) {
+ // wait
+ }
+ }
try (ObjectOutputStream objectOutputStream = new ObjectOutputStream(socket.getOutputStream())) {
objectOutputStream.writeObject(loggingEvent);
}
@@ -84,6 +91,17 @@
Thread.sleep(500);
}
+ System.out.println("");
+
+ for (int i = 0; i < received.size(); i++) {
+ for (String property : received.get(i).getPropertyNames()) {
+ System.out.println(property + " = " + received.get(i).getProperty(property));
+ }
+ System.out.println("========");
+ }
+
+ System.out.println("");
+
Assert.assertEquals(1, received.size());
Assert.assertEquals("decanter/collect/log/logger", received.get(0).getTopic());
diff --git a/itest/src/test/java/org/apache/karaf/decanter/itests/collector/LogCollectorTest.java b/itest/src/test/java/org/apache/karaf/decanter/itests/collector/LogCollectorTest.java
index bcecbba..a7b222c 100644
--- a/itest/src/test/java/org/apache/karaf/decanter/itests/collector/LogCollectorTest.java
+++ b/itest/src/test/java/org/apache/karaf/decanter/itests/collector/LogCollectorTest.java
@@ -52,7 +52,7 @@
return Stream.of(super.config(), options).flatMap(Stream::of).toArray(Option[]::new);
}
- @Test
+ @Test(timeout = 60000)
public void test() throws Exception {
// install log collector
System.out.println(executeCommand("feature:repo-add decanter " + System.getProperty("decanter.version")));
@@ -72,6 +72,21 @@
LOGGER.info("This is a test");
+ while (received.size() < 1) {
+ Thread.sleep(200);
+ }
+
+ System.out.println("");
+
+ for (int i = 0; i < received.size(); i++) {
+ for (String property : received.get(i).getPropertyNames()) {
+ System.out.println(property + " = " + received.get(i).getProperty(property));
+ }
+ System.out.println("========");
+ }
+
+ System.out.println("");
+
Assert.assertEquals(1, received.size());
Assert.assertEquals("decanter/collect/log/org_apache_karaf_decanter_itests_collector_LogCollectorTest", received.get(0).getTopic());
Assert.assertEquals("INFO", received.get(0).getProperty("level"));
diff --git a/itest/src/test/java/org/apache/karaf/decanter/itests/collector/MqttCollectorTest.java b/itest/src/test/java/org/apache/karaf/decanter/itests/collector/MqttCollectorTest.java
index 31d62b3..b5b1b54 100644
--- a/itest/src/test/java/org/apache/karaf/decanter/itests/collector/MqttCollectorTest.java
+++ b/itest/src/test/java/org/apache/karaf/decanter/itests/collector/MqttCollectorTest.java
@@ -23,6 +23,7 @@
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.junit.Assert;
+import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.ops4j.pax.exam.Configuration;
@@ -40,6 +41,9 @@
import org.osgi.service.event.EventConstants;
import org.osgi.service.event.EventHandler;
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.Socket;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Hashtable;
@@ -75,12 +79,22 @@
}
@Test(timeout = 60000)
+ @Ignore("Flaky test")
public void test() throws Exception {
+ System.out.println("Waiting ActiveMQ MQTT connector ...");
+ while (true) {
+ Thread.sleep(200);
+ try {
+ Socket socket = new Socket(InetAddress.getLocalHost(), 1883);
+ break;
+ } catch (IOException ioException) {
+ // no-op
+ }
+ }
+
// install decanter
System.out.println(executeCommand("feature:install decanter-collector-mqtt", new RolePrincipal("admin")));
- Thread.sleep(500);
-
// create event handler
List<Event> received = new ArrayList();
EventHandler eventHandler = new EventHandler() {
@@ -97,6 +111,8 @@
MqttClient client = new MqttClient("tcp://localhost:1883", "d:decanter:collector:test");
client.connect();
MqttMessage message = new MqttMessage();
+ message.setQos(0);
+ message.setRetained(true);
message.setPayload("This is a test".getBytes(StandardCharsets.UTF_8));
client.publish("decanter", message);
@@ -105,6 +121,17 @@
Thread.sleep(500);
}
+ System.out.println("");
+
+ for (int i = 0; i < received.size(); i++) {
+ for (String property : received.get(i).getPropertyNames()) {
+ System.out.println(property + " = " + received.get(i).getProperty(property));
+ }
+ System.out.println("========");
+ }
+
+ System.out.println("");
+
Assert.assertEquals(1, received.size());
Assert.assertEquals("decanter/collect/mqtt/decanter", received.get(0).getTopic());
@@ -112,6 +139,7 @@
Assert.assertEquals("root", received.get(0).getProperty("karafName"));
Assert.assertEquals("mqtt", received.get(0).getProperty("type"));
+ client.disconnect();
}
}
diff --git a/itest/src/test/java/org/apache/karaf/decanter/itests/collector/OshiCollectorTest.java b/itest/src/test/java/org/apache/karaf/decanter/itests/collector/OshiCollectorTest.java
new file mode 100644
index 0000000..0a7bd6c
--- /dev/null
+++ b/itest/src/test/java/org/apache/karaf/decanter/itests/collector/OshiCollectorTest.java
@@ -0,0 +1,97 @@
+/*
+ * 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.karaf.decanter.itests.collector;
+
+import org.apache.karaf.itests.KarafTestSupport;
+import org.apache.karaf.jaas.boot.principal.RolePrincipal;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.ops4j.pax.exam.Configuration;
+import org.ops4j.pax.exam.Option;
+import org.ops4j.pax.exam.junit.PaxExam;
+import org.ops4j.pax.exam.karaf.options.KarafDistributionOption;
+import org.ops4j.pax.exam.spi.reactors.ExamReactorStrategy;
+import org.ops4j.pax.exam.spi.reactors.PerClass;
+import org.osgi.service.event.Event;
+import org.osgi.service.event.EventConstants;
+import org.osgi.service.event.EventHandler;
+
+import java.util.ArrayList;
+import java.util.Hashtable;
+import java.util.List;
+import java.util.stream.Stream;
+
+@RunWith(PaxExam.class)
+@ExamReactorStrategy(PerClass.class)
+public class OshiCollectorTest extends KarafTestSupport {
+
+ @Configuration
+ public Option[] config() {
+ Option[] options = new Option[]{
+ KarafDistributionOption.editConfigurationFilePut("etc/system.properties", "decanter.version", System.getProperty("decanter.version"))
+ };
+ return Stream.of(super.config(), options).flatMap(Stream::of).toArray(Option[]::new);
+ }
+
+ @Test(timeout = 120000)
+ public void test() throws Exception {
+ System.out.println("Installing Decanter OSHI Collector ...");
+ System.out.println(executeCommand("feature:repo-add decanter " + System.getProperty("decanter.version"), new RolePrincipal("admin")));
+ System.out.println(executeCommand("feature:install decanter-collector-oshi", new RolePrincipal("admin")));
+
+ String configList = executeCommand("config:list '(service.pid=org.apache.karaf.decanter.collector.oshi)'");
+ while (!configList.contains("service.pid")) {
+ Thread.sleep(500);
+ configList = executeCommand("config:list '(service.pid=org.apache.karaf.decanter.collector.oshi)'");
+ }
+
+ System.out.println("Adding event handler ...");
+ final List<Event> received = new ArrayList<>();
+ EventHandler eventHandler = new EventHandler() {
+ @Override
+ public void handleEvent(Event event) {
+ received.add(event);
+ }
+ };
+ Hashtable serviceProperties = new Hashtable();
+ serviceProperties.put(EventConstants.EVENT_TOPIC, "decanter/collect/*");
+ bundleContext.registerService(EventHandler.class, eventHandler, serviceProperties);
+
+ System.out.println("Waiting events ...");
+ while (received.size() < 1) {
+ Thread.sleep(500);
+ }
+
+ System.out.println("");
+
+ for (int i = 0; i < received.size(); i++) {
+ for (String property : received.get(i).getPropertyNames()) {
+ System.out.println(property + " = " + received.get(i).getProperty(property));
+ }
+ System.out.println("======");
+ }
+
+ System.out.println("");
+
+ Assert.assertEquals(1, received.size());
+
+ Assert.assertEquals("decanter/collect/oshi", received.get(0).getProperty("event.topics"));
+ Assert.assertEquals("oshi", received.get(0).getProperty("type"));
+ }
+
+}
diff --git a/itest/src/test/java/org/apache/karaf/decanter/itests/collector/PrometheusCollectorTest.java b/itest/src/test/java/org/apache/karaf/decanter/itests/collector/PrometheusCollectorTest.java
new file mode 100644
index 0000000..4cca75d
--- /dev/null
+++ b/itest/src/test/java/org/apache/karaf/decanter/itests/collector/PrometheusCollectorTest.java
@@ -0,0 +1,135 @@
+/*
+ * 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.karaf.decanter.itests.collector;
+
+import org.apache.karaf.itests.KarafTestSupport;
+import org.apache.karaf.jaas.boot.principal.RolePrincipal;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.ops4j.pax.exam.Configuration;
+import org.ops4j.pax.exam.MavenUtils;
+import org.ops4j.pax.exam.Option;
+import org.ops4j.pax.exam.junit.PaxExam;
+import org.ops4j.pax.exam.karaf.options.KarafDistributionOption;
+import org.ops4j.pax.exam.spi.reactors.ExamReactorStrategy;
+import org.ops4j.pax.exam.spi.reactors.PerClass;
+import org.osgi.service.event.Event;
+import org.osgi.service.event.EventConstants;
+import org.osgi.service.event.EventHandler;
+import org.osgi.service.http.HttpService;
+
+import javax.inject.Inject;
+import javax.servlet.ServletException;
+import javax.servlet.http.HttpServlet;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+import java.io.*;
+import java.util.ArrayList;
+import java.util.Hashtable;
+import java.util.List;
+import java.util.stream.Stream;
+
+@RunWith(PaxExam.class)
+@ExamReactorStrategy(PerClass.class)
+public class PrometheusCollectorTest extends KarafTestSupport {
+
+ @Inject
+ private HttpService httpService;
+
+ @Configuration
+ public Option[] config() {
+ String karafVersion = MavenUtils.getArtifactVersion("org.apache.karaf", "apache-karaf");
+ Option[] options = new Option[]{
+ KarafDistributionOption.editConfigurationFilePut("etc/system.properties", "decanter.version", System.getProperty("decanter.version")),
+ KarafDistributionOption.features("mvn:org.apache.karaf.features/standard/" + karafVersion + "/xml/features", "http")
+ };
+ return Stream.of(super.config(), options).flatMap(Stream::of).toArray(Option[]::new);
+ }
+
+ @Test(timeout = 120000)
+ public void test() throws Exception {
+ System.out.println("Deploying Prometheus test servlet ...");
+ httpService.registerServlet("/prometheus", new HttpServlet() {
+ @Override
+ public void doGet(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
+ try (BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(response.getOutputStream()))) {
+ writer.write("# HELP Test Test\n");
+ writer.write("# TYPE Test gauge\n");
+ writer.write("Test 0.0");
+ writer.flush();
+ }
+ }
+ }, null, null);
+
+ String httpList = executeCommand("http:list");
+ while (!httpList.contains("Deployed")) {
+ Thread.sleep(500);
+ httpList = executeCommand("http:list");
+ }
+
+ System.out.println("Adding test event handler ...");
+ // create event handler
+ List<Event> received = new ArrayList();
+ EventHandler eventHandler = new EventHandler() {
+ @Override
+ public void handleEvent(Event event) {
+ received.add(event);
+ }
+ };
+ Hashtable serviceProperties = new Hashtable();
+ serviceProperties.put(EventConstants.EVENT_TOPIC, "decanter/collect/*");
+ bundleContext.registerService(EventHandler.class, eventHandler, serviceProperties);
+
+ System.out.println("Installing Decanter Collector Prometheus ...");
+ File file = new File(System.getProperty("karaf.etc"), "org.apache.karaf.decanter.collector.prometheus.cfg");
+ try (BufferedWriter writer = new BufferedWriter(new FileWriter(file))) {
+ writer.write("prometheus.url=http://localhost:" + getHttpPort() + "/prometheus");
+ writer.flush();
+ }
+ String configList = executeCommand("config:list '(service.pid=org.apache.karaf.decanter.collector.prometheus)'");
+ while (!configList.contains("service.pid")) {
+ Thread.sleep(500);
+ configList = executeCommand("config:list '(service.pid=org.apache.karaf.decanter.collector.prometheus)'");
+ }
+ System.out.println(executeCommand("feature:repo-add decanter " + System.getProperty("decanter.version"), new RolePrincipal("admin")));
+ System.out.println(executeCommand("feature:install decanter-collector-prometheus", new RolePrincipal("admin")));
+
+ System.out.println("Waiting events ...");
+ while (received.size() < 1) {
+ Thread.sleep(500);
+ }
+
+ System.out.println("");
+
+ for (int i = 0; i < received.size(); i++) {
+ for (String property : received.get(i).getPropertyNames()) {
+ System.out.println(property + " = " + received.get(i).getProperty(property));
+ }
+ System.out.println("========");
+ }
+
+ System.out.println("");
+
+ Assert.assertEquals(1, received.size());
+
+ Assert.assertEquals(0.0, received.get(0).getProperty("Test"));
+ Assert.assertEquals("decanter/collect/prometheus", received.get(0).getProperty("event.topics"));
+ Assert.assertEquals("prometheus", received.get(0).getProperty("type"));
+ }
+
+}
diff --git a/itest/src/test/java/org/apache/karaf/decanter/itests/collector/RestCollectorTest.java b/itest/src/test/java/org/apache/karaf/decanter/itests/collector/RestCollectorTest.java
index 522184e..8c59043 100644
--- a/itest/src/test/java/org/apache/karaf/decanter/itests/collector/RestCollectorTest.java
+++ b/itest/src/test/java/org/apache/karaf/decanter/itests/collector/RestCollectorTest.java
@@ -65,8 +65,9 @@
return Stream.of(super.config(), options).flatMap(Stream::of).toArray(Option[]::new);
}
- @Test
+ @Test(timeout = 240000)
public void test() throws Exception {
+ System.out.println("Deploying test servlet ...");
// register servlet
httpService.registerServlet("/test", new HttpServlet() {
@Override
@@ -85,6 +86,7 @@
}
Assert.assertTrue(httpList.contains("/test"));
+ System.out.println("Adding test event handler ...");
// create event handler
List<Event> received = new ArrayList();
EventHandler eventHandler = new EventHandler() {
@@ -98,12 +100,19 @@
bundleContext.registerService(EventHandler.class, eventHandler, serviceProperties);
// configure collector
+ System.out.println("Installing Decanter Collector Rest ...");
File file = new File(System.getProperty("karaf.etc"), "org.apache.karaf.decanter.collector.rest-1.cfg");
try (BufferedWriter writer = new BufferedWriter(new FileWriter(file))) {
writer.write("url=http://localhost:" + getHttpPort() + "\n");
writer.write("paths=test\n");
writer.write("unmarshaller.target=(dataFormat=json)");
}
+ String configList = executeCommand("config:list '(service.factoryPid=org.apache.karaf.decanter.collector.rest)'");
+ while (!configList.contains("service.pid")) {
+ Thread.sleep(500);
+ configList = executeCommand("config:list '(service.factoryPid=org.apache.karaf.decanter.collector.rest)'");
+ }
+ System.out.println(configList);
// install decanter
System.out.println(executeCommand("feature:repo-add decanter " + System.getProperty("decanter.version")));
@@ -114,6 +123,17 @@
Thread.sleep(500);
}
+ System.out.println("");
+
+ for (int i = 0; i < received.size(); i++) {
+ for (String property : received.get(i).getPropertyNames()) {
+ System.out.println(property + " = " + received.get(i).getProperty(property));
+ }
+ System.out.println("========");
+ }
+
+ System.out.println("");
+
Assert.assertEquals(1, received.size());
Assert.assertEquals("decanter/collect/rest", received.get(0).getTopic());
diff --git a/itest/src/test/java/org/apache/karaf/decanter/itests/collector/RestServletCollectorTest.java b/itest/src/test/java/org/apache/karaf/decanter/itests/collector/RestServletCollectorTest.java
index f737cae..e57d93e 100644
--- a/itest/src/test/java/org/apache/karaf/decanter/itests/collector/RestServletCollectorTest.java
+++ b/itest/src/test/java/org/apache/karaf/decanter/itests/collector/RestServletCollectorTest.java
@@ -53,7 +53,7 @@
return Stream.of(super.config(), options).flatMap(Stream::of).toArray(Option[]::new);
}
- @Test
+ @Test(timeout = 60000)
public void test() throws Exception {
// install decanter
System.out.println(executeCommand("feature:repo-add decanter " + System.getProperty("decanter.version")));
@@ -95,6 +95,17 @@
Thread.sleep(500);
}
+ System.out.println("");
+
+ for (int i = 0; i < received.size(); i++) {
+ for (String property : received.get(i).getPropertyNames()) {
+ System.out.println(property + " = " + received.get(i).getProperty(property));
+ }
+ System.out.println("========");
+ }
+
+ System.out.println("");
+
Assert.assertEquals("decanter/collect/rest-servlet", received.get(0).getTopic());
Assert.assertEquals("root", received.get(0).getProperty("karafName"));
Assert.assertEquals("restservlet", received.get(0).getProperty("type"));
diff --git a/itest/src/test/java/org/apache/karaf/decanter/itests/collector/SoapCollectorTest.java b/itest/src/test/java/org/apache/karaf/decanter/itests/collector/SoapCollectorTest.java
index 5f457b4..67949c6 100644
--- a/itest/src/test/java/org/apache/karaf/decanter/itests/collector/SoapCollectorTest.java
+++ b/itest/src/test/java/org/apache/karaf/decanter/itests/collector/SoapCollectorTest.java
@@ -65,7 +65,7 @@
return Stream.of(super.config(), options).flatMap(Stream::of).toArray(Option[]::new);
}
- @Test
+ @Test(timeout = 120000)
public void test() throws Exception {
httpService.registerServlet("/test", new HttpServlet() {
@Override
@@ -112,6 +112,17 @@
Thread.sleep(500);
}
+ System.out.println("");
+
+ for (int i = 0; i < received.size(); i++) {
+ for (String property : received.get(i).getPropertyNames()) {
+ System.out.println(property + " = " + received.get(i).getProperty(property));
+ }
+ System.out.println("========");
+ }
+
+ System.out.println("");
+
Assert.assertEquals(1, received.size());
Assert.assertEquals("decanter/collect/soap", received.get(0).getTopic());
diff --git a/itest/src/test/java/org/apache/karaf/decanter/itests/collector/SocketCollectorTest.java b/itest/src/test/java/org/apache/karaf/decanter/itests/collector/SocketCollectorTest.java
index 9eb6892..a850611 100644
--- a/itest/src/test/java/org/apache/karaf/decanter/itests/collector/SocketCollectorTest.java
+++ b/itest/src/test/java/org/apache/karaf/decanter/itests/collector/SocketCollectorTest.java
@@ -31,8 +31,10 @@
import org.osgi.service.event.EventConstants;
import org.osgi.service.event.EventHandler;
+import java.io.IOException;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;
+import java.net.InetAddress;
import java.net.Socket;
import java.util.ArrayList;
import java.util.Hashtable;
@@ -51,13 +53,22 @@
return Stream.of(super.config(), options).flatMap(Stream::of).toArray(Option[]::new);
}
- @Test
+ @Test(timeout = 60000)
public void test() throws Exception {
// install decanter
System.out.println(executeCommand("feature:repo-add decanter " + System.getProperty("decanter.version")));
System.out.println(executeCommand("feature:install decanter-collector-socket", new RolePrincipal("admin")));
- Thread.sleep(2000);
+ System.out.println("Waiting Decanter Collector socket ...");
+ while (true) {
+ try {
+ Thread.sleep(200);
+ Socket socket = new Socket(InetAddress.getLocalHost(), 34343);
+ break;
+ } catch (IOException ioException) {
+ // no-op
+ }
+ }
// create event handler
List<Event> received = new ArrayList();
@@ -80,6 +91,17 @@
Thread.sleep(500);
}
+ System.out.println("");
+
+ for (int i = 0; i < received.size(); i++) {
+ for (String property : received.get(i).getPropertyNames()) {
+ System.out.println(property + " = " + received.get(i).getProperty(property));
+ }
+ System.out.println("========");
+ }
+
+ System.out.println("");
+
Assert.assertEquals(1, received.size());
Assert.assertEquals("decanter/collect/socket", received.get(0).getTopic());
diff --git a/itest/src/test/java/org/apache/karaf/decanter/itests/collector/SystemCollectorTest.java b/itest/src/test/java/org/apache/karaf/decanter/itests/collector/SystemCollectorTest.java
index 25bb15d..8b979bf 100644
--- a/itest/src/test/java/org/apache/karaf/decanter/itests/collector/SystemCollectorTest.java
+++ b/itest/src/test/java/org/apache/karaf/decanter/itests/collector/SystemCollectorTest.java
@@ -80,6 +80,17 @@
Thread.sleep(500);
}
+ System.out.println("");
+
+ for (int i = 0; i < received.size(); i++) {
+ for (String property : received.get(i).getPropertyNames()) {
+ System.out.println(property + " = " + received.get(i).getProperty(property));
+ }
+ System.out.println("=========");
+ }
+
+ System.out.println("");
+
Assert.assertTrue(received.size() >= 1);
Assert.assertEquals("decanter/collect/system/command_df", received.get(0).getTopic());
diff --git a/itest/src/test/java/org/apache/karaf/decanter/itests/processor/AggregateProcessorTest.java b/itest/src/test/java/org/apache/karaf/decanter/itests/processor/AggregateProcessorTest.java
new file mode 100644
index 0000000..243df5e
--- /dev/null
+++ b/itest/src/test/java/org/apache/karaf/decanter/itests/processor/AggregateProcessorTest.java
@@ -0,0 +1,108 @@
+/*
+ * 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.karaf.decanter.itests.processor;
+
+import org.apache.karaf.itests.KarafTestSupport;
+import org.apache.karaf.jaas.boot.principal.RolePrincipal;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.ops4j.pax.exam.Configuration;
+import org.ops4j.pax.exam.Option;
+import org.ops4j.pax.exam.junit.PaxExam;
+import org.ops4j.pax.exam.karaf.options.KarafDistributionOption;
+import org.ops4j.pax.exam.spi.reactors.ExamReactorStrategy;
+import org.ops4j.pax.exam.spi.reactors.PerClass;
+import org.osgi.service.event.Event;
+import org.osgi.service.event.EventAdmin;
+import org.osgi.service.event.EventConstants;
+import org.osgi.service.event.EventHandler;
+
+import java.util.*;
+import java.util.stream.Stream;
+
+@RunWith(PaxExam.class)
+@ExamReactorStrategy(PerClass.class)
+public class AggregateProcessorTest extends KarafTestSupport {
+
+ @Configuration
+ public Option[] config() {
+ Option[] options = new Option[]{
+ KarafDistributionOption.editConfigurationFilePut("etc/system.properties", "decanter.version", System.getProperty("decanter.version"))
+ };
+ return Stream.of(super.config(), options).flatMap(Stream::of).toArray(Option[]::new);
+ }
+
+ @Test(timeout = 120000)
+ public void test() throws Exception {
+ System.out.println("Installing Decanter Processor Aggregate ...");
+ System.out.println(executeCommand("feature:repo-add decanter " + System.getProperty("decanter.version"), new RolePrincipal("admin")));
+ System.out.println(executeCommand("feature:install decanter-processor-aggregate", new RolePrincipal("admin")));
+
+ String configList = executeCommand("config:list '(service.pid=org.apache.karaf.decanter.processor.aggregate)'");
+ while (!configList.contains("service.pid")) {
+ Thread.sleep(500);
+ configList = executeCommand("config:list '(service.pid=org.apache.karaf.decanter.processor.aggregate)'");
+ }
+
+ System.out.println("Adding test event handler ...");
+ final List<Event> received = new ArrayList<>();
+ EventHandler handler = new EventHandler() {
+ @Override
+ public void handleEvent(Event event) {
+ received.add(event);
+ }
+ };
+ Hashtable<String, String> serviceProperties = new Hashtable<>();
+ serviceProperties.put(EventConstants.EVENT_TOPIC, "decanter/process/*");
+ bundleContext.registerService(EventHandler.class, handler, serviceProperties);
+
+ System.out.println("Sending test events ...");
+ EventAdmin dispatcher = getOsgiService(EventAdmin.class);
+ Map<String, Object> data1 = new HashMap<>();
+ data1.put("first", "first");
+ dispatcher.sendEvent(new Event("decanter/collect/test", data1));
+ Map<String, Object> data2 = new HashMap<>();
+ data2.put("second", "second");
+ dispatcher.sendEvent(new Event("decanter/collect/test", data2));
+ Map<String, Object> data3 = new HashMap<>();
+ data3.put("third", "third");
+ dispatcher.sendEvent(new Event("decanter/collect/test", data3));
+
+ System.out.println("Waiting aggregation ...");
+ while(received.size() < 1) {
+ Thread.sleep(500);
+ }
+
+ System.out.println("");
+
+ for (int i = 0; i < received.size(); i++) {
+ for (String property : received.get(i).getPropertyNames()) {
+ System.out.println(property + " = " + received.get(i).getProperty(property));
+ }
+ System.out.println("========");
+ }
+
+ System.out.println("");
+
+ Assert.assertEquals(1, received.size());
+ Assert.assertEquals("first", received.get(0).getProperty("0.first"));
+ Assert.assertEquals("second", received.get(0).getProperty("1.second"));
+ Assert.assertEquals("third", received.get(0).getProperty("2.third"));
+ }
+
+}
diff --git a/itest/src/test/java/org/apache/karaf/decanter/itests/processor/GroupByProcessorTest.java b/itest/src/test/java/org/apache/karaf/decanter/itests/processor/GroupByProcessorTest.java
new file mode 100644
index 0000000..6cdc618
--- /dev/null
+++ b/itest/src/test/java/org/apache/karaf/decanter/itests/processor/GroupByProcessorTest.java
@@ -0,0 +1,118 @@
+/*
+ * 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.karaf.decanter.itests.processor;
+
+import org.apache.karaf.itests.KarafTestSupport;
+import org.apache.karaf.jaas.boot.principal.RolePrincipal;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.ops4j.pax.exam.Configuration;
+import org.ops4j.pax.exam.Option;
+import org.ops4j.pax.exam.junit.PaxExam;
+import org.ops4j.pax.exam.karaf.options.KarafDistributionOption;
+import org.ops4j.pax.exam.spi.reactors.ExamReactorStrategy;
+import org.ops4j.pax.exam.spi.reactors.PerClass;
+import org.osgi.service.event.Event;
+import org.osgi.service.event.EventAdmin;
+import org.osgi.service.event.EventConstants;
+import org.osgi.service.event.EventHandler;
+
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.FileWriter;
+import java.util.*;
+import java.util.stream.Stream;
+
+@RunWith(PaxExam.class)
+@ExamReactorStrategy(PerClass.class)
+public class GroupByProcessorTest extends KarafTestSupport {
+
+ @Configuration
+ public Option[] config() {
+ Option[] options = new Option[]{
+ KarafDistributionOption.editConfigurationFilePut("etc/system.properties", "decanter.version", System.getProperty("decanter.version"))
+ };
+ return Stream.of(super.config(), options).flatMap(Stream::of).toArray(Option[]::new);
+ }
+
+ @Test
+ public void test() throws Exception {
+ System.out.println("Installing Decanter Processor GroupBy ...");
+
+ File file = new File(System.getProperty("karaf.etc"), "org.apache.karaf.decanter.processor.groupby.cfg");
+ try (BufferedWriter writer = new BufferedWriter(new FileWriter(file))) {
+ writer.write("groupBy=foo");
+ }
+
+ String configList = executeCommand("config:list '(service.pid=org.apache.karaf.decanter.processor.groupby)'");
+ while (!configList.contains("service.pid")) {
+ Thread.sleep(500);
+ configList = executeCommand("config:list '(service.pid=org.apache.karaf.decanter.processor.groupby)'");
+ }
+
+ System.out.println(executeCommand("feature:repo-add decanter " + System.getProperty("decanter.version"), new RolePrincipal("admin")));
+ System.out.println(executeCommand("feature:install decanter-processor-groupby", new RolePrincipal("admin")));
+
+ System.out.println("Adding test event handler ...");
+ final List<Event> received = new ArrayList<>();
+ EventHandler handler = new EventHandler() {
+ @Override
+ public void handleEvent(Event event) {
+ received.add(event);
+ }
+ };
+ Hashtable<String, String> serviceProperties = new Hashtable<>();
+ serviceProperties.put(EventConstants.EVENT_TOPIC, "decanter/process/*");
+ bundleContext.registerService(EventHandler.class, handler, serviceProperties);
+
+ System.out.println("Sending test events ...");
+ EventAdmin dispatcher = getOsgiService(EventAdmin.class);
+ Map<String, Object> data1 = new HashMap<>();
+ data1.put("foo", "bar");
+ data1.put("first", "first");
+ dispatcher.sendEvent(new Event("decanter/collect/test", data1));
+ Map<String, Object> data2 = new HashMap<>();
+ data2.put("other", "other");
+ data2.put("second", "second");
+ dispatcher.sendEvent(new Event("decanter/collect/test", data2));
+ Map<String, Object> data3 = new HashMap<>();
+ data3.put("foo", "bar");
+ data3.put("third", "third");
+ dispatcher.sendEvent(new Event("decanter/collect/test", data3));
+
+ System.out.println("Waiting events ...");
+ while (received.size() < 1) {
+ Thread.sleep(500);
+ }
+
+ System.out.println("");
+
+ for (int i = 0; i < received.size(); i++) {
+ for (String property : received.get(i).getPropertyNames()) {
+ System.out.println(property + " = " + received.get(i).getProperty(property));
+ }
+ System.out.println("========");
+ }
+
+ System.out.println("");
+
+ Assert.assertEquals(1, received.size());
+
+ }
+
+}
diff --git a/manual/src/main/asciidoc/user-guide/appenders.adoc b/manual/src/main/asciidoc/user-guide/appenders.adoc
index 6fc28aa..d27a9e0 100644
--- a/manual/src/main/asciidoc/user-guide/appenders.adoc
+++ b/manual/src/main/asciidoc/user-guide/appenders.adoc
@@ -663,7 +663,7 @@
`MetricRegistry`. You can register this `MetricRegistry` in your own application or use a Dropwizard Metrics Reporter
to "push" these metrics to some backend.
-The `decanter-appender-dropwizard` feature provides the Decanter event handler registering the harvested data ino the
+The `decanter-appender-dropwizard` feature provides the Decanter event handler registering the harvested data into the
`MetricRegistry`:
----
diff --git a/manual/src/main/asciidoc/user-guide/processors.adoc b/manual/src/main/asciidoc/user-guide/processors.adoc
index 42bf87e..7696ee0 100644
--- a/manual/src/main/asciidoc/user-guide/processors.adoc
+++ b/manual/src/main/asciidoc/user-guide/processors.adoc
@@ -52,7 +52,7 @@
By default, the "merged" event is sent every minute. You can change this using the `period` configuration.
-You can provisiong `etc/org.apache.karaf.decanter.processor.aggregate.cfg` configuration file with:
+You can provision `etc/org.apache.karaf.decanter.processor.aggregate.cfg` configuration file with:
----
period=120 # this is the period in seconds
@@ -76,4 +76,62 @@
overwrite=true
----
-Then, if a property already exist in the aggregator, its value will be overwritten by the new event value received in the aggregator.
\ No newline at end of file
+Then, if a property already exist in the aggregator, its value will be overwritten by the new event value received in the aggregator.
+
+==== GroupBy
+
+This processor "groups" events containing same properties values during a period.
+
+For instance, you configure the GroupBy processor to group events using `foo` and `bar` properties. Then you receive
+the following three events:
+
+1. first event containing: `{ "foo":"foo","bar":"bar","first":"value1" }`
+2. second event containing: `{ "hello":"world","second":"value2" }`
+3. third event containing: `{ "foo":"foo","bar":"bar","third":"value3"}`
+
+The groupBy processor will create (and send) one event containing:
+
+* if you choose to "flatten" the properties, the event will contain: `{ "foo":"foo", "bar":"bar", "first":"value1","third":"value3" }`
+* if you chosse not to "flatten" the properties, the event will contain: `{ "events":[ { "foo":"foo","bar":"bar","first":"value1" }, { "foo":"foo","bar":"bar","third":"value3" } ] }`
+
+You can install this processor using the `decanter-processor-groupby` feature:
+
+----
+karaf@root()> feature:install decanter-processor-groupby
+----
+
+By default, the "merged" event is sent every minute. You can change this using the `period` configuration.
+
+The GroupBy processor is configured via `etc/org.apache.karaf.decanter.processor.groupby.cfg` configuration file:
+
+----
+#
+# Decanter GroupBy processor
+#
+
+#
+# Destination dispatcher topics where to send the aggregated events
+#
+#target.topics=decanter/process/groupby
+
+#
+# Aggregation period in seconds
+#
+#period=60
+
+#
+# List of grouping properties
+#
+#groupBy=first,second
+
+#
+# If true, grouped events properties are flatten (all properties in the event) aka Map<String,Object>
+# If false, grouped events properties are inner grouped map aka Map<int, Map<String,Object>>
+#
+#flatten=true
+----
+
+* The `target.topics` property defines the list of Decanter topics (separated by `,`) where the resulting events will be sent.
+* The `period` property defines the retention period to accumulate the incoming events
+* The `groupBy` property defines the property names (separated by `,`) as grouping term
+* The `flatten` property defines the way the resulting event will be created. If `true`, all events properties will be store directly (flat) in the resulting event. If `false`, the resulting event will contain an array of properties (from the original grouped events).
\ No newline at end of file
diff --git a/pom.xml b/pom.xml
index b0e758a..b622b6f 100644
--- a/pom.xml
+++ b/pom.xml
@@ -43,7 +43,7 @@
<glassfish-json.version>1.1.4</glassfish-json.version>
<json-api.version>1.1.4</json-api.version>
<kafka.version>2.6.0</kafka.version>
- <karaf.version>4.2.8</karaf.version>
+ <karaf.version>4.2.10</karaf.version>
<kibana.version>3.1.1</kibana.version>
<kibana4.version>4.1.2</kibana4.version>
<kibana6.version>6.1.1</kibana6.version>
@@ -61,6 +61,7 @@
<module>alerting</module>
<module>manual</module>
<module>assembly</module>
+ <module>itest</module>
</modules>
<scm>
diff --git a/processor/groupby/pom.xml b/processor/groupby/pom.xml
new file mode 100644
index 0000000..f652aad
--- /dev/null
+++ b/processor/groupby/pom.xml
@@ -0,0 +1,78 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<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/xsd/maven-4.0.0.xsd">
+
+ <!--
+
+ 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.
+ -->
+
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.karaf.decanter</groupId>
+ <artifactId>processor</artifactId>
+ <version>2.6.0-SNAPSHOT</version>
+ <relativePath>../pom.xml</relativePath>
+ </parent>
+
+ <groupId>org.apache.karaf.decanter.processor</groupId>
+ <artifactId>org.apache.karaf.decanter.processor.groupby</artifactId>
+ <packaging>bundle</packaging>
+ <name>Apache Karaf :: Decanter :: Processor :: GroupBy</name>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>build-helper-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <phase>package</phase>
+ <goals>
+ <goal>attach-artifact</goal>
+ </goals>
+ <configuration>
+ <artifacts>
+ <artifact>
+ <file>src/main/cfg/org.apache.karaf.decanter.processor.groupby.cfg</file>
+ <type>cfg</type>
+ </artifact>
+ </artifacts>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.felix</groupId>
+ <artifactId>maven-bundle-plugin</artifactId>
+ <inherited>true</inherited>
+ <extensions>true</extensions>
+ <configuration>
+ <obrRepository>NONE</obrRepository>
+ <instructions>
+ <Bundle-SymbolicName>${project.artifactId}</Bundle-SymbolicName>
+ <Export-Package>!*</Export-Package>
+ <Import-Package>*</Import-Package>
+ <Private-Package>
+ org.apache.karaf.decanter.processor.groupby
+ </Private-Package>
+ </instructions>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+
+</project>
\ No newline at end of file
diff --git a/processor/groupby/src/main/cfg/org.apache.karaf.decanter.processor.groupby.cfg b/processor/groupby/src/main/cfg/org.apache.karaf.decanter.processor.groupby.cfg
new file mode 100644
index 0000000..26213d5
--- /dev/null
+++ b/processor/groupby/src/main/cfg/org.apache.karaf.decanter.processor.groupby.cfg
@@ -0,0 +1,43 @@
+################################################################################
+#
+# 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.
+#
+################################################################################
+
+#
+# Decanter GroupBy processor
+#
+
+#
+# Destination dispatcher topics where to send the aggregated events
+#
+#target.topics=decanter/process/groupby
+
+#
+# Aggregation period in seconds
+#
+#period=60
+
+#
+# List of grouping properties
+#
+#groupBy=first,second
+
+#
+# If true, grouped events properties are flatten (all properties in the event) aka Map<String,Object>
+# If false, grouped events properties are inner grouped map aka Map<int, Map<String,Object>>
+#
+#flatten=true
\ No newline at end of file
diff --git a/processor/groupby/src/main/java/org/apache/karaf/decanter/processor/groupby/GroupByProcessor.java b/processor/groupby/src/main/java/org/apache/karaf/decanter/processor/groupby/GroupByProcessor.java
new file mode 100644
index 0000000..ad6f8b6
--- /dev/null
+++ b/processor/groupby/src/main/java/org/apache/karaf/decanter/processor/groupby/GroupByProcessor.java
@@ -0,0 +1,133 @@
+/*
+ * 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.karaf.decanter.processor.groupby;
+
+import org.osgi.service.component.ComponentContext;
+import org.osgi.service.component.annotations.Activate;
+import org.osgi.service.component.annotations.Component;
+import org.osgi.service.component.annotations.Deactivate;
+import org.osgi.service.component.annotations.Reference;
+import org.osgi.service.event.Event;
+import org.osgi.service.event.EventAdmin;
+import org.osgi.service.event.EventConstants;
+import org.osgi.service.event.EventHandler;
+
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+@Component(
+ name = "org.apache.karaf.decanter.processor.groupby",
+ immediate = true,
+ property = EventConstants.EVENT_TOPIC + "=decanter/collect/*"
+)
+public class GroupByProcessor implements EventHandler {
+
+ @Reference
+ private EventAdmin dispatcher;
+
+ private String targetTopics;
+ private String groupBy;
+ private boolean flat;
+
+ private ConcurrentHashMap<Integer, List<Event>> accumulation = new ConcurrentHashMap<>();
+
+ private ScheduledExecutorService scheduledExecutorService;
+
+ @Activate
+ public void activate(ComponentContext componentContext) {
+ activate(componentContext.getProperties());
+ }
+
+ public void activate(Dictionary<String, Object> configuration) {
+ targetTopics = (configuration.get("target.topics") != null) ? configuration.get("target.topics").toString() : "decanter/process/aggregate";
+ long period = (configuration.get("period") != null) ? Long.parseLong(configuration.get("period").toString()) : 60L;
+ groupBy = (configuration.get("groupBy") != null) ? configuration.get("groupBy").toString() : null;
+ flat = (configuration.get("flat") != null) ? Boolean.parseBoolean(configuration.get("flat").toString()) : true;
+ scheduledExecutorService = Executors.newScheduledThreadPool(1);
+ scheduledExecutorService.scheduleAtFixedRate(new GroupByTask(), 0, period, TimeUnit.SECONDS);
+ }
+
+ @Deactivate
+ public void deactivate() {
+ scheduledExecutorService.shutdownNow();
+ }
+
+ @Override
+ public void handleEvent(Event event) {
+ String[] groups = groupBy.split(",");
+ int hash = 0;
+ for (String group : groups) {
+ if (event.getProperty(group) == null) {
+ return;
+ } else {
+ hash = hash + event.getProperty(group).hashCode();
+ }
+ }
+ if (accumulation.get(hash) == null) {
+ List<Event> events = new ArrayList<>();
+ events.add(event);
+ accumulation.put(hash, events);
+ } else {
+ accumulation.get(hash).add(event);
+ }
+ }
+
+ class GroupByTask implements Runnable {
+
+ @Override
+ public void run() {
+ if (accumulation.size() > 0) {
+ for (Integer hash : accumulation.keySet()) {
+ Map merge = new HashMap();
+ merge.put("processor", "groupBy");
+ if (flat) {
+ for (Event event : accumulation.get(hash)) {
+ for (String propertyName : event.getPropertyNames()) {
+ merge.put(propertyName, event.getProperty(propertyName));
+ }
+ }
+ } else {
+ List<Map<String, Object>> events = new ArrayList<>();
+ merge.put("events", events);
+ for (Event event : accumulation.get(hash)) {
+ Map<String,Object> properties = new HashMap<>();
+ for (String propertyName : event.getPropertyNames()) {
+ properties.put(propertyName, event.getProperty(propertyName));
+ }
+ events.add(properties);
+ }
+ }
+ // send event
+ String[] topics = targetTopics.split(",");
+ for (String topic : topics) {
+ dispatcher.postEvent(new Event(topic, merge));
+ }
+ }
+ accumulation.clear();
+ }
+ }
+
+ }
+
+ public void setDispatcher(EventAdmin dispatcher) {
+ this.dispatcher = dispatcher;
+ }
+
+}
diff --git a/processor/groupby/src/test/java/org/apache/karaf/decanter/processor/groupby/GroupByProcessorTest.java b/processor/groupby/src/test/java/org/apache/karaf/decanter/processor/groupby/GroupByProcessorTest.java
new file mode 100644
index 0000000..77e5047
--- /dev/null
+++ b/processor/groupby/src/test/java/org/apache/karaf/decanter/processor/groupby/GroupByProcessorTest.java
@@ -0,0 +1,170 @@
+/*
+ * 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.karaf.decanter.processor.groupby;
+
+import org.junit.Assert;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.osgi.service.event.Event;
+import org.osgi.service.event.EventAdmin;
+
+import java.util.*;
+
+public class GroupByProcessorTest {
+
+ @Test(timeout = 10000)
+ public void testFlatten() throws Exception {
+ MockDispatcher dispatcher = new MockDispatcher();
+
+ GroupByProcessor processor = new GroupByProcessor();
+ processor.setDispatcher(dispatcher);
+ Hashtable<String, Object> configuration = new Hashtable<>();
+ configuration.put("period", "2");
+ configuration.put("groupBy", "foo,bar");
+ configuration.put("flat", true);
+ processor.activate(configuration);
+
+ Map<String, Object> data1 = new HashMap<>();
+ data1.put("foo", "foo");
+ data1.put("bar", "bar");
+ data1.put("first", "first");
+ Event event1 = new Event("decanter/collect/first", data1);
+ processor.handleEvent(event1);
+
+ Map<String, Object> data2 = new HashMap<>();
+ data2.put("foo", "foo");
+ data2.put("bar", "bar");
+ data2.put("second", "second");
+ Event event2 = new Event("decanter/collect/second", data2);
+ processor.handleEvent(event2);
+
+ Map<String, Object> data3 = new HashMap<>();
+ data3.put("third", "third");
+ Event event3 = new Event("decanter/collect/third", data3);
+ processor.handleEvent(event3);
+
+ Map<String, Object> data4 = new HashMap<>();
+ data4.put("foo", "foo");
+ data4.put("bar", "bar");
+ data4.put("fourth", "fourth");
+ Event event4 = new Event("decanter/collect/fourth", data4);
+ processor.handleEvent(event4);
+
+ Map<String, Object> data5 = new HashMap<>();
+ data5.put("foo", "other");
+ data5.put("bar", "other");
+ data5.put("fifth", "fifth");
+ Event event5 = new Event("decanter/collect/fifth", data5);
+ processor.handleEvent(event5);
+
+ while (dispatcher.postedEvents.size() != 2) {
+ Thread.sleep(200);
+ }
+
+ Assert.assertEquals(2, dispatcher.postedEvents.size());
+
+ Assert.assertEquals("fifth", dispatcher.postedEvents.get(0).getProperty("fifth"));
+ Assert.assertEquals("other", dispatcher.postedEvents.get(0).getProperty("bar"));
+ Assert.assertEquals("other", dispatcher.postedEvents.get(0).getProperty("foo"));
+
+ Assert.assertEquals("bar", dispatcher.postedEvents.get(1).getProperty("bar"));
+ Assert.assertEquals("fourth", dispatcher.postedEvents.get(1).getProperty("fourth"));
+ Assert.assertEquals("first", dispatcher.postedEvents.get(1).getProperty("first"));
+ Assert.assertEquals("foo", dispatcher.postedEvents.get(1).getProperty("foo"));
+ Assert.assertEquals("second", dispatcher.postedEvents.get(1).getProperty("second"));
+
+ processor.deactivate();
+ }
+
+ @Test(timeout = 10000)
+ @Ignore("Avoid thread conflict with the other test")
+ public void testNotFlatten() throws Exception {
+ MockDispatcher dispatcher = new MockDispatcher();
+
+ GroupByProcessor processor = new GroupByProcessor();
+ processor.setDispatcher(dispatcher);
+ Hashtable<String, Object> configuration = new Hashtable<>();
+ configuration.put("period", "2");
+ configuration.put("groupBy", "foo,bar");
+ configuration.put("flat", false);
+ processor.activate(configuration);
+
+ Map<String, Object> data1 = new HashMap<>();
+ data1.put("foo", "foo");
+ data1.put("bar", "bar");
+ data1.put("first", "first");
+ Event event1 = new Event("decanter/collect/first", data1);
+ processor.handleEvent(event1);
+
+ Map<String, Object> data2 = new HashMap<>();
+ data2.put("foo", "foo");
+ data2.put("bar", "bar");
+ data2.put("second", "second");
+ Event event2 = new Event("decanter/collect/second", data2);
+ processor.handleEvent(event2);
+
+ Map<String, Object> data3 = new HashMap<>();
+ data3.put("third", "third");
+ Event event3 = new Event("decanter/collect/third", data3);
+ processor.handleEvent(event3);
+
+ Map<String, Object> data4 = new HashMap<>();
+ data4.put("foo", "foo");
+ data4.put("bar", "bar");
+ data4.put("fourth", "fourth");
+ Event event4 = new Event("decanter/collect/fourth", data4);
+ processor.handleEvent(event4);
+
+ Map<String, Object> data5 = new HashMap<>();
+ data5.put("foo", "other");
+ data5.put("bar", "other");
+ data5.put("fifth", "fifth");
+ Event event5 = new Event("decanter/collect/fifth", data5);
+ processor.handleEvent(event5);
+
+ while (dispatcher.postedEvents.size() != 2) {
+ Thread.sleep(200);
+ }
+
+ Assert.assertEquals(2, dispatcher.postedEvents.size());
+
+ List<Map<String, Object>> events = (List<Map<String, Object>>) dispatcher.postedEvents.get(0).getProperty("events");
+ Assert.assertEquals(1, events.size());
+
+ events = (List<Map<String, Object>>) dispatcher.postedEvents.get(1).getProperty("events");
+ Assert.assertEquals(3, events.size());
+
+ processor.deactivate();
+ }
+
+ class MockDispatcher implements EventAdmin {
+
+ public List<Event> postedEvents = new ArrayList<>();
+ public List<Event> sentEvents = new ArrayList<>();
+
+ @Override
+ public void postEvent(Event event) {
+ postedEvents.add(event);
+ }
+
+ @Override
+ public void sendEvent(Event event) {
+ sentEvents.add(event);
+ }
+ }
+
+}
diff --git a/processor/pom.xml b/processor/pom.xml
index a0749ea..dd95953 100644
--- a/processor/pom.xml
+++ b/processor/pom.xml
@@ -36,6 +36,7 @@
<modules>
<module>passthrough</module>
<module>aggregate</module>
+ <module>groupby</module>
</modules>
</project>
\ No newline at end of file
