You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@sling.apache.org by tm...@apache.org on 2019/04/18 15:26:45 UTC
[sling-org-apache-sling-distribution-journal-messages] branch
master updated: SLING-8360 - Move Journal messaging API to messages bundle
This is an automated email from the ASF dual-hosted git repository.
tmaret pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/sling-org-apache-sling-distribution-journal-messages.git
The following commit(s) were added to refs/heads/master by this push:
new 6830b03 SLING-8360 - Move Journal messaging API to messages bundle
6830b03 is described below
commit 6830b0388a62d4294c3890d6764d23e6302bf3b9
Author: tmaret <tm...@adobe.com>
AuthorDate: Thu Apr 18 16:37:30 2019 +0200
SLING-8360 - Move Journal messaging API to messages bundle
---
pom.xml | 25 +++++++--
.../sling/distribution/journal/FullMessage.java | 41 ++++++++++++++
.../sling/distribution/journal/HandlerAdapter.java | 64 ++++++++++++++++++++++
.../distribution/journal/JournalAvailable.java | 26 +++++++++
.../distribution/journal/JsonMessageSender.java | 25 +++++++++
.../sling/distribution/journal/MessageHandler.java | 23 ++++++++
.../sling/distribution/journal/MessageInfo.java | 30 ++++++++++
.../sling/distribution/journal/MessageSender.java | 27 +++++++++
.../distribution/journal/MessagingException.java | 33 +++++++++++
.../distribution/journal/MessagingProvider.java | 45 +++++++++++++++
.../apache/sling/distribution/journal/Reset.java | 21 +++++++
.../sling/distribution/journal/RunnableUtil.java | 37 +++++++++++++
.../sling/distribution/journal/package-info.java | 21 +++++++
13 files changed, 414 insertions(+), 4 deletions(-)
diff --git a/pom.xml b/pom.xml
index 39b36f0..2ffad79 100644
--- a/pom.xml
+++ b/pom.xml
@@ -55,10 +55,8 @@
<build>
<plugins>
<plugin>
- <groupId>org.apache.felix</groupId>
- <artifactId>maven-bundle-plugin</artifactId>
- <version>3.5.0</version>
- <extensions>true</extensions>
+ <groupId>biz.aQute.bnd</groupId>
+ <artifactId>bnd-maven-plugin</artifactId>
</plugin>
<plugin>
<groupId>com.github.os72</groupId>
@@ -96,6 +94,19 @@
<version>3.7.0</version>
</dependency>
+ <dependency>
+ <groupId>org.osgi</groupId>
+ <artifactId>osgi.annotation</artifactId>
+ <version>7.0.0</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>com.google.code.findbugs</groupId>
+ <artifactId>jsr305</artifactId>
+ <version>2.0.0</version>
+ </dependency>
+
<!-- Test -->
<dependency>
<groupId>junit</groupId>
@@ -108,5 +119,11 @@
<version>1.3_1</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>ch.qos.logback</groupId>
+ <artifactId>logback-classic</artifactId>
+ <version>1.2.3</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
diff --git a/src/main/java/org/apache/sling/distribution/journal/FullMessage.java b/src/main/java/org/apache/sling/distribution/journal/FullMessage.java
new file mode 100644
index 0000000..444c209
--- /dev/null
+++ b/src/main/java/org/apache/sling/distribution/journal/FullMessage.java
@@ -0,0 +1,41 @@
+/*
+ * 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.distribution.journal;
+
+import com.google.protobuf.GeneratedMessage;
+
+public class FullMessage<T extends GeneratedMessage> {
+
+ private MessageInfo info;
+ private T message;
+
+ public FullMessage(MessageInfo info, T message) {
+ this.info = info;
+ this.message = message;
+ }
+
+ public MessageInfo getInfo() {
+ return info;
+ }
+
+ public T getMessage() {
+ return message;
+ }
+
+}
diff --git a/src/main/java/org/apache/sling/distribution/journal/HandlerAdapter.java b/src/main/java/org/apache/sling/distribution/journal/HandlerAdapter.java
new file mode 100644
index 0000000..1b59c78
--- /dev/null
+++ b/src/main/java/org/apache/sling/distribution/journal/HandlerAdapter.java
@@ -0,0 +1,64 @@
+/*
+ * 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.distribution.journal;
+
+import java.lang.reflect.Method;
+
+import com.google.protobuf.ByteString;
+import com.google.protobuf.ExtensionRegistryLite;
+
+public class HandlerAdapter<T> {
+ private final MessageHandler<T> handler;
+ private final Method method;
+ private final ExtensionRegistryLite registry;
+ private final Class<T> type;
+
+ public HandlerAdapter(Class<T> type, MessageHandler<T> handler) {
+ this.type = type;
+ this.handler = handler;
+ try {
+ method = type.getMethod("parseFrom", ByteString.class, ExtensionRegistryLite.class);
+ } catch (Exception e) {
+ throw new IllegalArgumentException(e);
+ }
+ registry = ExtensionRegistryLite.newInstance();
+ }
+
+ public static <T> HandlerAdapter<T> create(Class<T> type, MessageHandler<T> handler) {
+ return new HandlerAdapter<>(type, handler);
+ }
+
+ @SuppressWarnings("unchecked")
+ private T parseFrom(ByteString payloadBytes) throws Exception {
+ return (T) method.invoke(null, payloadBytes, registry);
+ }
+
+ public void handle(MessageInfo info, ByteString payloadBytes) throws Exception {
+ T payload = parseFrom(payloadBytes);
+ handler.handle(info, payload);
+ }
+
+ public Class<?> getType() {
+ return type;
+ }
+
+ public MessageHandler<T> getHandler() {
+ return this.handler;
+ }
+}
diff --git a/src/main/java/org/apache/sling/distribution/journal/JournalAvailable.java b/src/main/java/org/apache/sling/distribution/journal/JournalAvailable.java
new file mode 100644
index 0000000..5c06954
--- /dev/null
+++ b/src/main/java/org/apache/sling/distribution/journal/JournalAvailable.java
@@ -0,0 +1,26 @@
+/*
+ * 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.distribution.journal;
+
+/**
+ * Marker service to signal that the Journal service is available and correctly configured
+ */
+public interface JournalAvailable {
+
+}
diff --git a/src/main/java/org/apache/sling/distribution/journal/JsonMessageSender.java b/src/main/java/org/apache/sling/distribution/journal/JsonMessageSender.java
new file mode 100644
index 0000000..82a4813
--- /dev/null
+++ b/src/main/java/org/apache/sling/distribution/journal/JsonMessageSender.java
@@ -0,0 +1,25 @@
+/*
+ * 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.distribution.journal;
+
+public interface JsonMessageSender<T> {
+
+ void send(String topic, T payload);
+
+}
\ No newline at end of file
diff --git a/src/main/java/org/apache/sling/distribution/journal/MessageHandler.java b/src/main/java/org/apache/sling/distribution/journal/MessageHandler.java
new file mode 100644
index 0000000..517a8fb
--- /dev/null
+++ b/src/main/java/org/apache/sling/distribution/journal/MessageHandler.java
@@ -0,0 +1,23 @@
+/*
+ * 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.distribution.journal;
+
+public interface MessageHandler<T> {
+ void handle(MessageInfo info, T message);
+}
diff --git a/src/main/java/org/apache/sling/distribution/journal/MessageInfo.java b/src/main/java/org/apache/sling/distribution/journal/MessageInfo.java
new file mode 100644
index 0000000..702682d
--- /dev/null
+++ b/src/main/java/org/apache/sling/distribution/journal/MessageInfo.java
@@ -0,0 +1,30 @@
+/*
+ * 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.distribution.journal;
+
+public interface MessageInfo {
+
+ String getTopic();
+
+ int getPartition();
+
+ long getOffset();
+
+ long getCreateTime();
+}
diff --git a/src/main/java/org/apache/sling/distribution/journal/MessageSender.java b/src/main/java/org/apache/sling/distribution/journal/MessageSender.java
new file mode 100644
index 0000000..83a9eaa
--- /dev/null
+++ b/src/main/java/org/apache/sling/distribution/journal/MessageSender.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.
+ */
+package org.apache.sling.distribution.journal;
+
+import com.google.protobuf.GeneratedMessage;
+
+public interface MessageSender<T extends GeneratedMessage> {
+
+ void send(String topic, T payload) throws MessagingException;
+
+}
\ No newline at end of file
diff --git a/src/main/java/org/apache/sling/distribution/journal/MessagingException.java b/src/main/java/org/apache/sling/distribution/journal/MessagingException.java
new file mode 100644
index 0000000..7a4855d
--- /dev/null
+++ b/src/main/java/org/apache/sling/distribution/journal/MessagingException.java
@@ -0,0 +1,33 @@
+/*
+ * 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.distribution.journal;
+
+public class MessagingException extends RuntimeException {
+
+ private static final long serialVersionUID = 2381842058778043644L;
+
+ public MessagingException(String message) {
+ super(message);
+ }
+
+ public MessagingException(String message, Exception cause) {
+ super(message, cause);
+ }
+
+}
diff --git a/src/main/java/org/apache/sling/distribution/journal/MessagingProvider.java b/src/main/java/org/apache/sling/distribution/journal/MessagingProvider.java
new file mode 100644
index 0000000..c73dfb1
--- /dev/null
+++ b/src/main/java/org/apache/sling/distribution/journal/MessagingProvider.java
@@ -0,0 +1,45 @@
+/*
+ * 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.distribution.journal;
+
+import java.io.Closeable;
+
+import com.google.protobuf.GeneratedMessage;
+
+public interface MessagingProvider {
+
+ <T extends GeneratedMessage> MessageSender<T> createSender();
+
+ <T> Closeable createPoller(String topicName, Reset reset, HandlerAdapter<?>... adapters);
+
+ Closeable createPoller(String topicName, Reset reset, String assign,
+ HandlerAdapter<?>... adapters);
+
+ <T> JsonMessageSender<T> createJsonSender();
+
+ <T> Closeable createJsonPoller(String topicName, Reset reset, MessageHandler<T> handler,
+ Class<T> type);
+
+ void assertTopic(String topic) throws MessagingException;
+
+ long retrieveOffset(String topicName, Reset reset);
+
+ String assignTo(long offset);
+
+}
\ No newline at end of file
diff --git a/src/main/java/org/apache/sling/distribution/journal/Reset.java b/src/main/java/org/apache/sling/distribution/journal/Reset.java
new file mode 100644
index 0000000..283a7d4
--- /dev/null
+++ b/src/main/java/org/apache/sling/distribution/journal/Reset.java
@@ -0,0 +1,21 @@
+/*
+ * 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.distribution.journal;
+
+public enum Reset { earliest, latest }
\ No newline at end of file
diff --git a/src/main/java/org/apache/sling/distribution/journal/RunnableUtil.java b/src/main/java/org/apache/sling/distribution/journal/RunnableUtil.java
new file mode 100644
index 0000000..10b04ec
--- /dev/null
+++ b/src/main/java/org/apache/sling/distribution/journal/RunnableUtil.java
@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.sling.distribution.journal;
+
+import javax.annotation.ParametersAreNonnullByDefault;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@ParametersAreNonnullByDefault
+public class RunnableUtil {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RunnableUtil.class);
+
+ public static Thread startBackgroundThread(Runnable runnable, String threadName) {
+ Thread thread = new Thread(runnable, threadName);
+ thread.setDaemon(true);
+ thread.start();
+ return thread;
+ }
+}
diff --git a/src/main/java/org/apache/sling/distribution/journal/package-info.java b/src/main/java/org/apache/sling/distribution/journal/package-info.java
new file mode 100644
index 0000000..82ea548
--- /dev/null
+++ b/src/main/java/org/apache/sling/distribution/journal/package-info.java
@@ -0,0 +1,21 @@
+/*
+ * 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.
+ */
+@org.osgi.annotation.versioning.Version("0.0.1")
+package org.apache.sling.distribution.journal;
+