Merge branch 'master' into support-updating-referenced-WoT-ThingModel

eclipse-ditto · Jan 27, 2025 · 3d4d025 · 3d4d025
2 parents 7125a94 + 42dfa7c
commit 3d4d025
Show file tree

Hide file tree

Showing 88 changed files with 943 additions and 658 deletions.
diff --git a/.github/workflows/system-tests.yml b/.github/workflows/system-tests.yml
@@ -118,21 +118,14 @@ jobs:
           cache: 'maven'
           java-version: 21
 
-      - name: Cache local Maven repository
-        uses: actions/cache@v4
-        with:
-          path: ~/.m2/repository
-          key: ${{ runner.os }}-maven-${{ hashFiles('**/pom.xml') }}
-          restore-keys: |
-            ${{ runner.os }}-maven-
-
       - name: Set env vars dynamically
         run: |
           echo "JAVA_TOOL_OPTIONS=-Duser.home=$HOME" >> $GITHUB_ENV
           echo "M2_REPO=$HOME/.m2" >> $GITHUB_ENV
           if [[ $GITHUB_ACTOR != "nektos/act" ]]; then
             echo "HOST_M2_REPO=$HOME/.m2" >> $GITHUB_ENV
           fi
+          echo "DITTO_BRANCH_NO_SLASH=${{ env.DITTO_BRANCH }}" | sed 's/\//-/g' >> $GITHUB_ENV
 
       - name: Debug maven repo
         run: |
@@ -358,8 +351,8 @@ jobs:
         if:  env.GITHUB_ACTOR != 'nektos/act'
         uses: actions/upload-artifact@v3
         with:
-          name: system-test-results-${{ env.DITTO_BRANCH }}-${{ github.run_number }}
-          path: 'ditto-testing/**/target/failsafe-reports/**/*.xml'
+          name: system-test-results-${{ env.DITTO_BRANCH_NO_SLASH }}-${{ github.run_number }}
+          path: 'ditto-testing/system*/**/target/failsafe-reports/**/*.xml'
 
       - name: Upload services logs
         if:  env.GITHUB_ACTOR != 'nektos/act'
@@ -369,15 +362,16 @@ jobs:
           path: 'ditto-testing/docker/*.log'
 
       - name: List Test Result files
+        if: always()
         run: |
