Move DMF message converter in amqp-api (#3143)

- move in amqp to be in single place
- public DmfMessageConverter that could be used everywhere directly (instead of factory methods)
- defualt amqpMessageConverter bean renamed to dmfMessageConverter
- trusted packages configured (for dmfMessageConverter) with hawkbit.dmf.trusted-packages - default hwakbit dmf model package

Signed-off-by: Avgustin Marinov <Avgustin.Marinov@bosch.com>

hawkbit-dmf/hawkbit-dmf-amqp/src/main/java/org/eclipse/hawkbit/amqp/DmfApiConfiguration.java

@@ -9,14 +9,18 @@
  */
 package org.eclipse.hawkbit.amqp;
 
+import static org.eclipse.hawkbit.dmf.DmfMessageConverter.DMF_JSON_MODEL_PACKAGE;
+
 import java.sql.SQLException;
 import java.time.Duration;
+import java.util.Arrays;
 import java.util.List;
 import java.util.regex.Pattern;
 
 import lombok.ToString;
 import lombok.extern.slf4j.Slf4j;
 import org.eclipse.hawkbit.artifact.urlresolver.ArtifactUrlResolver;
+import org.eclipse.hawkbit.dmf.DmfMessageConverter;
 import org.eclipse.hawkbit.repository.ConfirmationManagement;
 import org.eclipse.hawkbit.repository.ControllerManagement;
 import org.eclipse.hawkbit.repository.DeploymentManagement;
@@ -27,9 +31,6 @@ import org.eclipse.hawkbit.repository.TargetManagement;
 import org.eclipse.hawkbit.repository.model.DistributionSet;
 import org.eclipse.hawkbit.repository.model.SoftwareModule;
 import org.eclipse.hawkbit.repository.model.Target;
-import org.jspecify.annotations.NonNull;
-import org.jspecify.annotations.Nullable;
-import org.springframework.amqp.core.Message;
 import org.springframework.amqp.listener.ConditionalRejectingErrorHandler;
 import org.springframework.amqp.listener.FatalExceptionStrategy;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
@@ -95,18 +96,20 @@ public class DmfApiConfiguration {
     }
 
     @Bean
-    @ConditionalOnMissingBean(name = "amqpMessageConverter") // override it if needed to add / edit trusted packages or need other customization
-    public MessageConverter amqpMessageConverter(final JsonMapper jsonMapper) {
-        return DmfApiConfiguration.messageConverter(jsonMapper);
+    @ConditionalOnMissingBean(name = "dmfMessageConverter") // override it if needed to add / edit trusted packages or need other customization
+    public MessageConverter dmfMessageConverter(
+            final JsonMapper jsonMapper,
+            @Value("${hawkbit.dmf.trusted-packages:" + DMF_JSON_MODEL_PACKAGE + "}") final String trustedPackages) {
+        return new DmfMessageConverter(jsonMapper, Arrays.stream(trustedPackages.split(",")).map(String::trim).toArray(String[]::new));
     }
 
     /**
      * @return {@link RabbitTemplate} with automatic retry, published confirms and {@link JacksonJsonMessageConverter}.
      */
     @Bean
-    public RabbitTemplate rabbitTemplate(@Qualifier("amqpMessageConverter") final MessageConverter amqpMessageConverter) {
+    public RabbitTemplate rabbitTemplate(@Qualifier("dmfMessageConverter") final MessageConverter dmfMessageConverter) {
         final RabbitTemplate rabbitTemplate = new RabbitTemplate(rabbitConnectionFactory);
-        rabbitTemplate.setMessageConverter(amqpMessageConverter);
+        rabbitTemplate.setMessageConverter(dmfMessageConverter);
 
         // the same policy the previously used default ExponentialBackOffPolicy applied
         rabbitTemplate.setRetryTemplate(new RetryTemplate(RetryPolicy.builder()
@@ -186,31 +189,6 @@ public class DmfApiConfiguration {
                 deploymentManagement);
     }
 
-    // since spring-amqp 4.0.4 not all packages are assumed trusted for type converter (only java.land and java.util)
-    // so e need to add hawkbit (and eventual extension packages as trusted
-    // also (again since spring-amqp 4.0.4) the conversion from empty payload fail (which probably is fine since it is JSON)
-    // however, (for backward compatibility, e.g. THING_REMOVED doesn't define payload and could be empty byte[]) we assume that
-    // empty payload is empty byte[] and not try to convert it to Object (which fail since it is not JSON)
-    static @NonNull JacksonJsonMessageConverter messageConverter(final JsonMapper jsonMapper) {
-        return messageConverter(jsonMapper, "org.eclipse.hawkbit.dmf.json.model");
-    }
-
-    public static @NonNull JacksonJsonMessageConverter messageConverter(final JsonMapper jsonMapper, final String... trustedPackages) {
-        return new JacksonJsonMessageConverter(jsonMapper, trustedPackages) {
-
-            @Override
-            public @NonNull Object fromMessage(@NonNull final Message message, final @Nullable Object conversionHint) {
-                // default converter tries to convert empty body payload to Object (since rabbit 4.0.4)
-                // which probably is correct since it has to be JSON - however, in this case we assume - empty byte[]
-                if (message.getBody().length == 0) {
-                    return message.getBody();
-                } else {
-                    return super.fromMessage(message, conversionHint);
-                }
-            }
-        };
-    }
-
     @ToString
     private static class SqlFatalExceptionStrategy implements FatalExceptionStrategy {
 

hawkbit-dmf/hawkbit-dmf-amqp/src/test/java/org/eclipse/hawkbit/amqp/AmqpMessageDispatcherServiceTest.java

@@ -26,13 +26,13 @@ import org.eclipse.hawkbit.artifact.model.ArtifactHashes;
 import org.eclipse.hawkbit.artifact.model.StoredArtifactInfo;
 import org.eclipse.hawkbit.artifact.urlresolver.ArtifactUrl;
 import org.eclipse.hawkbit.artifact.urlresolver.ArtifactUrlResolver;
+import org.eclipse.hawkbit.dmf.DmfMessageConverter;
 import org.eclipse.hawkbit.dmf.amqp.api.EventTopic;
 import org.eclipse.hawkbit.dmf.amqp.api.MessageHeaderKey;
 import org.eclipse.hawkbit.dmf.amqp.api.MessageType;
 import org.eclipse.hawkbit.dmf.json.model.DmfActionRequest;
 import org.eclipse.hawkbit.dmf.json.model.DmfDownloadAndUpdateRequest;
 import org.eclipse.hawkbit.dmf.json.model.DmfSoftwareModule;
-import org.eclipse.hawkbit.rabbitmq.test.AmqpTestConfiguration;
 import org.eclipse.hawkbit.repository.SystemManagement;
 import org.eclipse.hawkbit.repository.TargetManagement.Create;
 import org.eclipse.hawkbit.repository.event.remote.CancelTargetAssignmentEvent;
@@ -64,7 +64,6 @@ import org.springframework.amqp.support.converter.DefaultJacksonJavaTypeMapper;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.test.context.ActiveProfiles;
 import org.springframework.test.context.TestPropertySource;
-import tools.jackson.databind.json.JsonMapper;
 
 /**
  * Feature: Component Tests - Device Management Federation API<br/>
@@ -95,7 +94,7 @@ class AmqpMessageDispatcherServiceTest extends AbstractIntegrationTest {
                 Create.builder().controllerId(CONTROLLER_ID).securityToken(TEST_TOKEN).address(AMQP_URI.toString()).build());
 
         this.rabbitTemplate = Mockito.mock(RabbitTemplate.class);
-        when(rabbitTemplate.getMessageConverter()).thenReturn(AmqpTestConfiguration.messageConverter(new JsonMapper()));
+        when(rabbitTemplate.getMessageConverter()).thenReturn(new DmfMessageConverter());
 
         senderService = Mockito.mock(DefaultAmqpMessageSenderService.class);
 

hawkbit-dmf/hawkbit-dmf-amqp/src/test/java/org/eclipse/hawkbit/amqp/AmqpMessageHandlerServiceTest.java

@@ -24,6 +24,7 @@ import java.net.URI;
 import java.util.Map;
 import java.util.Optional;
 
+import org.eclipse.hawkbit.dmf.DmfMessageConverter;
 import org.eclipse.hawkbit.dmf.amqp.api.EventTopic;
 import org.eclipse.hawkbit.dmf.amqp.api.MessageHeaderKey;
 import org.eclipse.hawkbit.dmf.amqp.api.MessageType;
@@ -56,7 +57,6 @@ import org.springframework.amqp.core.MessageProperties;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.support.converter.MessageConversionException;
 import org.springframework.amqp.support.converter.MessageConverter;
-import tools.jackson.databind.json.JsonMapper;
 
 /**
  * Feature: Component Tests - Device Management Federation API<br/>
@@ -108,7 +108,7 @@ class AmqpMessageHandlerServiceTest {
     @SuppressWarnings({ "rawtypes", "unchecked" })
     void before() {
         TenantConfigHelper.setTenantConfigurationManagement(tenantConfigurationManagement);
-        messageConverter = DmfApiConfiguration.messageConverter(new JsonMapper());
+        messageConverter = new DmfMessageConverter();
         lenient().when(rabbitTemplate.getMessageConverter()).thenReturn(messageConverter);
 
         amqpMessageHandlerService = new AmqpMessageHandlerService(

hawkbit-dmf/hawkbit-dmf-amqp/src/test/java/org/eclipse/hawkbit/amqp/BaseAmqpServiceTest.java

@@ -15,6 +15,7 @@ import static org.mockito.Mockito.when;
 
 import java.util.List;
 
+import org.eclipse.hawkbit.dmf.DmfMessageConverter;
 import org.eclipse.hawkbit.dmf.json.model.DmfActionStatus;
 import org.eclipse.hawkbit.dmf.json.model.DmfActionUpdateStatus;
 import org.eclipse.hawkbit.repository.event.remote.entity.TargetCreatedEvent;
@@ -29,7 +30,6 @@ import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessageProperties;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.support.converter.MessageConversionException;
-import tools.jackson.databind.json.JsonMapper;
 
 /**
  * Feature: Component Tests - Device Management Federation API<br/>
@@ -54,7 +54,7 @@ class BaseAmqpServiceTest {
     @Test
     void convertMessageTest() {
         final DmfActionUpdateStatus actionUpdateStatus = createActionStatus();
-        when(rabbitTemplate.getMessageConverter()).thenReturn(DmfApiConfiguration.messageConverter(new JsonMapper()));
+        when(rabbitTemplate.getMessageConverter()).thenReturn(new DmfMessageConverter());
 
         final Message message = rabbitTemplate.getMessageConverter().toMessage(actionUpdateStatus, createJsonProperties());
         final DmfActionUpdateStatus convertedActionUpdateStatus = baseAmqpService.convertMessage(message, DmfActionUpdateStatus.class);
@@ -91,7 +91,7 @@ class BaseAmqpServiceTest {
     @ExpectEvents({ @Expect(type = TargetCreatedEvent.class, count = 0) })
     void updateActionStatusWithInvalidJsonContent() {
         final Message message = createMessage("Invalid Json".getBytes());
-        when(rabbitTemplate.getMessageConverter()).thenReturn(DmfApiConfiguration.messageConverter(new JsonMapper()));
+        when(rabbitTemplate.getMessageConverter()).thenReturn(new DmfMessageConverter());
 
         assertThatExceptionOfType(MessageConversionException.class)
                 .as("Expected MessageConversionException for invalid JSON")
@@ -111,5 +111,4 @@ class BaseAmqpServiceTest {
     private DmfActionUpdateStatus createActionStatus() {
         return new DmfActionUpdateStatus(1L, DmfActionStatus.RUNNING, null, 2L, List.of("Message 1", "Message 2"), 2);
     }
-
-}
+}