diff --git a/hawkbit-repository/hawkbit-repository-core/src/main/java/org/eclipse/hawkbit/event/EventJacksonMessageConverter.java b/hawkbit-repository/hawkbit-repository-core/src/main/java/org/eclipse/hawkbit/event/EventJacksonMessageConverter.java index 039dfb16c..cfd2b98d2 100644 --- a/hawkbit-repository/hawkbit-repository-core/src/main/java/org/eclipse/hawkbit/event/EventJacksonMessageConverter.java +++ b/hawkbit-repository/hawkbit-repository-core/src/main/java/org/eclipse/hawkbit/event/EventJacksonMessageConverter.java @@ -10,15 +10,30 @@ package org.eclipse.hawkbit.event; import com.fasterxml.jackson.databind.ObjectMapper; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.converter.MappingJackson2MessageConverter; import org.springframework.util.MimeType; public class EventJacksonMessageConverter extends MappingJackson2MessageConverter { + public static final MimeType APPLICATION_REMOTE_EVENT_JSON = new MimeType("application", "remote-event-json"); + + public EventJacksonMessageConverter() { - super(new MimeType("application", "remote-event-json")); + super(APPLICATION_REMOTE_EVENT_JSON); ObjectMapper objectMapper = new ObjectMapper(); EventType.getNamedTypes().forEach(objectMapper::registerSubtypes); setObjectMapper(objectMapper); } + + @Override + protected Object convertToInternal(final Object payload, final MessageHeaders headers, final Object conversionHint) { + return super.convertToInternal(payload, headers, conversionHint); + } + + @Override + protected Object convertFromInternal(final Message message, final Class targetClass, final Object conversionHint) { + return super.convertFromInternal(message, targetClass, conversionHint); + } } diff --git a/hawkbit-repository/hawkbit-repository-core/src/main/java/org/eclipse/hawkbit/event/EventProtoStuffMessageConverter.java b/hawkbit-repository/hawkbit-repository-core/src/main/java/org/eclipse/hawkbit/event/EventProtoStuffMessageConverter.java index 0e3aa222e..2b8d2504c 100644 --- a/hawkbit-repository/hawkbit-repository-core/src/main/java/org/eclipse/hawkbit/event/EventProtoStuffMessageConverter.java +++ b/hawkbit-repository/hawkbit-repository-core/src/main/java/org/eclipse/hawkbit/event/EventProtoStuffMessageConverter.java @@ -9,6 +9,8 @@ */ package org.eclipse.hawkbit.event; +import java.nio.ByteBuffer; + import io.protostuff.LinkedBuffer; import io.protostuff.ProtobufIOUtil; import io.protostuff.Schema; @@ -22,20 +24,28 @@ import org.springframework.messaging.converter.MessageConversionException; import org.springframework.util.MimeType; /** - * A customize message converter for the spring cloud events. The converter is registered for the application/binary+protostuff type. - *

- * The clazz-type-information is encoded into the message payload infront with a length of {@link #EVENT_TYPE_LENGTH}. This is necessary - * due in case of rabbitMQ batching the message headers will be merged together and custom message header information will get lost. - * So in this implementation the information about the event-type is encoded in the payload of the message directly using the encoded - * values of {@link EventType}. + * A custom message converter for Spring Cloud Stream using Protostuff serialization. + * The converter is registered for the {@code application/binary+protostuff} content type. + * It embeds the {@link EventType} metadata inside the message payload to preserve the type + * during deserialization — even when message headers may be lost (e.g. with RabbitMQ batching). + * + *

+ * Message Structure: + *

+ * + * This format allows decoding messages without relying on external headers, ensuring robustness + * in systems where header information may be merged or dropped. */ @Slf4j public class EventProtoStuffMessageConverter extends AbstractMessageConverter { public static final MimeType APPLICATION_BINARY_PROTOSTUFF = new MimeType("application", "binary+protostuff"); + private static final int HEADER_LENGTH_PREFIX_SIZE = 4; - /** The length of the class type length of the payload. */ - private static final byte EVENT_TYPE_LENGTH = 2; public EventProtoStuffMessageConverter() { super(APPLICATION_BINARY_PROTOSTUFF); @@ -47,11 +57,11 @@ public class EventProtoStuffMessageConverter extends AbstractMessageConverter { } @Override - public Object convertFromInternal(final Message message, final Class targetClass, final Object conversionHint) { + protected Object convertFromInternal(final Message message, final Class targetClass, final Object conversionHint) { final Object objectPayload = message.getPayload(); if (objectPayload instanceof byte[] payload) { final byte[] clazzHeader = extractClazzHeader(payload); - final byte[] content = extraxtContent(payload); + final byte[] content = extractContent(payload); final EventType eventType = readClassHeader(clazzHeader); return readContent(eventType, content); @@ -86,17 +96,23 @@ public class EventProtoStuffMessageConverter extends AbstractMessageConverter { } private static byte[] extractClazzHeader(final byte[] payload) { - final byte[] clazzHeader = new byte[EVENT_TYPE_LENGTH]; - System.arraycopy(payload, 0, clazzHeader, 0, EVENT_TYPE_LENGTH); + ByteBuffer wrapper = ByteBuffer.wrap(payload); + int headerLength = wrapper.getInt(); + byte[] clazzHeader = new byte[headerLength]; + wrapper.get(clazzHeader); return clazzHeader; } - private static byte[] extraxtContent(final byte[] payload) { - final byte[] content = new byte[payload.length - EVENT_TYPE_LENGTH]; - System.arraycopy(payload, EVENT_TYPE_LENGTH, content, 0, content.length); + private static byte[] extractContent(final byte[] payload) { + ByteBuffer wrapper = ByteBuffer.wrap(payload); + int headerLength = wrapper.getInt(); + byte[] content = new byte[payload.length - HEADER_LENGTH_PREFIX_SIZE - headerLength]; + wrapper.position(HEADER_LENGTH_PREFIX_SIZE + headerLength); + wrapper.get(content); return content; } + private static EventType readClassHeader(final byte[] typeInformation) { final Schema schema = RuntimeSchema.getSchema(EventType.class); final EventType deserializedType = schema.newMessage(); @@ -117,9 +133,16 @@ public class EventProtoStuffMessageConverter extends AbstractMessageConverter { log.error("There is no mapping to EventType for the given class {}", clazz); throw new MessageConversionException("Missing EventType for given class : " + clazz); } - @SuppressWarnings("unchecked") final Schema schema = (Schema) RuntimeSchema - .getSchema((Class) EventType.class); + + @SuppressWarnings("unchecked") + final Schema schema = (Schema) RuntimeSchema.getSchema((Class) EventType.class); final LinkedBuffer buffer = LinkedBuffer.allocate(); - return ProtobufIOUtil.toByteArray(clazzEventType, schema, buffer); + byte[] typeBytes = ProtobufIOUtil.toByteArray(clazzEventType, schema, buffer); + + ByteBuffer result = ByteBuffer.allocate(HEADER_LENGTH_PREFIX_SIZE + typeBytes.length); + result.putInt(typeBytes.length); + result.put(typeBytes); + return result.array(); } + } \ No newline at end of file diff --git a/hawkbit-repository/hawkbit-repository-core/src/test/java/org/eclipse/hawkbit/event/AbstractEventMessageConverterTest.java b/hawkbit-repository/hawkbit-repository-core/src/test/java/org/eclipse/hawkbit/event/AbstractEventMessageConverterTest.java new file mode 100644 index 000000000..9464a642e --- /dev/null +++ b/hawkbit-repository/hawkbit-repository-core/src/test/java/org/eclipse/hawkbit/event/AbstractEventMessageConverterTest.java @@ -0,0 +1,87 @@ +/** + * Copyright (c) 2025 Contributors to the Eclipse Foundation + * + * This program and the accompanying materials are made + * available under the terms of the Eclipse Public License 2.0 + * which is available at https://www.eclipse.org/legal/epl-2.0/ + * + * SPDX-License-Identifier: EPL-2.0 + */ +package org.eclipse.hawkbit.event; + +import org.eclipse.hawkbit.repository.event.remote.AbstractRemoteEvent; +import org.eclipse.hawkbit.repository.event.remote.entity.TargetCreatedEvent; +import org.eclipse.hawkbit.repository.event.remote.service.TargetCreatedServiceEvent; +import org.eclipse.hawkbit.repository.model.Target; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.Mock; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.converter.MessageConverter; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.when; + +import java.util.HashMap; + +abstract class AbstractEventMessageConverterTest { + + protected final MessageConverter messageConverter; + + @Mock + protected Message messageMock; + + @Mock + protected Target targetMock; + + @BeforeEach + void before() { + when(targetMock.getId()).thenReturn(1L); + } + + AbstractEventMessageConverterTest(MessageConverter messageConverter) { + this.messageConverter = messageConverter; + } + + /** + * Verifies that the TargetCreatedEvent can be successfully serialized and deserialized + */ + @Test + void successfullySerializeAndDeserializeEvent() { + final TargetCreatedEvent targetCreatedEvent = new TargetCreatedEvent(targetMock); + // serialize + assertSerializeAndDeserialize(targetCreatedEvent, TargetCreatedEvent.class); + assertSerializeAndDeserialize(targetCreatedEvent, TargetCreatedEvent.class); + } + + @Test + void successfullySerializeAndDeserializeServiceEvent() { + final TargetCreatedEvent targetCreatedEvent = new TargetCreatedEvent(targetMock); + final TargetCreatedServiceEvent targetCreatedServiceEvent = new TargetCreatedServiceEvent(targetCreatedEvent); + assertSerializeAndDeserialize(targetCreatedServiceEvent, TargetCreatedServiceEvent.class); + } + + void assertSerializeAndDeserialize(T event, Class expectedClass) { + // serialize + Object serializedEvent = null; + if (messageConverter instanceof EventProtoStuffMessageConverter protoStuff) { + serializedEvent = protoStuff.convertToInternal(event, new MessageHeaders(new HashMap<>()), null); + } else if (messageConverter instanceof EventJacksonMessageConverter jackson) { + serializedEvent = jackson.convertToInternal(event, new MessageHeaders(new HashMap<>()), null); + } + assertThat(serializedEvent).isInstanceOf(byte[].class); + + // deserialize + when(messageMock.getPayload()).thenReturn(serializedEvent); + Object deserializedEvent = null; + if (messageConverter instanceof EventProtoStuffMessageConverter protoStuff) { + deserializedEvent = protoStuff.convertFromInternal(messageMock, AbstractRemoteEvent.class, null); + } else if (messageConverter instanceof EventJacksonMessageConverter jackson) { + deserializedEvent = jackson.convertFromInternal(messageMock, AbstractRemoteEvent.class, null); + } + assertThat(deserializedEvent) + .isInstanceOf(expectedClass) + .isEqualTo(event); + } +} diff --git a/hawkbit-repository/hawkbit-repository-core/src/test/java/org/eclipse/hawkbit/event/EventJacksonMessageConverterTest.java b/hawkbit-repository/hawkbit-repository-core/src/test/java/org/eclipse/hawkbit/event/EventJacksonMessageConverterTest.java index 2a721c834..a7b5a31a0 100644 --- a/hawkbit-repository/hawkbit-repository-core/src/test/java/org/eclipse/hawkbit/event/EventJacksonMessageConverterTest.java +++ b/hawkbit-repository/hawkbit-repository-core/src/test/java/org/eclipse/hawkbit/event/EventJacksonMessageConverterTest.java @@ -9,69 +9,13 @@ */ package org.eclipse.hawkbit.event; -import static org.assertj.core.api.Assertions.assertThat; -import static org.mockito.Mockito.when; - -import java.util.HashMap; - -import org.eclipse.hawkbit.repository.event.remote.AbstractRemoteEvent; -import org.eclipse.hawkbit.repository.event.remote.entity.TargetCreatedEvent; -import org.eclipse.hawkbit.repository.model.Target; -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; -import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; -import org.springframework.messaging.Message; -import org.springframework.messaging.MessageHeaders; @ExtendWith(MockitoExtension.class) -class EventJacksonMessageConverterTest { +class EventJacksonMessageConverterTest extends AbstractEventMessageConverterTest { - private final TestEventJacksonMessageConverter underTest = new TestEventJacksonMessageConverter(); - - @Mock - private Target targetMock; - - @Mock - private Message messageMock; - - @BeforeEach - void before() { - when(targetMock.getId()).thenReturn(1L); - } - - /** - * Verifies that the TargetCreatedEvent can be successfully serialized and deserialized - */ - @Test - void successfullySerializeAndDeserializeEvent() { - final TargetCreatedEvent targetCreatedEvent = new TargetCreatedEvent(targetMock); - // serialize - final Object serializedEvent = underTest.convertToInternal(targetCreatedEvent, - new MessageHeaders(new HashMap<>()), null); - assertThat(serializedEvent).isInstanceOf(byte[].class); - - // deserialize - when(messageMock.getPayload()).thenReturn(serializedEvent); - final Object deserializedEvent = underTest.convertFromInternal(messageMock, AbstractRemoteEvent.class, null); - assertThat(deserializedEvent) - .isInstanceOf(TargetCreatedEvent.class) - .isEqualTo(targetCreatedEvent); - } - - /** - * Test subclass to expose protected methods for testing. - */ - private static class TestEventJacksonMessageConverter extends EventJacksonMessageConverter { - @Override - public Object convertToInternal(Object payload, MessageHeaders headers, Object conversionHint) { - return super.convertToInternal(payload, headers, conversionHint); - } - - @Override - public Object convertFromInternal(Message message, Class targetClass, Object conversionHint) { - return super.convertFromInternal(message, targetClass, conversionHint); - } + EventJacksonMessageConverterTest() { + super(new EventJacksonMessageConverter()); } } \ No newline at end of file diff --git a/hawkbit-repository/hawkbit-repository-core/src/test/java/org/eclipse/hawkbit/event/EventProtoStuffMessageConverterTest.java b/hawkbit-repository/hawkbit-repository-core/src/test/java/org/eclipse/hawkbit/event/EventProtoStuffMessageConverterTest.java index 260472c2b..8502e81b2 100644 --- a/hawkbit-repository/hawkbit-repository-core/src/test/java/org/eclipse/hawkbit/event/EventProtoStuffMessageConverterTest.java +++ b/hawkbit-repository/hawkbit-repository-core/src/test/java/org/eclipse/hawkbit/event/EventProtoStuffMessageConverterTest.java @@ -9,59 +9,24 @@ */ package org.eclipse.hawkbit.event; -import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatExceptionOfType; -import static org.mockito.Mockito.when; import java.io.Serial; import java.util.HashMap; -import org.eclipse.hawkbit.repository.event.remote.AbstractRemoteEvent; import org.eclipse.hawkbit.repository.event.remote.entity.RemoteEntityEvent; -import org.eclipse.hawkbit.repository.event.remote.entity.TargetCreatedEvent; import org.eclipse.hawkbit.repository.model.Target; -import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; -import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; -import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.converter.MessageConversionException; @ExtendWith(MockitoExtension.class) -class EventProtoStuffMessageConverterTest { +class EventProtoStuffMessageConverterTest extends AbstractEventMessageConverterTest { - private final EventProtoStuffMessageConverter underTest = new EventProtoStuffMessageConverter(); - - @Mock - private Target targetMock; - - @Mock - private Message messageMock; - - @BeforeEach - void before() { - when(targetMock.getId()).thenReturn(1L); - } - - /** - * Verifies that the TargetCreatedEvent can be successfully serialized and deserialized - */ - @Test - void successfullySerializeAndDeserializeEvent() { - final TargetCreatedEvent targetCreatedEvent = new TargetCreatedEvent(targetMock); - // serialize - final Object serializedEvent = underTest.convertToInternal(targetCreatedEvent, - new MessageHeaders(new HashMap<>()), null); - assertThat(serializedEvent).isInstanceOf(byte[].class); - - // deserialize - when(messageMock.getPayload()).thenReturn(serializedEvent); - final Object deserializedEvent = underTest.convertFromInternal(messageMock, AbstractRemoteEvent.class, null); - assertThat(deserializedEvent) - .isInstanceOf(TargetCreatedEvent.class) - .isEqualTo(targetCreatedEvent); + EventProtoStuffMessageConverterTest() { + super(new EventProtoStuffMessageConverter()); } /** @@ -74,7 +39,7 @@ class EventProtoStuffMessageConverterTest { assertThatExceptionOfType(MessageConversionException.class) .as("Missing MessageConversationException for un-defined event-type") - .isThrownBy(() -> underTest.convertToInternal(dummyEvent, messageHeaders, null)); + .isThrownBy(() -> ((EventProtoStuffMessageConverter)messageConverter).convertToInternal(dummyEvent, messageHeaders, null)); } /** diff --git a/hawkbit-repository/hawkbit-repository-jpa/src/test/java/org/eclipse/hawkbit/repository/event/remote/AbstractRemoteEventTest.java b/hawkbit-repository/hawkbit-repository-jpa/src/test/java/org/eclipse/hawkbit/repository/event/remote/AbstractRemoteEventTest.java index 1ff670d1a..edc1999db 100644 --- a/hawkbit-repository/hawkbit-repository-jpa/src/test/java/org/eclipse/hawkbit/repository/event/remote/AbstractRemoteEventTest.java +++ b/hawkbit-repository/hawkbit-repository-jpa/src/test/java/org/eclipse/hawkbit/repository/event/remote/AbstractRemoteEventTest.java @@ -8,50 +8,29 @@ * SPDX-License-Identifier: EPL-2.0 */ package org.eclipse.hawkbit.repository.event.remote; - -import static org.junit.jupiter.api.Assertions.fail; - import java.util.Map; - -import com.fasterxml.jackson.core.JsonProcessingException; -import com.fasterxml.jackson.databind.ObjectMapper; +import org.eclipse.hawkbit.event.EventJacksonMessageConverter; import org.eclipse.hawkbit.event.EventProtoStuffMessageConverter; import org.eclipse.hawkbit.repository.event.TenantAwareEvent; import org.eclipse.hawkbit.repository.jpa.AbstractJpaIntegrationTest; -import org.junit.jupiter.api.BeforeEach; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.test.context.TestConfiguration; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Import; -import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.support.MutableMessageHeaders; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; -import org.springframework.messaging.converter.AbstractMessageConverter; -import org.springframework.messaging.converter.MappingJackson2MessageConverter; -import org.springframework.messaging.converter.MessageConverter; -import org.springframework.util.MimeTypeUtils; /** * Test the remote entity events. */ @SuppressWarnings("java:S6813") // constructor injects are not possible for test classes -@Import(AbstractRemoteEventTest.EventProtoStuffTestConfig.class) public abstract class AbstractRemoteEventTest extends AbstractJpaIntegrationTest { - @Autowired - private EventProtoStuffMessageConverter eventProtoStuffMessageConverter; + private EventProtoStuffMessageConverter eventProtoStuffMessageConverter = new EventProtoStuffMessageConverter(); - private AbstractMessageConverter jacksonMessageConverter; + private EventJacksonMessageConverter jacksonMessageConverter = new EventJacksonMessageConverter(); - @BeforeEach - public void setup() { - this.jacksonMessageConverter = new MappingJackson2MessageConverter(); - } @SuppressWarnings("unchecked") protected T createJacksonEvent(final T event) { - final Message message = createJsonMessage(event); + final Message message = createJsonMessage(event); return (T) jacksonMessageConverter.fromMessage(message, event.getClass()); } @@ -68,23 +47,9 @@ public abstract class AbstractRemoteEventTest extends AbstractJpaIntegrationTest ); } - private Message createJsonMessage(final Object event) { - try { - String json = new ObjectMapper().writeValueAsString(event); - return MessageBuilder.withPayload(json) - .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON) - .build(); - } catch (JsonProcessingException e) { - fail(e.getMessage()); - } - return null; + private Message createJsonMessage(final Object event) { + return jacksonMessageConverter.toMessage(event, new MutableMessageHeaders(Map.of(MessageHeaders.CONTENT_TYPE, + EventJacksonMessageConverter.APPLICATION_REMOTE_EVENT_JSON))); } - @TestConfiguration - static class EventProtoStuffTestConfig { - @Bean - public MessageConverter eventProtoBufConverter() { - return new EventProtoStuffMessageConverter(); - } - } } \ No newline at end of file