Issue eclipse-ditto#878: Merge branch 'master' into feature/pubsub-group

Signed-off-by: Yufei Cai <yufei.cai@bosch.io>

Conflicts:
	services/connectivity/messaging/src/test/java/org/eclipse/ditto/services/connectivity/messaging/persistence/ConnectionPersistenceActorTest.java

bom/pom.xml

@@ -56,6 +56,7 @@
         <amqp-client.version>5.7.3</amqp-client.version>
         <reactive-streams.version>1.0.3</reactive-streams.version>
         <netty.version>4.1.50.Final</netty.version>
+        <cloundevents.version>2.0.0.RC1</cloundevents.version>
 
         <slf4j.version>1.7.28</slf4j.version>
         <logback.version>1.2.3</logback.version>
@@ -97,7 +98,7 @@
         <jmh.version>1.21</jmh.version>
 
         <scalatest.version>3.1.2</scalatest.version>
-        <flapdoodle.version>3.0.0</flapdoodle.version>
+        <docker-java.version>3.2.6</docker-java.version>
         <system-rules.version>1.19.0</system-rules.version>
     </properties>
 
@@ -326,6 +327,12 @@
                 <version>${jjwt.version}</version>
             </dependency>
 
+            <dependency>
+                <groupId>io.cloudevents</groupId>
+                <artifactId>cloudevents-http-basic</artifactId>
+                <version>${cloundevents.version}</version>
+            </dependency>
+
             <!-- ### Indirect "runtime" dependencies we want to pin to a common version -->
             <dependency>
                 <groupId>org.scala-lang</groupId>
@@ -1001,14 +1008,18 @@
                 <version>${akka-persistence-inmemory.version}</version>
                 <scope>test</scope>
             </dependency>
-
             <dependency>
-                <groupId>de.flapdoodle.embed</groupId>
-                <artifactId>de.flapdoodle.embed.mongo</artifactId>
-                <version>${flapdoodle.version}</version>
+                <groupId>com.github.docker-java</groupId>
+                <artifactId>docker-java-core</artifactId>
+                <version>${docker-java.version}</version>
+                <scope>test</scope>
+            </dependency>
+            <dependency>
+                <groupId>com.github.docker-java</groupId>
+                <artifactId>docker-java-transport-zerodep</artifactId>
+                <version>${docker-java.version}</version>
                 <scope>test</scope>
             </dependency>
-
             <dependency>
                 <groupId>org.openjdk.jmh</groupId>
                 <artifactId>jmh-core</artifactId>

deployment/docker/docker-compose.yml

@@ -12,7 +12,7 @@ version: '2.4'
 
 services:
   mongodb:
-    image: docker.io/mongo:3.6
+    image: docker.io/mongo:4.2
     networks:
       default:
         aliases:

deployment/docker/sandbox/docker-compose.yml

@@ -12,7 +12,7 @@ version: '2.4'
 
 services:
   mongodb:
-    image: docker.io/mongo:3.6
+    image: docker.io/mongo:4.2
     networks:
       default:
         aliases:

deployment/kubernetes/mongodb/mongodb.yaml

@@ -23,7 +23,7 @@ spec:
     spec:
       containers:
         - name: mongodb
-          image: docker.io/mongo:3.6
+          image: docker.io/mongo:4.2
           command:
             - mongod
             - --storageEngine

documentation/src/main/resources/_data/sidebars/ditto_sidebar.yml

@@ -251,6 +251,9 @@ entries:
           - title: WebSocket protocol binding
             url: /httpapi-protocol-bindings-websocket.html
             output: web
+          - title: Cloud Events HTTP protocol binding
+            url: /httpapi-protocol-bindings-cloudevents.html
+            output: web
           - title: Server sent events
             url: /httpapi-sse.html
             output: web

documentation/src/main/resources/pages/ditto/architecture-services-gateway.md

@@ -27,5 +27,6 @@ The gateway service has no persistence by its own.
 * translate [Ditto Protocol](protocol-overview.html) messages incoming via the [WebSocket](httpapi-protocol-bindings-websocket.html)
   to [commands](basic-signals-command.html) and translates [command responses](basic-signals-commandresponse.html) back
   to [Ditto Protocol](protocol-overview.html) response messages
