SLING-5646 MoM API and JMS implementation with example usage by Jobs implementation.
Squashes 27 commits from https://github.com/ieb/sling/tree/jobs_28 as follows.
Added first stab at a message oriented job subsystem
Added basic implementation for the manager keeping queue implementation details abstracted
Added ActiveMQ implementation for queues and topics, and fixed a number of the the SPI interfaces in the process
Basic Test coverate for OOT JMS Broker
Extracted a MOM API with no Jobs or JMS references, test coverage for ActiveMQ impl is 100% class, 93% method, 75% line
Added missing license headers, documentation and cleaned out unused interfaces
Fixed JMS Transaction issue found by @tmaret
Coverage for the majority of the jobs code is complete
Basic unit test coverage complete, core has 94% by lines, AMQ 74%, 100% classes and methods
Added testing environment for a runnign server, Not working yet
Added ability for detect when the OSGi container has completed bundle startup without having to perform http requests
Start at IT testing with Crankstart
Fixed issues with shutdown inside a Crankstart container
Working Crankstart IT framework
Version that uses Q->Jobs->JobConsumer pattern
Added a Queue Factory to allow configuration of multiple queues between the MOM API and Job Subsystem and move JobConsumers to register with a Job type
Migrated Subscribers and QueueReaders to a OSGi whiteboard pattern after discussion on Sling Dev
Changes the JobConsumer to use a Callback rather than return a job. This was suggested offlist by others in Adobe as a way of improving resource consumption
Added Types to improve type safety in certain areas after suggestions offlist
Fixed issue with OSGi startup in IntelliJ caused by version 4 of the
Felix framework bundle being present inside the maven pom. Strangely a
command line build was not impacted.
Added integration test bundle to test service. Adjusted some of the APIs to make using the Job Sub System easier
Integration tests now starting jobs from messages
Fixed Startup to work in real Sling/AEM container. The Active MQ OSGi bundle contains additional dependencies that cause all sorts of problems, AMQ is now being embedded into the AMQ MOM Impl bundle.
Fixed Queue expriy bug and added AEM Fiddle to run jobs
Added Documentation for configuration and default sample configuration
Added Explicit requeue mechanims rather than relying on AMQ's requeue capabilities
Moved MoM to new Home
git-svn-id: https://svn.apache.org/repos/asf/sling/trunk@1754255 13f79535-47bb-0310-9956-ffa450edef68
diff --git a/README.md b/README.md
new file mode 100644
index 0000000..311c22b
--- /dev/null
+++ b/README.md
@@ -0,0 +1,62 @@
+# Message Oriented Middleware API ( or message passing API).
+
+This bundle contains an API to support 2 types of message passing. Pub/Sub and Queue.
+Pub/Sub message passing supports the publication of messages to multiple subscribers. Queue
+support guaranteed delivery of a message to one and only one receiver. Pub/Sub messages are
+organised by a Topic name. Queues are named. There are no implementation details in this
+bundle and no bindings to any one MoM implementation, although it is expected that this
+API could be implemented over either JMS or a AMQP client.
+
+# Publish Subscribe
+
+Messages that are sent to a topic by publishing are received by all subscribers that were active at the time the message is published.
+
+To publish a message use the [TopicManager API](src/main/java/org/apache/sling/mom/TopicManager.java)
+
+ topicManager.publish(String topic, Map<String, Object> message);
+
+ where:
+ topic is the name of the topic to which the message is published
+ message is the message as a Map or Maps.
+
+To subscribe to a topic the caller must also use the [TopicManager API](src/main/java/org/apache/sling/mom/TopicManager.java)
+
+ Subscription subscription = subscribe(Subscriber subscriber, String[] topicNames,MessageFilter filter);
+
+ where:
+ subscription is a Subsctipion objects which must be disposed (call dispose()) when the subscription ends.
+ topicNames is an array of topic names.
+ messageFilter is a MessageFilter implementation that accepts only those messages the subscriber is interested in.
+
+The API does not impose any stcuture on topic names, but the underlying implementation might.
+
+# Queue
+
+A Queue implementation guarantees that messages will be delivered to one and only once QueueReader
+in the order in which the messages were added to the Queue. The QueueReader implementation may
+request to re-queue messages. The implementation should retry requeued messages after a suitable delay.
+
+To add a message to a named queue use the [QueueManager API](src/main/java/org/apache/sling/mom/QueueManager.java)
+
+
+ queueManager.add(String name, Map<String, Object> message);
+
+ where:
+ name is the name of the queue.
+ message is the message in map of maps form.
+
+To recieve messages from the queue ise the [QueueManager API](src/main/java/org/apache/sling/mom/QueueManager.java)
+
+ QueueSession queueSession = queueManager.open( QueueReader queueReader, String name, MessageFilter messageFilter);
+
+ where:
+ queueSession is a QueueSession instance that must be closed (call close()) when the the queue reader requires no
+ more messages,
+ queueReader is a QueueReader implemetnation that will get delivered messages from the queue.
+ name is the name of the queueu.
+ messageFilter is a message filter intended to accept messages.
+
+The QueueManager implementation will deliver messages to the queueReader.onMessage(...) method. The implementation of the QueueReader.onMessage method
+may process the message, returning normally, or request the message is requeued by throwing a RequeueMessageExption.
+
+
diff --git a/pom.xml b/pom.xml
new file mode 100644
index 0000000..6bdb924
--- /dev/null
+++ b/pom.xml
@@ -0,0 +1,92 @@
+<?xml version="1.0" encoding="ISO-8859-1"?>
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.sling</groupId>
+ <artifactId>sling</artifactId>
+ <version>26</version>
+ <relativePath />
+ </parent>
+
+ <artifactId>org.apache.sling.mom</artifactId>
+ <packaging>bundle</packaging>
+ <version>0.0.1-SNAPSHOT</version>
+
+ <name>Apache Sling Message oriented Middleware API</name>
+ <description>
+ An API to support message passing using queues or publish/subscribe patterns.
+ </description>
+
+ <scm>
+ <connection>scm:svn:http://svn.apache.org/repos/asf/sling/trunk/contrib/commons/mom/api</connection>
+ <developerConnection>scm:svn:https://svn.apache.org/repos/asf/sling/trunk/contrib/commons/mom/api</developerConnection>
+ <url>http://svn.apache.org/viewvc/sling/trunk/contrib/commons/mom/api</url>
+ </scm>
+
+ <properties>
+ <site.jira.version.id>12315369</site.jira.version.id>
+ <sling.java.version>7</sling.java.version>
+ <exam.version>4.4.0</exam.version>
+ <url.version>2.4.5</url.version>
+ <bundle.build.dir>${basedir}/target</bundle.build.dir>
+ <bundle.file.name>${bundle.build.dir}/${project.build.finalName}.jar</bundle.file.name>
+ <min.port>37000</min.port>
+ <max.port>37999</max.port>
+ </properties>
+
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.felix</groupId>
+ <artifactId>maven-scr-plugin</artifactId>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.felix</groupId>
+ <artifactId>maven-bundle-plugin</artifactId>
+ <extensions>true</extensions>
+ </plugin>
+ </plugins>
+ </build>
+ <reporting>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-javadoc-plugin</artifactId>
+ <configuration>
+ <excludePackageNames>
+ </excludePackageNames>
+ </configuration>
+ </plugin>
+ </plugins>
+ </reporting>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.felix</groupId>
+ <artifactId>org.apache.felix.scr.annotations</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.google.code.findbugs</groupId>
+ <artifactId>jsr305</artifactId>
+ <version>2.0.1</version>
+ </dependency>
+ </dependencies>
+</project>
diff --git a/src/main/java/org/apache/sling/mom/MessageFilter.java b/src/main/java/org/apache/sling/mom/MessageFilter.java
new file mode 100644
index 0000000..da52294
--- /dev/null
+++ b/src/main/java/org/apache/sling/mom/MessageFilter.java
@@ -0,0 +1,44 @@
+/*
+ * 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.sling.mom;
+
+import java.util.Map;
+
+/**
+ * Created by ieb on 30/03/2016.
+ * Filter inbound messages, optionally implemented by QueueReaders.
+ */
+public interface MessageFilter {
+ /**
+ * Provides message filtering, the implementation should return true if it wants to get the message, false if not.
+ * It should make these checks as fast as possible with minimal overhead to avoid consuming resources. Do not implement
+ * this method to process the message. Implementation code calling this method will be very latency sensitive
+ * and subscriptions using slow implementations may get unsubscribed.
+ * @param name the name of the queue or topic the message was sent on.
+ * @param mapMessage the message content.
+ * @return true if the message is to be allowed through the filter.
+ */
+ boolean accept(Types.Name name, Map<String, Object> mapMessage);
+
+
+
+
+
+}
diff --git a/src/main/java/org/apache/sling/mom/QueueManager.java b/src/main/java/org/apache/sling/mom/QueueManager.java
new file mode 100644
index 0000000..2db6ded
--- /dev/null
+++ b/src/main/java/org/apache/sling/mom/QueueManager.java
@@ -0,0 +1,39 @@
+/*
+ * 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.sling.mom;
+
+import java.util.Map;
+
+/**
+ * Created by ieb on 31/03/2016.
+ * Manages named queues allowing messages to be added to the queue and a queue reader to be opened to read messages from a queue.
+ */
+public interface QueueManager {
+
+ /**
+ * Add the message to a queue identified by name.
+ * @param name the name of the queue.
+ * @param message the message to post to the queue.
+ */
+ void add(Types.QueueName name, Map<String, Object> message);
+
+
+
+}
diff --git a/src/main/java/org/apache/sling/mom/QueueReader.java b/src/main/java/org/apache/sling/mom/QueueReader.java
new file mode 100644
index 0000000..d8db29c
--- /dev/null
+++ b/src/main/java/org/apache/sling/mom/QueueReader.java
@@ -0,0 +1,51 @@
+/*
+ * 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.sling.mom;
+
+import java.util.Map;
+
+/**
+ * Created by ieb on 30/03/2016.
+ * A queue reader receives messages from the queue in the onMessage method. It should avoid performing processing in
+ * the onMessage method aiming to return as fast as possible.
+ *
+ * This interface should be implemented as an OSGi Service. The implementation of the MoM API should register any services
+ * implementing QueueReader using the OSGi Whiteboard pattern.
+ */
+public interface QueueReader {
+
+
+ /**
+ * Configuration property name for QueueReaders implemented using a whiteboard pattern.
+ */
+ String QUEUE_NAME_PROP = "queue-name";
+
+ /**
+ * Receive messages from the queue. If the message cant be processed at this time a RequeueMessageException must be thrown to cause the message
+ * to be requeued. Any other exception will cause the message to fail without a retry. Messages that fail are dropped and not notified except the logs.
+ * Implementors should avoid failing any message without a retry.
+ * @param queueName the name of the queue
+ * @param message the message
+ * @throws RequeueMessageException when the message must be re-queued.
+ */
+ void onMessage(Types.QueueName queueName, Map<String, Object> message) throws RequeueMessageException;
+
+
+}
diff --git a/src/main/java/org/apache/sling/mom/RequeueMessageException.java b/src/main/java/org/apache/sling/mom/RequeueMessageException.java
new file mode 100644
index 0000000..978b6b1
--- /dev/null
+++ b/src/main/java/org/apache/sling/mom/RequeueMessageException.java
@@ -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.
+ */
+package org.apache.sling.mom;
+
+/**
+ * Created by ieb on 02/04/2016.
+ * Thown when a queue reader cant process a message and needs to indicate that the message sould be re-queued and retried some time later.
+ */
+public class RequeueMessageException extends Exception {
+
+ /**
+ *
+ * @param message
+ */
+ public RequeueMessageException(String message) {
+ super(message);
+ }
+
+ /**
+ *
+ * @param message
+ * @param cause
+ */
+ public RequeueMessageException(String message, Throwable cause) {
+ super(message, cause);
+ }
+}
diff --git a/src/main/java/org/apache/sling/mom/Subscriber.java b/src/main/java/org/apache/sling/mom/Subscriber.java
new file mode 100644
index 0000000..abb906e
--- /dev/null
+++ b/src/main/java/org/apache/sling/mom/Subscriber.java
@@ -0,0 +1,46 @@
+/*
+ * 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.sling.mom;
+
+import java.util.Map;
+
+/**
+ * Created by ieb on 30/03/2016.
+ *
+ * To implement a topic subscriber implement this interface, register it as an OSGi service and the TopicManager
+ * which will implement a OSGi Whiteboard pattern will register it based on the values in the OSGi property "topics".
+ * The component may optionally implement MessageFilter if it wants to separate filtering messages sooner.
+ */
+public interface Subscriber {
+
+ /**
+ * This is a String[] OSGi property containing the topic names this subscriber should subscribe to.
+ */
+ String TOPIC_NAMES_PROP = "topics";
+
+ /**
+ * Will be called with each message matching the filters the TopicListener is registered with.
+ *
+ * @param topic
+ * @param message
+ */
+ void onMessage(Types.TopicName topic, Map<String, Object> message);
+
+
+}
diff --git a/src/main/java/org/apache/sling/mom/TopicManager.java b/src/main/java/org/apache/sling/mom/TopicManager.java
new file mode 100644
index 0000000..357d865
--- /dev/null
+++ b/src/main/java/org/apache/sling/mom/TopicManager.java
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.sling.mom;
+
+
+import javax.annotation.Nonnull;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Created by ieb on 30/03/2016.
+ * Manages Topics allowing callers to publish messages onto a Topic and Subscribe to a topic.
+ *
+ * To create a subscriber implement the Subscriber interface and the implementation of TopicManager should
+ * implement the OSGi whiteboard pattern.
+ */
+public interface TopicManager {
+
+
+
+ /**
+ * Publish a message to a topic with a command name.
+ * @param name the name
+ * @param commandName the command name
+ * @param message the message
+ */
+ void publish(Types.TopicName name, Types.CommandName commandName, Map<String, Object> message);
+
+
+
+}
diff --git a/src/main/java/org/apache/sling/mom/Types.java b/src/main/java/org/apache/sling/mom/Types.java
new file mode 100644
index 0000000..3d7358d
--- /dev/null
+++ b/src/main/java/org/apache/sling/mom/Types.java
@@ -0,0 +1,100 @@
+/*
+ * 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.sling.mom;
+
+/**
+ * Created by ieb on 13/04/2016.
+ */
+public final class Types {
+ private Types() {
+ }
+
+ public interface Name {
+ }
+ public interface TopicName extends Name {
+ }
+ public interface QueueName extends Name {
+ }
+ public interface CommandName {
+ }
+
+ public static TopicName topicName(String topicName) {
+ return new TopicNameImpl(topicName);
+ }
+
+ public static QueueName queueName(String queueName) {
+ return new QueueNameImpl(queueName);
+ }
+
+ public static CommandName commandName(String commandName) {
+ return new CommandNameImpl(commandName);
+ }
+
+
+ private static class QueueNameImpl extends StringWrapper implements QueueName {
+
+ private QueueNameImpl(String value) {
+ super(value);
+ }
+ }
+
+ private static class CommandNameImpl extends StringWrapper implements CommandName {
+
+ private CommandNameImpl(String value) {
+ super(value);
+ }
+ }
+
+ private static class TopicNameImpl extends StringWrapper implements TopicName {
+
+ private TopicNameImpl(String value) {
+ super(value);
+ }
+ }
+
+ private static class StringWrapper implements Comparable<String> {
+
+
+ private String value;
+
+ private StringWrapper(String value) {
+ this.value = value;
+ }
+
+ @Override
+ public int hashCode() {
+ return value.hashCode();
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ return value.equals(obj.toString());
+ }
+
+ @Override
+ public int compareTo(String o) {
+ return value.compareTo(o);
+ }
+
+ public String toString() {
+ return value;
+ }
+ }
+
+}
diff --git a/src/main/java/org/apache/sling/mom/package-info.java b/src/main/java/org/apache/sling/mom/package-info.java
new file mode 100644
index 0000000..d4d7a37
--- /dev/null
+++ b/src/main/java/org/apache/sling/mom/package-info.java
@@ -0,0 +1,27 @@
+/*
+ * 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.
+ */
+
+/**
+ * Created by ieb on 30/03/2016.
+ */
+
+@Version("1.0.0")
+package org.apache.sling.mom;
+
+import aQute.bnd.annotation.Version;
\ No newline at end of file