-          ls -lah ./**/target/failsafe-reports/*IT.xml
+          ls -lah ./system*/**/target/failsafe-reports/*IT.xml
         working-directory: ditto-testing
 
       - name: Publish Test Results
         uses: dorny/test-reporter@v1
         with:
-          name: Test Results | ${{ env.DITTO_BRANCH }} ${{ github.run_number }}
-          path: ./**/target/failsafe-reports/*IT.xml
+          name: Test Results | ${{ env.DITTO_BRANCH_NO_SLASH }} ${{ github.run_number }}
+          path: ./system*/**/target/failsafe-reports/*IT.xml
           reporter: java-junit
           working-directory: ditto-testing
 
diff --git a/README.md b/README.md
@@ -7,7 +7,7 @@
 # Eclipse Ditto™
 
 [![Join the chat at https://gitter.im/eclipse/ditto](https://badges.gitter.im/eclipse/ditto.svg)](https://gitter.im/eclipse/ditto?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=badge)
-[![Build Status](https://github.com/eclipse-ditto/ditto/workflows/build/badge.svg)](https://github.com/eclipse-ditto/ditto/actions?query=workflow%3Abuild)
+[![Build Status](https://github.com/eclipse-ditto/ditto/actions/workflows/maven.yml/badge.svg)](https://github.com/eclipse-ditto/ditto/actions/workflows/maven.yml?query=workflow%3Abuild)
 [![Maven Central](https://img.shields.io/maven-central/v/org.eclipse.ditto/ditto?label=maven)](https://search.maven.org/search?q=g:org.eclipse.ditto)
 [![Docker pulls](https://img.shields.io/docker/pulls/eclipse/ditto-things.svg)](https://hub.docker.com/search?q=eclipse%2Fditto&type=image)
 [![License](https://img.shields.io/badge/License-EPL%202.0-green.svg)](https://opensource.org/licenses/EPL-2.0)

diff --git a/bom/pom.xml b/bom/pom.xml
@@ -70,7 +70,7 @@
         <cloudevents.version>2.5.0</cloudevents.version>
 
         <slf4j.version>2.0.16</slf4j.version>
-        <logback.version>1.5.7</logback.version>
+        <logback.version>1.5.13</logback.version>
         <logstash-logback-encoder.version>8.0</logstash-logback-encoder.version>
         <fluency.version>2.7.2</fluency.version>
         <janino.version>3.1.12</janino.version>

diff --git a/...va/org/eclipse/ditto/connectivity/service/mapping/javascript/ScriptedOutgoingMapping.java b/...va/org/eclipse/ditto/connectivity/service/mapping/javascript/ScriptedOutgoingMapping.java
@@ -42,6 +42,7 @@
 import org.eclipse.ditto.protocol.JsonifiableAdaptable;
 import org.eclipse.ditto.protocol.ProtocolFactory;
 import org.mozilla.javascript.Callable;
+import org.mozilla.javascript.ConsString;
 import org.mozilla.javascript.Context;
 import org.mozilla.javascript.ContextFactory;
 import org.mozilla.javascript.NativeArray;
@@ -51,6 +52,8 @@
 import org.mozilla.javascript.Scriptable;
 import org.mozilla.javascript.Undefined;
 import org.mozilla.javascript.typedarrays.NativeArrayBuffer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Mapping function for outgoing messages based on JavaScript.
@@ -63,6 +66,7 @@ public final class ScriptedOutgoingMapping implements MappingFunction<Adaptable,
     private static final String EXTERNAL_MESSAGE_BYTE_PAYLOAD = "bytePayload";
 
     private static final String OUTGOING_FUNCTION_NAME = "mapFromDittoProtocolMsgWrapper";
+    private static final Logger log = LoggerFactory.getLogger(ScriptedOutgoingMapping.class);
 
     @Nullable private final ContextFactory contextFactory;
     @Nullable private final Scriptable scope;
@@ -159,14 +163,33 @@ private static JsonObject toJsonObject(final NativeObject nativeObject) {
         nativeObject.forEach((key, value) -> {
             try {
                 if (value instanceof String) {
-                    objectBuilder.set(key.toString(), JsonFactory.readFrom(value.toString()));
+                    objectBuilder.set(key.toString(), JsonFactory.newValue(value.toString()));
+                } else if (value instanceof ConsString consString) {
+                    objectBuilder.set(key.toString(), JsonFactory.newValue(consString.toString()));
                 } else if (value instanceof NativeArray nativeArray) {
                     objectBuilder.set(key.toString(), toJsonArray(nativeArray));
                 } else if (value instanceof NativeObject nativeSubObject) {
                     objectBuilder.set(key.toString(), toJsonObject(nativeSubObject));
+                } else if (value instanceof Boolean boolValue) {
+                    objectBuilder.set(key.toString(), JsonFactory.newValue(boolValue));
+                } else if (value instanceof Integer intValue) {
+                    objectBuilder.set(key.toString(), JsonFactory.newValue(intValue));
+                } else if (value instanceof Double doubleValue) {
+                    objectBuilder.set(key.toString(), JsonFactory.newValue(doubleValue));
+                } else if (value == null || value instanceof Undefined) {
+                    objectBuilder.set(key.toString(), JsonFactory.nullLiteral());
+                } else {
+                    if (log.isDebugEnabled()){
+                        log.debug("Unsupported type: {}, adding as string: {}", value.getClass().getName(), value);
+                    }
+                    objectBuilder.set(key.toString(), value.toString());
                 }
             } catch (final JsonParseException e) {
-                objectBuilder.set(key.toString(), value.toString());
+                if (value != null) {
+                    objectBuilder.set(key.toString(), value.toString());
+                } else {
+                    objectBuilder.set(key.toString(), JsonFactory.nullLiteral());
+                }
             }
         });
         return objectBuilder.build();

diff --git a/...ce/src/main/java/org/eclipse/ditto/connectivity/service/messaging/BasePublisherActor.java b/...ce/src/main/java/org/eclipse/ditto/connectivity/service/messaging/BasePublisherActor.java
@@ -482,7 +482,12 @@ private SendingOrDropped publishToGenericTarget(final ExpressionResolver resolve
                     .tag(SpanTagKey.CONNECTION_TARGET.getTagForValue(publishTarget.toString()))
                     .start();
             final var mappedMessageWithTraceContext =
-                    mappedMessage.withHeaders(startedSpan.propagateContext(mappedMessage.getHeaders()));
+                    mappedMessage.withHeaders(startedSpan.propagateContext(
+                            DittoHeaders.newBuilder(mappedMessage.getHeaders())
+                                    .removeHeader(DittoHeaderDefinition.W3C_TRACEPARENT.getKey())
+                                    .build()
+                                    .asCaseSensitiveMap()
+                    ));
 
             final CompletionStage<SendResult> responsesFuture = publishMessage(outboundSource,
                     autoAckTarget,

diff --git a/...rc/main/java/org/eclipse/ditto/connectivity/service/messaging/amqp/AmqpConsumerActor.java b/...rc/main/java/org/eclipse/ditto/connectivity/service/messaging/amqp/AmqpConsumerActor.java
@@ -37,6 +37,13 @@
 import javax.jms.MessageListener;
 import javax.jms.TextMessage;
 
+import org.apache.pekko.Done;
+import org.apache.pekko.actor.ActorRef;
+import org.apache.pekko.actor.Props;
+import org.apache.pekko.actor.Status;
+import org.apache.pekko.japi.pf.ReceiveBuilder;
+import org.apache.pekko.pattern.Patterns;
+import org.apache.pekko.stream.javadsl.Sink;
 import org.apache.qpid.jms.JmsAcknowledgeCallback;
 import org.apache.qpid.jms.JmsMessageConsumer;
 import org.apache.qpid.jms.message.JmsMessage;
@@ -63,20 +70,12 @@
 import org.eclipse.ditto.connectivity.service.messaging.internal.ConnectionFailure;
 import org.eclipse.ditto.connectivity.service.messaging.internal.RetrieveAddressStatus;
 import org.eclipse.ditto.connectivity.service.messaging.monitoring.logs.InfoProviderFactory;
-import org.eclipse.ditto.internal.utils.pekko.logging.ThreadSafeDittoLoggingAdapter;
 import org.eclipse.ditto.internal.utils.config.InstanceIdentifierSupplier;
+import org.eclipse.ditto.internal.utils.pekko.logging.ThreadSafeDittoLoggingAdapter;
 import org.eclipse.ditto.internal.utils.tracing.DittoTracing;
 import org.eclipse.ditto.internal.utils.tracing.span.SpanOperationName;
 import org.eclipse.ditto.internal.utils.tracing.span.TracingSpans;
 
-import org.apache.pekko.Done;
-import org.apache.pekko.actor.ActorRef;
-import org.apache.pekko.actor.Props;
-import org.apache.pekko.actor.Status;
-import org.apache.pekko.japi.pf.ReceiveBuilder;
-import org.apache.pekko.pattern.Patterns;
-import org.apache.pekko.stream.javadsl.Sink;
-
 /**
  * Actor which receives message from an AMQP source and forwards them to a {@code MessageMappingProcessorActor}.
  */
