chore(next): Unify rabbitmq starter with starter eda

async/async-commons/src/main/java/org/reactivecommons/async/commons/converters/json/CloudEventBuilderExt.java

@@ -2,6 +2,7 @@
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import io.cloudevents.CloudEvent;
+import io.cloudevents.CloudEventData;
 import lombok.SneakyThrows;
 import lombok.experimental.UtilityClass;
 
@@ -16,6 +17,10 @@ public static byte[] asBytes(Object object) {
         return mapper.writeValueAsBytes(object);
     }
 
+    public static CloudEventData asCloudEventData(Object object) {
+        return () -> asBytes(object);
+    }
+
     @SneakyThrows
     public static <T> T fromCloudEventData(CloudEvent cloudEvent, Class<T> classValue) {
         return mapper.readValue(Objects.requireNonNull(cloudEvent.getData()).toBytes(), classValue);

async/async-kafka/src/test/java/org/reactivecommons/async/kafka/listeners/GenericMessageListenerTest.java

@@ -72,8 +72,6 @@ void shouldStartListener() {
         Headers header = new RecordHeaders().add("contentType", "application/json".getBytes(StandardCharsets.UTF_8));
         when(record.headers()).thenReturn(header);
         when(record.key()).thenReturn("key");
-        ReceiverOffset receiverOffset = mock(ReceiverOffset.class);
-        when(record.receiverOffset()).thenReturn(receiverOffset);
 
         Flux<ReceiverRecord<String, byte[]>> flux = Flux.just(record);
         when(receiver.listen(anyString(), any(List.class))).thenReturn(flux);
@@ -91,7 +89,6 @@ void shouldStartListener() {
         StepVerifier.create(flow).expectNext("").verifyComplete();
         // Assert
         verify(topologyCreator, times(1)).createTopics(any(List.class));
-        verify(receiverOffset, atLeastOnce()).acknowledge();
     }