arunsrajan commented on code in PR #26010:
URL: https://github.com/apache/camel/pull/26010#discussion_r3916119989
##########
components/camel-alibaba/camel-alibaba-eventbridge/src/main/java/org/apache/camel/component/alibaba/eventbridge/AlibabaEventBridgeUtils.java:
##########
@@ -62,117 +74,346 @@ public static EventBridgeClient
createClient(AlibabaEventBridgeEndpoint endpoint
}
public static ClientConfigurations
createClientConfigurations(AlibabaEventBridgeEndpoint endpoint, Exchange
exchange) {
- ClientConfigurations configuration = new ClientConfigurations();
- configuration.setOperation(
- OpenApiClientSupport.resolveString(exchange,
AlibabaEventBridgeProperties.OPERATION, endpoint.getOperation()));
- configuration.setEventBusName(
- OpenApiClientSupport.resolveString(exchange,
AlibabaEventBridgeProperties.EVENT_BUS_NAME,
- endpoint.getEventBusName()));
- configuration.setEventSource(
+ String defaultBusName = OpenApiClientSupport.resolveString(
+ exchange, AlibabaEventBridgeProperties.EVENT_BUS_NAME,
endpoint.getEventBusName());
+
+ return new ClientConfigurations(
+ OpenApiClientSupport.resolveString(exchange,
AlibabaEventBridgeProperties.OPERATION, endpoint.getOperation()),
+ defaultBusName,
OpenApiClientSupport.resolveString(exchange,
AlibabaEventBridgeProperties.EVENT_SOURCE,
- endpoint.getEventSource()));
- configuration.setEventType(
- OpenApiClientSupport.resolveString(exchange,
AlibabaEventBridgeProperties.EVENT_TYPE, endpoint.getEventType()));
- configuration.setEventSubject(
+ endpoint.getEventSource()),
+ OpenApiClientSupport.resolveString(exchange,
AlibabaEventBridgeProperties.EVENT_TYPE, endpoint.getEventType()),
OpenApiClientSupport.resolveString(exchange,
AlibabaEventBridgeProperties.EVENT_SUBJECT,
- endpoint.getEventSubject()));
- return configuration;
+ endpoint.getEventSubject()),
+ OpenApiClientSupport.resolveBoolean(exchange,
AlibabaEventBridgeProperties.VALIDATE_EVENT_SOURCE,
+ endpoint.isValidateEventSource()),
+ OpenApiClientSupport.resolveBoolean(exchange,
AlibabaEventBridgeProperties.VALIDATE_EVENT_TYPE,
+ endpoint.isValidateEventType()),
+ OpenApiClientSupport.resolveBoolean(exchange,
AlibabaEventBridgeProperties.VALIDATE_EVENT_SPEC,
+ endpoint.isValidateEventSpec()),
+ resolveAllowedEventBuses(exchange,
AlibabaEventBridgeProperties.ALLOWED_EVENT_SOURCES,
+ endpoint.getAllowedEventSources(), defaultBusName),
+ OpenApiClientSupport.resolveLong(exchange,
AlibabaEventBridgeProperties.EVENT_SOURCE_CACHE_TTL,
+ endpoint.getEventSourceCacheTtl()));
}
public static List<CloudEvent> resolveCloudEvents(Exchange exchange,
ClientConfigurations configuration) {
+ return resolveCloudEvents(exchange, configuration, null, null);
+ }
+
+ public static List<CloudEvent> resolveCloudEvents(
+ Exchange exchange, ClientConfigurations configuration,
+ EventSourceCache eventSourceCache, EventBridgeClient client) {
Object body = exchange.getMessage().getBody();
List<CloudEvent> events = new ArrayList<>();
+ MapCloudEventValidator mapValidator = new
MapCloudEventValidator(eventSourceCache);
if (body instanceof List<?> listBody) {
for (Object item : listBody) {
- events.add(toCloudEvent(item, configuration));
+ events.add(toCloudEvent(item, configuration, mapValidator,
client));
}
return events;
}
- events.add(toCloudEvent(body, configuration));
+ events.add(toCloudEvent(body, configuration, mapValidator, client));
return events;
}
- private static CloudEvent toCloudEvent(Object body, ClientConfigurations
configuration) {
+ private static CloudEvent toCloudEvent(
+ Object body, ClientConfigurations configuration,
+ MapCloudEventValidator mapValidator, EventBridgeClient client) {
if (body instanceof CloudEvent cloudEvent) {
+ mapValidator.validateCloudEvent(cloudEvent, configuration, client);
return cloudEvent;
}
if (body instanceof Map<?, ?> mapBody) {
- 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");
- }
-
- EventBuilder builder = EventBuilder.builder()
- .withSource(URI.create(source))
- .withType(type)
- .withAliyunEventBus(eventBusName);
-
- if (ObjectHelper.isNotEmpty(subject)) {
- builder.withSubject(subject);
- }
- if (data != null) {
- builder.withJsonStringData(data);
- }
- return builder.build();
+ return mapValidator.validateAndBuild(mapBody, configuration,
client);
}
if (body instanceof String stringBody) {
- if (ObjectHelper.isEmpty(configuration.getEventSource())
- || ObjectHelper.isEmpty(configuration.getEventType())
- || ObjectHelper.isEmpty(configuration.getEventBusName())) {
+ if (ObjectHelper.isEmpty(configuration.eventSource())
+ || ObjectHelper.isEmpty(configuration.eventType())
+ || ObjectHelper.isEmpty(configuration.eventBusName())) {
throw new IllegalArgumentException("Event source, type and
event bus name are required when body is a string");
}
+ mapValidator.validateBusSourceAndType(
+ configuration.eventBusName(), configuration.eventSource(),
configuration.eventType(),
+ configuration, client);
+
EventBuilder builder = EventBuilder.builder()
- .withSource(URI.create(configuration.getEventSource()))
- .withType(configuration.getEventType())
- .withAliyunEventBus(configuration.getEventBusName())
+ .withSource(URI.create(configuration.eventSource()))
+ .withType(configuration.eventType())
+ .withAliyunEventBus(configuration.eventBusName())
.withJsonStringData(stringBody);
- if (ObjectHelper.isNotEmpty(configuration.getEventSubject())) {
- builder.withSubject(configuration.getEventSubject());
+ if (ObjectHelper.isNotEmpty(configuration.eventSubject())) {
+ builder.withSubject(configuration.eventSubject());
}
return builder.build();
}
throw new IllegalArgumentException("Exchange body must be a
CloudEvent, Map or JSON string");
}
- private static String stringValue(Object value, String fallback) {
- if (value == null) {
- return fallback;
+ /**
+ * Resolves and parses allowed event buses, sources, and source-scoped
event types from Header, Property, or
+ * Endpoint option.
+ */
+ public static Map<String, AllowedEventBus> resolveAllowedEventBuses(
+ Exchange exchange, String name, String endpointValue, String
defaultBusName) {
+ Object raw = exchange.getIn().getHeader(name);
+ if (raw == null) {
+ raw = exchange.getProperty(name);
}
- if (value instanceof String stringValue) {
- return stringValue;
+ if (raw == null) {
+ raw = endpointValue;
+ }
+
+ if (raw == null) {
+ return Collections.emptyMap();
+ }
+
+ if (raw instanceof Map<?, ?> map) {
+ return parseAllowedBusesFromMap(map, defaultBusName);
+ }
+
+ if (raw instanceof Collection<?> coll) {
+ return parseAllowedBusesFromCollection(coll, defaultBusName);
}
- return value.toString();
+
+ if (raw instanceof String str) {
+ return parseAllowedBusesFromString(str, defaultBusName);
+ }
+
+ return Collections.emptyMap();
}
- private static String jsonDataValue(Object value) {
- if (value == null) {
- return null;
+ private static Map<String, AllowedEventBus>
parseAllowedBusesFromMap(Map<?, ?> map, String defaultBusName) {
+ Map<String, AllowedEventBus> result = new HashMap<>();
+ for (Map.Entry<?, ?> entry : map.entrySet()) {
+ String busOrSource = String.valueOf(entry.getKey()).trim();
+ Object value = entry.getValue();
+
+ if (value instanceof AllowedEventBus allowedBus) {
+ result.put(allowedBus.eventBusName(), allowedBus);
+ } else if (value instanceof Map<?, ?> subMap) {
+ Map<String, AllowedEventSource> sources = new HashMap<>();
+ for (Map.Entry<?, ?> subEntry : subMap.entrySet()) {
+ String src = String.valueOf(subEntry.getKey()).trim();
+ Set<String> types = toStringSet(subEntry.getValue());
+ sources.put(src, new AllowedEventSource(src, types));
+ }
+ result.put(busOrSource, new AllowedEventBus(busOrSource,
sources));
+ } else if (value instanceof Collection<?> || value instanceof
String) {
+ String busName = ObjectHelper.isNotEmpty(defaultBusName) ?
defaultBusName : "*";
+ AllowedEventBus bus = result.computeIfAbsent(busName, k -> new
AllowedEventBus(busName, new HashMap<>()));
+ Map<String, AllowedEventSource> modifiable = new
HashMap<>(bus.allowedSources());
+ modifiable.put(busOrSource, new
AllowedEventSource(busOrSource, toStringSet(value)));
+ result.put(busName, new AllowedEventBus(busName, modifiable));
+ }
+ }
+ return result;
+ }
+
+ private static Map<String, AllowedEventBus>
parseAllowedBusesFromCollection(Collection<?> coll, String defaultBusName) {
+ Map<String, AllowedEventBus> result = new HashMap<>();
+ String busName = ObjectHelper.isNotEmpty(defaultBusName) ?
defaultBusName : "*";
+ Map<String, AllowedEventSource> sources = new HashMap<>();
Review Comment:
Good catch, thank you @gnodet!
I have removed the colon-based fallback parsing from `parseSourcesBlock`.
Now, mapping between event source and allowed event types is performed
exclusively via `->` and `=`. When neither operator is present, the entry is
parsed cleanly as the full source string without type restrictions.
I've also:
- Added unit tests in `MapCloudEventValidationTest`
(`testSingleBusDslWithHttpUriAndPort` and `testSingleBusDslWithUrnAndEquals`)
verifying that URIs containing scheme colons and port numbers (e.g.
`http://example.com:8080/events`, `urn:custom:source`) are parsed without
accidental splitting.
- Updated `alibaba-eventbridge-component.adoc` and the catalog documentation
mirror to clarify this behavior.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