+* accepts [Ditto Protocol](protocol-overview.html) messages incoming via the [Cloud Events HTTP Binding](httpapi-protocol-bindings-cloudevents.html)
 * subscribe for [events](basic-signals-event.html) in Ditto cluster and emits [change notifications](basic-changenotifications.html)
   via connected [WebSocket](httpapi-protocol-bindings-websocket.html) clients or via [SSEs](httpapi-sse.html)

documentation/src/main/resources/pages/ditto/connectivity-overview.md

@@ -24,3 +24,4 @@ The following connection types are supported:
 * [MQTT 5](connectivity-protocol-bindings-mqtt5.html)
 * [HTTP 1.1](connectivity-protocol-bindings-http.html)
 * [Kafka 2.x](connectivity-protocol-bindings-kafka2.html)
+

documentation/src/main/resources/pages/ditto/httpapi-protocol-bindings-cloudevents.md

@@ -0,0 +1,105 @@
+---
+title: Cloud Events HTTP protocol binding
+keywords: binding, protocol, http, cloudevents
+tags: [binding, protocol, http]
+permalink: httpapi-protocol-bindings-cloudevents.html
+---
+
+Implements the [HTTP Protocol Binding for CloudEvents - Version 1.0](https://github.com/cloudevents/spec/blob/v1.0/http-protocol-binding.md).
+
+Unless mentioned otherwise, the endpoint following the Cloud Events specification for the HTTP binding in version 1.0.
+
+## Cloud Events features
+
+The Cloud Events endpoint provides an alternative to the other connectivity APIs to stream data into your instance.
+
+## Cloud Events endpoint
+
+The Cloud Events endpoint is accessible at these URLs (depending on which API version to use):
+
+```
+http://localhost:8080/api/1/cloudevents
+http://localhost:8080/api/2/cloudevents
+```
+
+### Authentication
+
+A user who connects to the Cloud Events endpoint can be authenticated by using
+
+* HTTP BASIC Authentication by providing a username and the password of a user managed within nginx or
+* a JSON Web Token (JWT) issued by an OpenID connect provider.
+
+See [Authenticate](basic-auth.html) for more details.
+
+## Cloud Events protocol format
+
+The source must be a Cloud Event, encoded in the format for the HTTP binding. It may be encoded in *binary content mode*
+or in *structural content mode*.
+
+The *data content type* of the event must be `application/json`.
+
+The *data schema* must start with `ditto:`, for example `ditto:some-schema`.
+
+The events *payload* must in the Ditto Protocol JSON format as defined in the
+[Protocol specification](protocol-specification.html).
+
+## Publishing events to the endpoint
+
+Publishing events to the endpoint can be done by directly sending HTTP requests, conforming to the Cloud Events
+HTTP binding specification. Or by using other technologies that have adopted Cloud Events.
+
+### Knative eventing
+
+The endpoint can directly be configured as a [Knative eventing](https://knative.dev/docs/eventing/) destination.
+
+In the following example, a Knative eventing flow is configured to normalize the payload with a Vorto converter
+and send the result to Ditto's cloud events endpoint:
+
+~~~yaml
+apiVersion: flows.knative.dev/v1
+kind: Sequence
+metadata:
+ name: digital-twin
+spec:
+  channelTemplate:
+    apiVersion: messaging.knative.dev/v1alpha1
+    kind: KafkaChannel
+    spec:
+      numPartitions: 1
+      replicationFactor: 1
+  steps:
+  - ref:
+      # Convert incoming payload to the Ditto format
+      apiVersion: serving.knative.dev/v1
+      kind: Service
+      name: vorto-converter
+      namespace: digital-twin
+  reply:
+    # Deliver to Ditto Cloud Events endpoint
+    uri: http://ditto:ditto@ditto-nginx.digital-twin.svc.cluster.local:8080/api/2/cloudevents
+~~~
+
+This sequence itself can again be the target of another operation.
+
+### Direct invocation
+
+Of course, it is also possible to directly access the Cloud Events endpoint through HTTP:
+
+An example HTTP request could look like this:
+
+~~~
+POST /api/2/cloudevents HTTP/1.1
+ce-specversion: 1.0
+ce-type: my.ditto.event
+ce-time: 2020-11-24T14:35:00Z
+ce-id: f7b197fe-2e59-11eb-a8f4-d45d6455d2cc
+ce-source: /my/source
+ce-dataschema: ditto:some-schema
+Content-Type: application/json; charset=utf-8
+
+{
+    ... Ditto Protocol JSON ...
+}
+~~~
+
+For more information, see [HTTP Protocol Binding for CloudEvents - Version 1.0](https://github.com/cloudevents/spec/blob/v1.0/http-protocol-binding.md).

json/src/main/java/org/eclipse/ditto/json/JsonFactory.java

@@ -14,7 +14,10 @@
 
 import static java.util.Objects.requireNonNull;
 
+import java.io.ByteArrayInputStream;
+import java.io.InputStreamReader;
 import java.io.Reader;
+import java.nio.charset.StandardCharsets;
 import java.text.MessageFormat;
 import java.util.Arrays;
 import java.util.Collection;
@@ -37,6 +40,9 @@
 @Immutable
 public final class JsonFactory {
 
+    private static final String NULL_STRING = "null";
+    private static final byte[] NULL_DATA = NULL_STRING.getBytes(StandardCharsets.UTF_8);
+
     /*
      * This utility class is not meant to be instantiated.
      */
@@ -255,8 +261,42 @@ public static JsonObject newObject(final String jsonString) {
         }
     }
 
+    /**
+     * Creates a JSON object from the given byte array.
+     *
+     * @param jsonData the byte array that represents the JSON object.
+     * @return the JSON object that has been created from the data.
+     * @throws NullPointerException if {@code jsonData} is {@code null}.
+     * @throws IllegalArgumentException if {@code jsonData} is empty.
+     * @throws JsonParseException if {@code jsonData} does not contain a valid JSON object.
+     * @since 1.5.0
+     */
+    public static JsonObject newObject(final byte[] jsonData) {
+        requireNonNull(jsonData, "The JSON data to create a JSON object from must not be null!");
+        if (jsonData.length == 0) {
+            throw new IllegalArgumentException("The JSON data to create a JSON object from must not be empty!");
+        }
+
+        if (isJsonNullLiteralData(jsonData)) {
+            return nullObject();
+        } else {
+            final Reader reader = new InputStreamReader(new ByteArrayInputStream(jsonData));
+            final JsonValue jsonValue = JsonValueParser.fromReader().apply(reader);
+            if (!jsonValue.isObject()) {
+                final String msgPattern = "<{0}> is not a valid JSON object!";
+                throw JsonParseException.newBuilder()
+                        .message(MessageFormat.format(msgPattern, jsonValue)).build();
+            }
+            return jsonValue.asObject();
+        }
+    }
+
     private static boolean isJsonNullLiteralString(final String s) {
-        return "null".equals(s);
+        return NULL_STRING.equals(s);
+    }
+
+    private static boolean isJsonNullLiteralData(final byte[] data) {
+        return Arrays.equals(NULL_DATA, data);
     }
 
     /**

json/src/main/java/org/eclipse/ditto/json/JsonObject.java

@@ -38,6 +38,20 @@ static JsonObject of(final String jsonObjectString) {
         return JsonFactory.newObject(jsonObjectString);
     }
 
+    /**
+     * Creates a {@code JsonObject} from the given byte array.
+     *
+     * @param jsonData the byte array that represents the JSON object.
+     * @return the JSON object that has been created from the data.
+     * @throws NullPointerException if {@code jsonData} is {@code null}.
+     * @throws IllegalArgumentException if {@code jsonData} is empty.
+     * @throws JsonParseException if {@code jsonData} does not represent a valid JSON object.
+     * @since 1.5.0
+     */
+    static JsonObject of(final byte[] jsonData) {
+        return JsonFactory.newObject(jsonData);
+    }
+
     /**
      * Returns a new mutable builder with a fluent API for a {@code JsonObject}.
      *

legal/NOTICE.md

@@ -28,6 +28,7 @@ SPDX-License-Identifier: EPL-2.0
 * Copyright 2019 Kiwigrid GmbH
 * Copyright 2020 Bosch.IO GmbH
 * Copyright 2020 DevBoost GmbH
+* Copyright 2020 Red Hat Inc
 
 All content is the property of the respective authors or their employers.
 For more information regarding authorship of content, please consult the