@@ -342,10 +341,11 @@ private void messageConsumerFailed(final Status.Failure failure) {
         handleAddressStatus(addressStatus);
     }
 
-    private void handleJmsMessage(final JmsMessage message) {
+    private void handleJmsMessage(final JmsMessage message) throws JMSException {
         Map<String, String> headers = null;
-        String correlationId = null;
-        var startedSpan = TracingSpans.emptyStartedSpan(SpanOperationName.of("amqp_consume"));
+        var startedSpan = TracingSpans.emptyStartedSpan(
+                SpanOperationName.of("amqp_consume: " + message.getJMSDestination())
+        );
         try {
             recordIncomingForRateLimit(message.getJMSMessageID());
             if (logger.isDebugEnabled()) {
@@ -358,12 +358,17 @@ private void handleJmsMessage(final JmsMessage message) {
                         ackType);
             }
             headers = extractHeadersMapFromJmsMessage(message);
-            correlationId = headers.get(DittoHeaderDefinition.CORRELATION_ID.getKey());
+            final String correlationId = headers.get(DittoHeaderDefinition.CORRELATION_ID.getKey());
             startedSpan = DittoTracing.newPreparedSpan(headers, startedSpan.getOperationName())
                     .correlationId(correlationId)
                     .connectionId(connectionId)
                     .start();
-            headers = startedSpan.propagateContext(headers);
+            headers = startedSpan.propagateContext(DittoHeaders.of(headers)
+                    .toBuilder()
+                    .removeHeader(DittoHeaderDefinition.W3C_TRACEPARENT.getKey())
+                    .build()
+                    .asCaseSensitiveMap()
+            );
             final ExternalMessageBuilder builder = ExternalMessageFactory.newExternalMessageBuilder(headers);
             final ExternalMessage externalMessage = extractPayloadFromMessage(message, builder)
                     .withAuthorizationContext(source.getAuthorizationContext())

diff --git a/.../java/org/eclipse/ditto/connectivity/service/messaging/kafka/KafkaMessageTransformer.java b/.../java/org/eclipse/ditto/connectivity/service/messaging/kafka/KafkaMessageTransformer.java
@@ -24,6 +24,7 @@
 
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.common.header.Header;
+import org.apache.pekko.kafka.ConsumerMessage;
 import org.eclipse.ditto.base.model.common.ByteBufferUtils;
 import org.eclipse.ditto.base.model.common.CharsetDeterminer;
 import org.eclipse.ditto.base.model.exceptions.DittoRuntimeException;
@@ -41,8 +42,6 @@
 import org.eclipse.ditto.internal.utils.tracing.DittoTracing;
 import org.eclipse.ditto.internal.utils.tracing.span.SpanOperationName;
 
-import org.apache.pekko.kafka.ConsumerMessage;
-
 /**
  * Transforms incoming messages from Apache Kafka to {@link org.eclipse.ditto.connectivity.api.ExternalMessage}.
  */
@@ -108,11 +107,17 @@ public TransformationResult transform(final ConsumerRecord<String, ByteBuffer> c
         final String correlationId = messageHeaders
                 .getOrDefault(DittoHeaderDefinition.CORRELATION_ID.getKey(), UUID.randomUUID().toString());
 
-        final var startedSpan = DittoTracing.newPreparedSpan(messageHeaders, SpanOperationName.of("kafka_consume"))
-                .correlationId(correlationId)
+        final var startedSpan = DittoTracing.newPreparedSpan(messageHeaders,
+                        SpanOperationName.of("kafka_consume: " + consumerRecord.topic())
+                ).correlationId(correlationId)
                 .connectionId(connectionId)
                 .start();
-        messageHeaders = startedSpan.propagateContext(messageHeaders);
+        messageHeaders = startedSpan.propagateContext(DittoHeaders.of(messageHeaders)
+                .toBuilder()
+                .removeHeader(DittoHeaderDefinition.W3C_TRACEPARENT.getKey())
+                .build()
+                .asCaseSensitiveMap()
+        );
 
         try {
             final String key = consumerRecord.key();

diff --git a/...java/org/eclipse/ditto/connectivity/service/messaging/rabbitmq/RabbitMQConsumerActor.java b/...java/org/eclipse/ditto/connectivity/service/messaging/rabbitmq/RabbitMQConsumerActor.java
@@ -22,6 +22,11 @@
 
 import javax.annotation.Nullable;
 
+import org.apache.pekko.Done;
+import org.apache.pekko.NotUsed;
+import org.apache.pekko.actor.Props;
+import org.apache.pekko.japi.pf.ReceiveBuilder;
+import org.apache.pekko.stream.javadsl.Sink;
 import org.eclipse.ditto.base.model.common.CharsetDeterminer;
 import org.eclipse.ditto.base.model.exceptions.DittoRuntimeException;
 import org.eclipse.ditto.base.model.headers.DittoHeaderDefinition;
@@ -51,12 +56,6 @@
 import com.rabbitmq.client.Delivery;
 import com.rabbitmq.client.Envelope;
 
-import org.apache.pekko.Done;
-import org.apache.pekko.NotUsed;
-import org.apache.pekko.actor.Props;
-import org.apache.pekko.japi.pf.ReceiveBuilder;
-import org.apache.pekko.stream.javadsl.Sink;
-
 
 /**
  * Actor which receives message from an RabbitMQ source and forwards them to a {@code MessageMappingProcessorActor}.
@@ -135,7 +134,9 @@ private void handleDelivery(final Delivery delivery) {
         final Envelope envelope = delivery.getEnvelope();
         final byte[] body = delivery.getBody();
 
-        var startedSpan = TracingSpans.emptyStartedSpan(SpanOperationName.of("rabbitmq_consume"));
+        var startedSpan = TracingSpans.emptyStartedSpan(
+                SpanOperationName.of("rabbitmq_consume: " + envelope.getExchange())
+        );
         Map<String, String> headers = null;
         try {
             @Nullable final String correlationId = properties.getCorrelationId();
@@ -150,7 +151,12 @@ private void handleDelivery(final Delivery delivery) {
                     .connectionId(connectionId)
                     .correlationId(correlationId)
                     .start();
-            headers = startedSpan.propagateContext(headers);
+            headers = startedSpan.propagateContext(DittoHeaders.of(headers)
+                    .toBuilder()
+                    .removeHeader(DittoHeaderDefinition.W3C_TRACEPARENT.getKey())
+                    .build()
+                    .asCaseSensitiveMap()
+            );
 
             final ExternalMessageBuilder externalMessageBuilder =
                     ExternalMessageFactory.newExternalMessageBuilder(headers);

diff --git a/deployment/docker/sandbox/docker-compose.yml b/deployment/docker/sandbox/docker-compose.yml
@@ -217,14 +217,6 @@ services:
       options:
         max-size: 10m
 
-#  node-red:
-#    image: docker.io/nodered/node-red-docker:latest
-#    volumes:
-#       - /opt/docker-volumes/nodered-data:/data:rw
-#       - ./nodered/settings.js:/data/settings.js:ro
-#       - ./nodered/package.json:/data/package.json:rw
-#       - ./nodered/flows-ditto.json:/data/flows-ditto.json:rw
-
   nginx:
     image: docker.io/nginx:1.27-alpine
     deploy:

diff --git a/deployment/docker/sandbox/nginx.conf b/deployment/docker/sandbox/nginx.conf
@@ -274,16 +274,6 @@ http {
       proxy_set_header              Host                $http_host;
     }
 
-#    location /nodered {
-#      proxy_pass http://node-red:1880/nodered;
-#      proxy_http_version 1.1;
-#      proxy_set_header   Host $host;
-#      proxy_set_header Upgrade $http_upgrade;
-#      proxy_set_header Connection 'upgrade';
-#      proxy_set_header Host $host;
-#      proxy_cache_bypass $http_upgrade;
-#    }
-
     location /oauth2/google/token {
       # set $args '';
       proxy_set_body $request_body&client_secret=TODOinsertOauth2clientSecret;