diff --git a/components/camel-alibaba/camel-alibaba-eventbridge/src/main/docs/alibaba-eventbridge-component.adoc b/components/camel-alibaba/camel-alibaba-eventbridge/src/main/docs/alibaba-eventbridge-component.adoc index 5cda75c05ac11..c7eb1ce4706f0 100644 --- a/components/camel-alibaba/camel-alibaba-eventbridge/src/main/docs/alibaba-eventbridge-component.adoc +++ b/components/camel-alibaba/camel-alibaba-eventbridge/src/main/docs/alibaba-eventbridge-component.adoc @@ -44,13 +44,83 @@ include::partial$component-endpoint-headers.adoc[] == Usage +=== Message headers evaluated by the EventBridge producer + +[width="100%",cols="10%,10%,80%",options="header",] +|======================================================================= +|Header |Type |Description + +|`CamelAlibabaEventBridgeOperation` |`String` | Name of operation to invoke + +|`CamelAlibabaEventBridgeEventBusName` |`String` | Event bus name to publish events to (overrides the endpoint option) + +|`CamelAlibabaEventBridgeEventSource` |`String` | Event source URI (overrides the endpoint option) + +|`CamelAlibabaEventBridgeEventType` |`String` | Event type (overrides the endpoint option) + +|`CamelAlibabaEventBridgeEventSubject` |`String` | Event subject (overrides the endpoint option) + +|======================================================================= + +If any of the above headers are set, they will override their corresponding query parameter value. + +=== Message headers set by the EventBridge producer + +[width="100%",cols="10%,10%,80%",options="header",] +|======================================================================= +|Header |Type |Description + +|`CamelAlibabaEventBridgeRequestId` |`String` | Alibaba Cloud request ID returned by EventBridge + +|======================================================================= + +=== Event payload Map keys evaluated by the producer + +When the message body is a `Map` (or `List` for publishing multiple events), the following keys are evaluated: + +[width="100%",cols="10%,10%,80%",options="header",] +|======================================================================= +|Key |Type |Description + +|`eventBusName` |`String` | Event bus name (overrides header or endpoint option) + +|`eventSource` |`String` | Event source URI (overrides header or endpoint option) + +|`eventType` |`String` | Event type (overrides header or endpoint option) + +|`eventSubject` |`String` | Event subject (overrides header or endpoint option) + +|`eventData` |`Object` | Event payload data (serialized to JSON) + +|======================================================================= + +=== Response metadata in message body + +The `putEvents` producer operation returns structured response metadata in the message body (`Map`): + +[width="100%",cols="10%,10%,80%",options="header",] +|======================================================================= +|Key |Type |Description + +|`requestId` |`String` | Request identifier returned by EventBridge + +|`resourceOwnerAccountId` |`String` | Resource owner account identifier + +|`failedEntryCount` |`Integer` | Number of failed entries + +|`entryList` |`List` | List of entry results, where each entry map contains `eventId`, `errorCode`, and `errorMessage` + +|======================================================================= + === Operations The component supports the following operations: * `putEvents` - publish one or more CloudEvents (producer) -=== Producer example +== Examples + +=== Publish event with String / JSON body [source,java] ---- @@ -59,6 +129,30 @@ from("direct:start") .to("alibaba-eventbridge:putEvents?eventBusName=my-bus&eventSource=camel.test&eventType=OrderCreated®ion=cn-hangzhou&accessKey=RAW(accessKey)&secretKey=RAW(secretKey)"); ---- -== Examples +=== Publish event with Map body + +[source,java] +---- +Map event = new HashMap<>(); +event.put("eventBusName", "my-bus"); +event.put("eventSource", "camel.test"); +event.put("eventType", "OrderCreated"); +event.put("eventSubject", "order/123"); +event.put("eventData", Map.of("orderId", "123")); + +from("direct:start") + .setBody(constant(event)) + .to("alibaba-eventbridge:putEvents?region=cn-hangzhou&accessKey=RAW(accessKey)&secretKey=RAW(secretKey)"); +---- + +=== Publish multiple events + +[source,java] +---- +from("direct:start") + .setBody(constant(List.of(event1, event2))) + .to("alibaba-eventbridge:putEvents?region=cn-hangzhou&accessKey=RAW(accessKey)&secretKey=RAW(secretKey)"); +---- For more examples, see the unit tests in the `camel-alibaba-eventbridge` module. + diff --git a/components/camel-alibaba/camel-alibaba-eventbridge/src/main/java/org/apache/camel/component/alibaba/eventbridge/AlibabaEventBridgeUtils.java b/components/camel-alibaba/camel-alibaba-eventbridge/src/main/java/org/apache/camel/component/alibaba/eventbridge/AlibabaEventBridgeUtils.java index 302337731ef2d..622fb226bdc3a 100644 --- a/components/camel-alibaba/camel-alibaba-eventbridge/src/main/java/org/apache/camel/component/alibaba/eventbridge/AlibabaEventBridgeUtils.java +++ b/components/camel-alibaba/camel-alibaba-eventbridge/src/main/java/org/apache/camel/component/alibaba/eventbridge/AlibabaEventBridgeUtils.java @@ -32,6 +32,7 @@ import com.google.gson.Gson; import org.apache.camel.Exchange; import org.apache.camel.component.alibaba.common.OpenApiClientSupport; +import org.apache.camel.component.alibaba.eventbridge.constants.AlibabaEventBridgeConstants; import org.apache.camel.component.alibaba.eventbridge.constants.AlibabaEventBridgeProperties; import org.apache.camel.component.alibaba.eventbridge.models.ClientConfigurations; import org.apache.camel.util.ObjectHelper; @@ -99,11 +100,13 @@ private static CloudEvent toCloudEvent(Object body, ClientConfigurations configu } if (body instanceof Map mapBody) { - String eventBusName = stringValue(mapBody.get("eventBusName"), configuration.getEventBusName()); - String source = stringValue(mapBody.get("source"), configuration.getEventSource()); - String type = stringValue(mapBody.get("type"), configuration.getEventType()); - String subject = stringValue(mapBody.get("subject"), configuration.getEventSubject()); - String data = jsonDataValue(mapBody.get("data")); + String eventBusName + = stringValue(mapBody.get(AlibabaEventBridgeConstants.EVENT_BUS_NAME), configuration.getEventBusName()); + String source = stringValue(mapBody.get(AlibabaEventBridgeConstants.EVENT_SOURCE), configuration.getEventSource()); + String type = stringValue(mapBody.get(AlibabaEventBridgeConstants.EVENT_TYPE), configuration.getEventType()); + String subject + = stringValue(mapBody.get(AlibabaEventBridgeConstants.EVENT_SUBJECT), configuration.getEventSubject()); + String data = jsonDataValue(mapBody.get(AlibabaEventBridgeConstants.EVENT_DATA)); if (ObjectHelper.isEmpty(source) || ObjectHelper.isEmpty(type) || ObjectHelper.isEmpty(eventBusName)) { throw new IllegalArgumentException("Event source, type and event bus name are required"); @@ -174,20 +177,21 @@ private static String jsonDataValue(Object value) { public static Map toPutEventsMap(PutEventsResponse response) { Map map = new HashMap<>(); - map.put("requestId", response.getRequestId()); - map.put("resourceOwnerAccountId", response.getResourceOwnerAccountId()); - map.put("failedEntryCount", response.getFailedEntryCount()); + map.put(AlibabaEventBridgeConstants.EVENT_RESPONSE_REQUEST_IDENTIFIER, response.getRequestId()); + map.put(AlibabaEventBridgeConstants.EVENT_RESPONSE_RESOURCE_OWNER_ACCOUNT_IDENTIFIER, + response.getResourceOwnerAccountId()); + map.put(AlibabaEventBridgeConstants.EVENT_RESPONSE_FAILED_ENTRY_COUNT, response.getFailedEntryCount()); if (response.getEntryList() != null) { List> entries = new ArrayList<>(); for (PutEventsResponseEntry entry : response.getEntryList()) { Map entryMap = new HashMap<>(); - entryMap.put("eventId", entry.getEventId()); - entryMap.put("errorCode", entry.getErrorCode()); - entryMap.put("errorMessage", entry.getErrorMessage()); + entryMap.put(AlibabaEventBridgeConstants.EVENT_RESPONSE_ID, entry.getEventId()); + entryMap.put(AlibabaEventBridgeConstants.EVENT_RESPONSE_ERROR_CODE, entry.getErrorCode()); + entryMap.put(AlibabaEventBridgeConstants.EVENT_RESPONSE_ERROR_MESSAGE, entry.getErrorMessage()); entries.add(entryMap); } - map.put("entryList", entries); + map.put(AlibabaEventBridgeConstants.EVENT_RESPONSE_ENTRY_LIST, entries); } return map; } diff --git a/components/camel-alibaba/camel-alibaba-eventbridge/src/main/java/org/apache/camel/component/alibaba/eventbridge/constants/AlibabaEventBridgeConstants.java b/components/camel-alibaba/camel-alibaba-eventbridge/src/main/java/org/apache/camel/component/alibaba/eventbridge/constants/AlibabaEventBridgeConstants.java new file mode 100644 index 0000000000000..042ffbdba3f3a --- /dev/null +++ b/components/camel-alibaba/camel-alibaba-eventbridge/src/main/java/org/apache/camel/component/alibaba/eventbridge/constants/AlibabaEventBridgeConstants.java @@ -0,0 +1,40 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.alibaba.eventbridge.constants; + +/** + * Constants for Alibaba EventBridge payload and response dictionary keys. + */ +public final class AlibabaEventBridgeConstants { + + public static final String EVENT_BUS_NAME = "eventBusName"; + public static final String EVENT_SOURCE = "eventSource"; + public static final String EVENT_TYPE = "eventType"; + public static final String EVENT_SUBJECT = "eventSubject"; + public static final String EVENT_DATA = "eventData"; + + public static final String EVENT_RESPONSE_REQUEST_IDENTIFIER = "requestId"; + public static final String EVENT_RESPONSE_RESOURCE_OWNER_ACCOUNT_IDENTIFIER = "resourceOwnerAccountId"; + public static final String EVENT_RESPONSE_FAILED_ENTRY_COUNT = "failedEntryCount"; + public static final String EVENT_RESPONSE_ID = "eventId"; + public static final String EVENT_RESPONSE_ERROR_CODE = "errorCode"; + public static final String EVENT_RESPONSE_ERROR_MESSAGE = "errorMessage"; + public static final String EVENT_RESPONSE_ENTRY_LIST = "entryList"; + + private AlibabaEventBridgeConstants() { + } +} diff --git a/components/camel-alibaba/camel-alibaba-eventbridge/src/main/java/org/apache/camel/component/alibaba/eventbridge/constants/AlibabaEventBridgeHeaders.java b/components/camel-alibaba/camel-alibaba-eventbridge/src/main/java/org/apache/camel/component/alibaba/eventbridge/constants/AlibabaEventBridgeHeaders.java index 685135a85cb02..758775dcf9494 100644 --- a/components/camel-alibaba/camel-alibaba-eventbridge/src/main/java/org/apache/camel/component/alibaba/eventbridge/constants/AlibabaEventBridgeHeaders.java +++ b/components/camel-alibaba/camel-alibaba-eventbridge/src/main/java/org/apache/camel/component/alibaba/eventbridge/constants/AlibabaEventBridgeHeaders.java @@ -18,23 +18,23 @@ import org.apache.camel.spi.Metadata; -public final class AlibabaEventBridgeHeaders { +public sealed class AlibabaEventBridgeHeaders permits AlibabaEventBridgeProperties { @Metadata(label = "producer", description = "Event bus name override", javaType = "String") - public static final String EVENT_BUS_NAME = AlibabaEventBridgeProperties.EVENT_BUS_NAME; + public static final String EVENT_BUS_NAME = "CamelAlibabaEventBridgeEventBusName"; @Metadata(label = "producer", description = "Event source override", javaType = "String") - public static final String EVENT_SOURCE = AlibabaEventBridgeProperties.EVENT_SOURCE; + public static final String EVENT_SOURCE = "CamelAlibabaEventBridgeEventSource"; @Metadata(label = "producer", description = "Event type override", javaType = "String") - public static final String EVENT_TYPE = AlibabaEventBridgeProperties.EVENT_TYPE; + public static final String EVENT_TYPE = "CamelAlibabaEventBridgeEventType"; @Metadata(label = "producer", description = "Event subject override", javaType = "String") - public static final String EVENT_SUBJECT = AlibabaEventBridgeProperties.EVENT_SUBJECT; + public static final String EVENT_SUBJECT = "CamelAlibabaEventBridgeEventSubject"; @Metadata(label = "producer", description = "Alibaba Cloud request id", javaType = "String") - public static final String REQUEST_ID = AlibabaEventBridgeProperties.REQUEST_ID; + public static final String REQUEST_ID = "CamelAlibabaEventBridgeRequestId"; - private AlibabaEventBridgeHeaders() { + AlibabaEventBridgeHeaders() { } } diff --git a/components/camel-alibaba/camel-alibaba-eventbridge/src/main/java/org/apache/camel/component/alibaba/eventbridge/constants/AlibabaEventBridgeProperties.java b/components/camel-alibaba/camel-alibaba-eventbridge/src/main/java/org/apache/camel/component/alibaba/eventbridge/constants/AlibabaEventBridgeProperties.java index 102d13c721f38..23b79683d5deb 100644 --- a/components/camel-alibaba/camel-alibaba-eventbridge/src/main/java/org/apache/camel/component/alibaba/eventbridge/constants/AlibabaEventBridgeProperties.java +++ b/components/camel-alibaba/camel-alibaba-eventbridge/src/main/java/org/apache/camel/component/alibaba/eventbridge/constants/AlibabaEventBridgeProperties.java @@ -18,26 +18,11 @@ import org.apache.camel.spi.Metadata; -public final class AlibabaEventBridgeProperties { +public final class AlibabaEventBridgeProperties extends AlibabaEventBridgeHeaders { @Metadata(label = "producer", description = "Operation to perform", javaType = "String") public static final String OPERATION = "CamelAlibabaEventBridgeOperation"; - @Metadata(label = "producer", description = "Event bus name override", javaType = "String") - public static final String EVENT_BUS_NAME = "CamelAlibabaEventBridgeEventBusName"; - - @Metadata(label = "producer", description = "Event source override", javaType = "String") - public static final String EVENT_SOURCE = "CamelAlibabaEventBridgeEventSource"; - - @Metadata(label = "producer", description = "Event type override", javaType = "String") - public static final String EVENT_TYPE = "CamelAlibabaEventBridgeEventType"; - - @Metadata(label = "producer", description = "Event subject override", javaType = "String") - public static final String EVENT_SUBJECT = "CamelAlibabaEventBridgeEventSubject"; - - @Metadata(label = "producer", description = "Request id returned by EventBridge", javaType = "String") - public static final String REQUEST_ID = "CamelAlibabaEventBridgeRequestId"; - private AlibabaEventBridgeProperties() { } } diff --git a/components/camel-alibaba/camel-alibaba-eventbridge/src/test/java/org/apache/camel/component/alibaba/eventbridge/PutEventsTest.java b/components/camel-alibaba/camel-alibaba-eventbridge/src/test/java/org/apache/camel/component/alibaba/eventbridge/PutEventsTest.java index 0ea82472e2976..98ad284edae0e 100644 --- a/components/camel-alibaba/camel-alibaba-eventbridge/src/test/java/org/apache/camel/component/alibaba/eventbridge/PutEventsTest.java +++ b/components/camel-alibaba/camel-alibaba-eventbridge/src/test/java/org/apache/camel/component/alibaba/eventbridge/PutEventsTest.java @@ -18,14 +18,17 @@ import java.nio.charset.StandardCharsets; import java.util.HashMap; +import java.util.List; import java.util.Map; import com.aliyun.eventbridge.EventBridgeClient; import com.aliyun.eventbridge.models.CloudEvent; import com.aliyun.eventbridge.models.PutEventsResponse; +import com.aliyun.eventbridge.models.PutEventsResponseEntry; import org.apache.camel.BindToRegistry; import org.apache.camel.Exchange; import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.alibaba.eventbridge.constants.AlibabaEventBridgeConstants; import org.apache.camel.component.alibaba.eventbridge.constants.AlibabaEventBridgeHeaders; import org.apache.camel.component.mock.MockEndpoint; import org.apache.camel.test.junit6.CamelTestSupport; @@ -81,8 +84,8 @@ void testPutEvents() throws Exception { Exchange exchange = mock.getExchanges().get(0); assertThat(exchange.getMessage().getBody(Map.class)) - .containsEntry("requestId", "req-eb-1") - .containsEntry("failedEntryCount", 0); + .containsEntry(AlibabaEventBridgeConstants.EVENT_RESPONSE_REQUEST_IDENTIFIER, "req-eb-1") + .containsEntry(AlibabaEventBridgeConstants.EVENT_RESPONSE_FAILED_ENTRY_COUNT, 0); assertThat(exchange.getMessage().getHeader(AlibabaEventBridgeHeaders.REQUEST_ID)).isEqualTo("req-eb-1"); verify(eventBridgeClient).putEvents(anyList()); @@ -100,10 +103,10 @@ void testPutEventsWithMapBody() throws Exception { mock.expectedMinimumMessageCount(1); Map event = new HashMap<>(); - event.put("eventBusName", testConfiguration.getProperty("eventBusName")); - event.put("source", testConfiguration.getProperty("eventSource")); - event.put("type", testConfiguration.getProperty("eventType")); - event.put("data", Map.of("key", "value")); + event.put(AlibabaEventBridgeConstants.EVENT_BUS_NAME, testConfiguration.getProperty("eventBusName")); + event.put(AlibabaEventBridgeConstants.EVENT_SOURCE, testConfiguration.getProperty("eventSource")); + event.put(AlibabaEventBridgeConstants.EVENT_TYPE, testConfiguration.getProperty("eventType")); + event.put(AlibabaEventBridgeConstants.EVENT_DATA, Map.of("key", "value")); template.sendBody("direct:put", event); @@ -118,4 +121,86 @@ void testPutEventsWithMapBody() throws Exception { return data.contains("\"key\":\"value\""); })); } + + @Test + void testPutEventsWithListOfMapsAndSubjectAndEntryList() throws Exception { + PutEventsResponse response = new PutEventsResponse(); + response.setRequestId("req-eb-3"); + response.setResourceOwnerAccountId("123456789"); + response.setFailedEntryCount(0); + + PutEventsResponseEntry entry1 = new PutEventsResponseEntry(); + entry1.setEventId("event-id-1"); + entry1.setErrorCode("200"); + entry1.setErrorMessage("Success"); + + PutEventsResponseEntry entry2 = new PutEventsResponseEntry(); + entry2.setEventId("event-id-2"); + entry2.setErrorCode("200"); + entry2.setErrorMessage("Success"); + + response.setEntryList(List.of(entry1, entry2)); + + when(eventBridgeClient.putEvents(anyList())).thenReturn(response); + + MockEndpoint mock = getMockEndpoint("mock:result"); + mock.expectedMinimumMessageCount(1); + + Map event1 = new HashMap<>(); + event1.put(AlibabaEventBridgeConstants.EVENT_BUS_NAME, "custom-bus"); + event1.put(AlibabaEventBridgeConstants.EVENT_SOURCE, "camel.source.one"); + event1.put(AlibabaEventBridgeConstants.EVENT_TYPE, "OrderCreated"); + event1.put(AlibabaEventBridgeConstants.EVENT_SUBJECT, "order/1001"); + event1.put(AlibabaEventBridgeConstants.EVENT_DATA, Map.of("orderId", "1001")); + + Map event2 = new HashMap<>(); + event2.put(AlibabaEventBridgeConstants.EVENT_BUS_NAME, "custom-bus"); + event2.put(AlibabaEventBridgeConstants.EVENT_SOURCE, "camel.source.two"); + event2.put(AlibabaEventBridgeConstants.EVENT_TYPE, "OrderUpdated"); + event2.put(AlibabaEventBridgeConstants.EVENT_SUBJECT, "order/1002"); + event2.put(AlibabaEventBridgeConstants.EVENT_DATA, Map.of("orderId", "1002")); + + template.sendBody("direct:put", List.of(event1, event2)); + + mock.assertIsSatisfied(); + + Exchange exchange = mock.getExchanges().get(0); + Map body = exchange.getMessage().getBody(Map.class); + assertThat(body) + .containsEntry(AlibabaEventBridgeConstants.EVENT_RESPONSE_REQUEST_IDENTIFIER, "req-eb-3") + .containsEntry(AlibabaEventBridgeConstants.EVENT_RESPONSE_RESOURCE_OWNER_ACCOUNT_IDENTIFIER, "123456789") + .containsEntry(AlibabaEventBridgeConstants.EVENT_RESPONSE_FAILED_ENTRY_COUNT, 0) + .containsKey(AlibabaEventBridgeConstants.EVENT_RESPONSE_ENTRY_LIST); + + @SuppressWarnings("unchecked") + List> entryList + = (List>) body.get(AlibabaEventBridgeConstants.EVENT_RESPONSE_ENTRY_LIST); + assertThat(entryList).hasSize(2); + + Map entryMap1 = entryList.get(0); + assertThat(entryMap1) + .containsEntry(AlibabaEventBridgeConstants.EVENT_RESPONSE_ID, "event-id-1") + .containsEntry(AlibabaEventBridgeConstants.EVENT_RESPONSE_ERROR_CODE, "200") + .containsEntry(AlibabaEventBridgeConstants.EVENT_RESPONSE_ERROR_MESSAGE, "Success"); + + Map entryMap2 = entryList.get(1); + assertThat(entryMap2) + .containsEntry(AlibabaEventBridgeConstants.EVENT_RESPONSE_ID, "event-id-2") + .containsEntry(AlibabaEventBridgeConstants.EVENT_RESPONSE_ERROR_CODE, "200") + .containsEntry(AlibabaEventBridgeConstants.EVENT_RESPONSE_ERROR_MESSAGE, "Success"); + + assertThat(exchange.getMessage().getHeader(AlibabaEventBridgeHeaders.REQUEST_ID)).isEqualTo("req-eb-3"); + + verify(eventBridgeClient).putEvents(argThat(events -> { + if (events.size() != 2) { + return false; + } + CloudEvent ce1 = events.get(0); + CloudEvent ce2 = events.get(1); + return "order/1001".equals(ce1.getSubject()) + && "order/1002".equals(ce2.getSubject()) + && "camel.source.one".equals(ce1.getSource().toString()) + && "camel.source.two".equals(ce2.getSource().toString()); + })); + } }