This is an automated email from the ASF dual-hosted git repository. timothyjward pushed a commit to branch fix/cyclic-events in repository https://gitbox.apache.org/repos/asf/aries-typedevent.git
commit fc19f6af3cbe2ae26415a863efb183a834554c99 Author: Tim Ward <[email protected]> AuthorDate: Wed Nov 4 12:04:45 2026 +0000 Handle cyclic events without stopping delivery Flattening now tracks object identity so cycles survive as cyclic Maps, and a failed conversion is remembered rather than repeated. A StackOverflowError used to kill the delivery thread, so every later event was lost. Signed-off-by: Tim Ward <[email protected]> --- org.apache.aries.typedevent.bus/pom.xml | 1 - .../aries/typedevent/bus/impl/EventConverter.java | 368 +++++++++++++++++---- .../aries/typedevent/bus/impl/EventTask.java | 5 +- .../typedevent/bus/impl/MonitorEventTask.java | 10 +- .../typedevent/bus/impl/UnhandledEventTask.java | 9 +- .../typedevent/bus/impl/UntypedEventTask.java | 9 +- .../bus/impl/EventConverterCycleTest.java | 352 ++++++++++++++++++++ .../typedevent/bus/impl/TypedEventBusImplTest.java | 102 ++++++ .../bus/osgi/TypedEventMonitorIntegrationTest.java | 39 +++ .../org.apache.aries.typedevent.remote.spi/pom.xml | 19 +- .../remote/spi/RemoteEventMonitorImpl.java | 93 +++--- .../spi/RemoteEventMonitorImplFlattenTest.java | 92 ++++++ 12 files changed, 985 insertions(+), 114 deletions(-) diff --git a/org.apache.aries.typedevent.bus/pom.xml b/org.apache.aries.typedevent.bus/pom.xml index 86060d3..de8fcde 100644 --- a/org.apache.aries.typedevent.bus/pom.xml +++ b/org.apache.aries.typedevent.bus/pom.xml @@ -122,7 +122,6 @@ <groupId>org.osgi</groupId> <artifactId>org.osgi.util.promise</artifactId> <version>1.1.0</version> - <scope>runtime</scope> </dependency> </dependencies> diff --git a/org.apache.aries.typedevent.bus/src/main/java/org/apache/aries/typedevent/bus/impl/EventConverter.java b/org.apache.aries.typedevent.bus/src/main/java/org/apache/aries/typedevent/bus/impl/EventConverter.java index ef8a9e3..5e504b3 100644 --- a/org.apache.aries.typedevent.bus/src/main/java/org/apache/aries/typedevent/bus/impl/EventConverter.java +++ b/org.apache.aries.typedevent.bus/src/main/java/org/apache/aries/typedevent/bus/impl/EventConverter.java @@ -21,6 +21,7 @@ import static java.lang.ThreadLocal.withInitial; import static java.util.Collections.newSetFromMap; import java.lang.reflect.Array; +import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Type; import java.time.Duration; import java.time.Instant; @@ -33,23 +34,30 @@ import java.time.OffsetTime; import java.time.Year; import java.time.YearMonth; import java.time.ZonedDateTime; +import java.util.ArrayList; import java.util.Calendar; import java.util.Collection; import java.util.Date; import java.util.HashSet; import java.util.IdentityHashMap; -import java.util.List; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; import java.util.Map; import java.util.Set; import java.util.UUID; +import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; import org.apache.aries.typedevent.bus.spi.CustomEventConverter; import org.apache.aries.typedevent.bus.spi.TypeData; import org.osgi.framework.Filter; +import org.osgi.util.converter.ConversionException; import org.osgi.util.converter.Converter; import org.osgi.util.converter.ConverterFunction; import org.osgi.util.converter.Converters; import org.osgi.util.converter.TypeReference; +import org.osgi.util.promise.Deferred; +import org.osgi.util.promise.Promise; +import org.osgi.util.promise.Promises; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -63,15 +71,11 @@ public class EventConverter { private static final Logger _log = LoggerFactory.getLogger(EventConverter.class); - private static final TypeReference<List<Object>> LIST_OF_OBJECTS = new TypeReference<List<Object>>() { - }; - private static final TypeReference<Set<Object>> SET_OF_OBJECTS = new TypeReference<Set<Object>>() { - }; static final TypeReference<Map<String, Object>> MAP_WITH_STRING_KEYS = new TypeReference<Map<String, Object>>() { }; - private static final TypeReference<Map<Object, Object>> MAP_OF_OBJECT_TO_OBJECT = new TypeReference<Map<Object, Object>>() { - }; + // Converts a DTO to a Map of its raw field values, without flattening the values + private static final Converter shallowConverter = Converters.standardConverter(); private static final Converter eventConverter; private static final Set<Class<?>> safeClasses; private static final Set<Class<?>> specialClasses; @@ -151,27 +155,24 @@ public class EventConverter { // ensures we pick up any embedded lists of DTOs or enums if (o instanceof Collection) { if(_log.isTraceEnabled()) { - _log.trace("The collection {} will be processed to map the contained objects.", o); + _log.trace("The collection of type {} will be processed to map the contained objects.", sourceClass); } - if (o instanceof Set) { - return eventConverter.convert(o).to(SET_OF_OBJECTS); - } else { - return eventConverter.convert(o).to(LIST_OF_OBJECTS); - } + return flatten(o, o instanceof Set ? new LinkedHashSet<>() : new ArrayList<>(), + EventConverter::fillCollection); } // As with collections we remap nested maps to clean up any // undesirable types in the keys or values if (o instanceof Map) { if(_log.isTraceEnabled()) { - _log.trace("The map {} will be processed to remap the contained objects.", o); + _log.trace("The map of type {} will be processed to remap the contained objects.", sourceClass); } - return eventConverter.convert(o).to(MAP_OF_OBJECT_TO_OBJECT); + return flatten(o, new LinkedHashMap<>(), EventConverter::fillMap); } if (sourceClass.isArray()) { if(_log.isTraceEnabled()) { - _log.trace("The array {} will be processed to remap the contained objects.", o); + _log.trace("The array of type {} will be processed to remap the contained objects.", sourceClass); } int depth = 1; Class<?> arrayComponentType = sourceClass.getComponentType(); @@ -182,32 +183,138 @@ public class EventConverter { } if (safeClasses.contains(actualComponentType) || actualComponentType.isPrimitive()) { if(_log.isTraceEnabled()) { - _log.trace("The array {} does not need conversion.", o); + _log.trace("The array of type {} does not need conversion.", sourceClass); } return o; } else if (actualComponentType.isEnum()) { if(_log.isTraceEnabled()) { - _log.trace("The array {} will use String values.", o); + _log.trace("The array of type {} will use String values.", sourceClass); } // This becomes an n dimensional String array Class<?> stringArrayType = Array.newInstance(String.class, new int[depth]).getClass(); return eventConverter.convert(o).to(stringArrayType); } else { if(_log.isTraceEnabled()) { - _log.trace("The array {} is complex and will be treated as a list.", o); + _log.trace("The array of type {} is complex and its elements will be remapped.", sourceClass); } - // This is an array of something complicated, recursively turn it into a - // list of something, then make it into an array of the right type - List<Object> oList = eventConverter.convert(o).to(LIST_OF_OBJECTS); - return oList.toArray((Object[]) Array.newInstance(arrayComponentType, 0)); + // This is an array of something complicated, recursively remap the + // elements into an array of the right type + return flatten(o, Array.newInstance(arrayComponentType, Array.getLength(o)), + EventConverter::fillArray); } } if(_log.isTraceEnabled()) { - _log.trace("The object {} will be treated as a DTO.", o); + _log.trace("The object of type {} will be treated as a DTO.", sourceClass); } // If we get here then treat the type as a DTO - return eventConverter.convert(o).sourceAsDTO().to(MAP_WITH_STRING_KEYS); + return flatten(o, new LinkedHashMap<String, Object>(), EventConverter::fillDto); + } + + /** + * The state of one flattening operation on the current thread + */ + private static final class FlatteningContext { + // Maps each source object seen so far to the result built for it + final Map<Object, Visit> visits = new IdentityHashMap<>(); + // Nested conversions use the same context as the outermost one + int depth; + // Set if an object was reached again while it was still being flattened + boolean cyclic; + } + + private static final class Visit { + final Object result; + boolean inProgress = true; + + Visit(Object result) { + this.result = result; + } + } + + @FunctionalInterface + private interface Filler { + void fill(Object source, Object result); + } + + private static final ThreadLocal<FlatteningContext> flattening = new ThreadLocal<>(); + + private static FlatteningContext enterFlattening() { + FlatteningContext context = flattening.get(); + if (context == null) { + context = new FlatteningContext(); + flattening.set(context); + } + context.depth++; + return context; + } + + private static void exitFlattening(FlatteningContext context) { + if (--context.depth == 0) { + flattening.remove(); + } + } + + /** + * Flatten a source object into the (empty) result container. The result is + * registered before it is filled, so that if the source refers back to itself + * the same result is used again, and the flattened form has the same shape. + */ + private static Object flatten(Object source, Object result, Filler filler) { + FlatteningContext context = enterFlattening(); + try { + Visit visit = context.visits.get(source); + if (visit != null) { + if (visit.inProgress) { + context.cyclic = true; + } + return visit.result; + } + visit = new Visit(result); + context.visits.put(source, visit); + filler.fill(source, result); + visit.inProgress = false; + return result; + } finally { + exitFlattening(context); + } + } + + private static Object flattenValue(Object value) { + return value == null ? null : eventConverter.convert(value).to(Object.class); + } + + @SuppressWarnings("unchecked") + private static void fillCollection(Object source, Object result) { + Collection<Object> target = (Collection<Object>) result; + for (Object element : (Collection<?>) source) { + target.add(flattenValue(element)); + } + } + + @SuppressWarnings("unchecked") + private static void fillMap(Object source, Object result) { + Map<Object, Object> target = (Map<Object, Object>) result; + for (Map.Entry<?, ?> entry : ((Map<?, ?>) source).entrySet()) { + target.put(flattenValue(entry.getKey()), flattenValue(entry.getValue())); + } + } + + private static void fillArray(Object source, Object result) { + Object[] target = (Object[]) result; + Object[] elements = (Object[]) source; + for (int i = 0; i < elements.length; i++) { + target[i] = flattenValue(elements[i]); + } + } + + @SuppressWarnings("unchecked") + private static void fillDto(Object source, Object result) { + Map<String, Object> target = (Map<String, Object>) result; + Map<String, Object> fields = shallowConverter.convert(source).sourceAsDTO().to(MAP_WITH_STRING_KEYS); + for (Map.Entry<String, Object> field : fields.entrySet()) { + target.put(field.getKey(), flattenValue(field.getValue())); + } } static Object attemptRecovery(Object o, Type target) { @@ -216,14 +323,15 @@ public class EventConverter { if (errors.contains(o)) { if(_log.isWarnEnabled()) { - _log.warn("The map {} repeatedly failed conversion to type {}. This event cannot be converted", o, target); + // The map may be cyclic, so don't use its toString + _log.warn("A map of type {} repeatedly failed conversion to type {}. This event cannot be converted", o.getClass(), target); } return ConverterFunction.CANNOT_HANDLE; } try { if(_log.isWarnEnabled()) { - _log.warn("The map {} failed conversion to type {}. Convertsion will be reattempted treating the target as a DTO", o, target); + _log.warn("A map of type {} failed conversion to type {}. Conversion will be reattempted treating the target as a DTO", o.getClass(), target); } errors.add(o); return eventConverter.convert(o).targetAsDTO().to(target); @@ -234,10 +342,34 @@ public class EventConverter { return ConverterFunction.CANNOT_HANDLE; } + @SuppressWarnings("rawtypes") + private static final AtomicReferenceFieldUpdater<EventConverter, Promise> FLATTENED = AtomicReferenceFieldUpdater + .newUpdater(EventConverter.class, Promise.class, "flattened"); + + /** + * The flattened form of an event, and whether it contains a cycle + */ + private static final class Flattened { + final Map<String, Object> data; + final boolean cyclic; + + Flattened(Map<String, Object> data, boolean cyclic) { + this.data = data; + this.cyclic = cyclic; + } + } + private final Object originalEvent; private final CustomEventConverter custom; - private Map<String, Object> untypedEventDataForFiltering; + // Resolved by the first thread that needs the flattened event, so that an event + // which cannot be flattened fails once rather than once per handler. The field + // has to be volatile for the updater. + private volatile Promise<Flattened> flattened; + + // Only used for typed conversions, which are all made by the delivery thread. A + // stale value only costs another failed attempt. + private boolean typedConversionOverflowed; private EventConverter(Object event, CustomEventConverter custom) { this.originalEvent = event; @@ -247,7 +379,7 @@ public class EventConverter { private EventConverter(Map<String, Object> untypedEvent, CustomEventConverter custom) { this.originalEvent = untypedEvent; this.custom = custom; - this.untypedEventDataForFiltering = untypedEvent; + this.flattened = Promises.resolved(new Flattened(untypedEvent, false)); } public static EventConverter forTypedEvent(Object event, CustomEventConverter custom) { @@ -259,41 +391,123 @@ public class EventConverter { } public boolean applyFilter(Filter f) { - Map<String, ?> toTest; - if (untypedEventDataForFiltering == null) { - toTest = toUntypedEvent(); - } else { - toTest = untypedEventDataForFiltering; + Map<String, ?> toTest = tryUntypedEvent(); + // A filter cannot match an event with no untyped form + return toTest != null && f.matches(toTest); + } + + /** + * @return The untyped form of this event. If the event contains a cycle then + * the Maps, Lists and Sets of the result do too. + * @throws ConversionException if the event cannot be converted. The same + * exception is thrown each time, as the conversion is not repeated. + */ + public Map<String, Object> toUntypedEvent() { + Promise<Flattened> promise = flattened; + if (promise == null) { + promise = flatten(); + } + try { + return promise.getValue().data; + } catch (InvocationTargetException ite) { + Throwable cause = ite.getCause(); + throw cause instanceof ConversionException ? (ConversionException) cause + : new ConversionException("Unable to convert an event of type " + originalEvent.getClass().getName(), cause); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + throw new ConversionException("Interrupted while converting an event of type " + originalEvent.getClass().getName(), ie); + } + } + + /** + * @return The untyped form of this event, or null if it cannot be converted. + * The failure is logged by the first attempt, so callers need not. + */ + Map<String, Object> tryUntypedEvent() { + try { + return toUntypedEvent(); + } catch (ConversionException ce) { + return null; + } + } + + @SuppressWarnings("unchecked") + private Promise<Flattened> flatten() { + Deferred<Flattened> deferred = new Deferred<>(); + if (!FLATTENED.compareAndSet(this, null, deferred.getPromise())) { + // Another thread is converting the event, or has finished + return (Promise<Flattened>) FLATTENED.get(this); } - return f.matches(toTest); + try { + deferred.resolve(flattenOriginalEvent()); + } catch (Throwable t) { + if (t instanceof Error && !(t instanceof StackOverflowError)) { + // Don't leave other threads waiting for a result that will never arrive + deferred.fail(t); + throw (Error) t; + } + Class<?> eventType = originalEvent.getClass(); + // The event may be cyclic, so don't use its toString + _log.warn("The event of type {} cannot be converted to its untyped form. It will not be delivered to untyped handlers, " + + "or be matched by event filters or the event monitor.", eventType.getName(), t); + deferred.fail(new ConversionException("Unable to convert an event of type " + eventType.getName() + " to its untyped form", t)); + } + return deferred.getPromise(); } - public Map<String, Object> toUntypedEvent() { - if (untypedEventDataForFiltering == null) { - if(custom != null) { - if(_log.isDebugEnabled()) { - _log.debug("Using custom converter to convert {} to untyped data", originalEvent); - } - untypedEventDataForFiltering = custom.toUntypedEvent(originalEvent); - if(untypedEventDataForFiltering == null) { - if(_log.isDebugEnabled()) { - _log.debug("Custom event converter could not convert event {}. Falling back to built-in conversion", - originalEvent); - } - untypedEventDataForFiltering = eventConverter.convert(originalEvent).sourceAsDTO().to(MAP_WITH_STRING_KEYS); - } else { - return untypedEventDataForFiltering; - } - } else { - if(_log.isTraceEnabled()) { - _log.trace("Converting {} to untyped data", originalEvent); - } - untypedEventDataForFiltering = eventConverter.convert(originalEvent).sourceAsDTO().to(MAP_WITH_STRING_KEYS); - } - + private Flattened flattenOriginalEvent() { + if (custom != null) { + if(_log.isDebugEnabled()) { + _log.debug("Using custom converter to convert an event of type {} to untyped data", originalEvent.getClass()); + } + Map<String, Object> untyped = custom.toUntypedEvent(originalEvent); + if (untyped != null) { + return new Flattened(untyped, false); + } + if(_log.isDebugEnabled()) { + _log.debug("Custom event converter could not convert an event of type {}. Falling back to built-in conversion", + originalEvent.getClass()); + } + } else if(_log.isTraceEnabled()) { + _log.trace("Converting an event of type {} to untyped data", originalEvent.getClass()); + } + + // The context is shared with the nested conversions, and tells us about cycles + FlatteningContext context = enterFlattening(); + try { + Object result = eventConverter.convert(originalEvent).to(Object.class); + if (!(result instanceof Map)) { + throw new ConversionException("An event of type " + originalEvent.getClass().getName() + " cannot be converted to a Map"); + } + return new Flattened(withStringKeys((Map<?, ?>) result), context.cyclic); + } finally { + exitFlattening(context); + } + } + + @SuppressWarnings("unchecked") + private static Map<String, Object> withStringKeys(Map<?, ?> map) { + if (map.keySet().stream().allMatch(String.class::isInstance)) { + return (Map<String, Object>) map; + } + Map<String, Object> result = new LinkedHashMap<>(); + for (Map.Entry<?, ?> entry : map.entrySet()) { + result.put(eventConverter.convert(entry.getKey()).to(String.class), entry.getValue()); + } + return result; + } + + private boolean isKnownCyclic() { + Promise<Flattened> promise = flattened; + if (promise == null || !promise.isDone()) { + return false; + } + try { + return promise.getValue().cyclic; + } catch (InvocationTargetException | InterruptedException e) { + return false; } - return untypedEventDataForFiltering; } @SuppressWarnings("unchecked") @@ -306,13 +520,14 @@ public class EventConverter { } } else if (rawType.isInstance(originalEvent) && rawType == genericTarget) { if (_log.isDebugEnabled()) { - _log.debug("No need to convert {} to {} as it is already an instance", originalEvent, rawType); + _log.debug("No need to convert an event of type {} to {} as it is already an instance", originalEvent.getClass(), rawType); } + // This is also what keeps any cycles in the event intact return (T) originalEvent; } if (custom != null) { if (_log.isDebugEnabled()) { - _log.debug("Using custom converter to convert {} to {}", originalEvent, genericTarget); + _log.debug("Using custom converter to convert an event of type {} to {}", originalEvent.getClass(), genericTarget); } Object result = custom.toTypedEvent(originalEvent, typeData); if (result != null) { @@ -320,16 +535,31 @@ public class EventConverter { } if (_log.isDebugEnabled()) { _log.debug( - "Custom event converter could not convert event {} to {}. Falling back to built-in conversion", - originalEvent, genericTarget); - } - return eventConverter.convert(originalEvent).targetAsDTO().to(genericTarget); - } else { - if (_log.isDebugEnabled()) { - _log.debug("Converting {} to {}", originalEvent, genericTarget); + "Custom event converter could not convert an event of type {} to {}. Falling back to built-in conversion", + originalEvent.getClass(), genericTarget); } + } else if (_log.isDebugEnabled()) { + _log.debug("Converting an event of type {} to {}", originalEvent.getClass(), genericTarget); + } + return convertToTyped(genericTarget); + } + + private <T> T convertToTyped(Type genericTarget) { + if (typedConversionOverflowed || isKnownCyclic()) { + throw cyclicConversion(genericTarget, null); + } + try { return eventConverter.convert(originalEvent).targetAsDTO().to(genericTarget); + } catch (StackOverflowError so) { + // The converter has no handling for cycles when it builds DTOs + typedConversionOverflowed = true; + throw cyclicConversion(genericTarget, so); } } -} \ No newline at end of file + private IllegalArgumentException cyclicConversion(Type genericTarget, Throwable cause) { + return new IllegalArgumentException("An event of type " + originalEvent.getClass().getName() + + " contains a cycle and cannot be converted to " + genericTarget.getTypeName(), cause); + } + +} diff --git a/org.apache.aries.typedevent.bus/src/main/java/org/apache/aries/typedevent/bus/impl/EventTask.java b/org.apache.aries.typedevent.bus/src/main/java/org/apache/aries/typedevent/bus/impl/EventTask.java index 2681a10..54147ef 100644 --- a/org.apache.aries.typedevent.bus/src/main/java/org/apache/aries/typedevent/bus/impl/EventTask.java +++ b/org.apache.aries.typedevent.bus/src/main/java/org/apache/aries/typedevent/bus/impl/EventTask.java @@ -27,9 +27,10 @@ abstract class EventTask { public void notifyListener() { try { unsafeNotify(); - } catch (Exception e) { + } catch (Exception | StackOverflowError e) { // TODO blacklisting? - _log.error("The event delivery failed for task type.", getClass(), e); + // An overflow must not end the delivery thread, or no later event would be delivered + _log.error("The event delivery failed for task type {}.", getClass(), e); } } diff --git a/org.apache.aries.typedevent.bus/src/main/java/org/apache/aries/typedevent/bus/impl/MonitorEventTask.java b/org.apache.aries.typedevent.bus/src/main/java/org/apache/aries/typedevent/bus/impl/MonitorEventTask.java index 3ea41db..c7890fe 100644 --- a/org.apache.aries.typedevent.bus/src/main/java/org/apache/aries/typedevent/bus/impl/MonitorEventTask.java +++ b/org.apache.aries.typedevent.bus/src/main/java/org/apache/aries/typedevent/bus/impl/MonitorEventTask.java @@ -17,6 +17,8 @@ package org.apache.aries.typedevent.bus.impl; +import java.util.Map; + import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -40,6 +42,12 @@ class MonitorEventTask extends EventTask { if(_log.isTraceEnabled()) { _log.trace("Distributing event to the event monitor"); } - monitorImpl.event(eventType, eventData.toUntypedEvent()); + Map<String, Object> untyped = eventData.tryUntypedEvent(); + if (untyped == null) { + // Already logged when the conversion failed + _log.debug("The event for topic {} has no untyped form and is not passed to the event monitor", eventType); + return; + } + monitorImpl.event(eventType, untyped); } } \ No newline at end of file diff --git a/org.apache.aries.typedevent.bus/src/main/java/org/apache/aries/typedevent/bus/impl/UnhandledEventTask.java b/org.apache.aries.typedevent.bus/src/main/java/org/apache/aries/typedevent/bus/impl/UnhandledEventTask.java index 427d36e..e50837f 100644 --- a/org.apache.aries.typedevent.bus/src/main/java/org/apache/aries/typedevent/bus/impl/UnhandledEventTask.java +++ b/org.apache.aries.typedevent.bus/src/main/java/org/apache/aries/typedevent/bus/impl/UnhandledEventTask.java @@ -17,6 +17,8 @@ package org.apache.aries.typedevent.bus.impl; +import java.util.Map; + import org.osgi.service.typedevent.UnhandledEventHandler; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -41,6 +43,11 @@ class UnhandledEventTask extends EventTask { if(_log.isDebugEnabled()) { _log.debug("Distributing event data {} to the unhandled event handler {}", eventData, eventProcessor); } - eventProcessor.notifyUnhandled(topic, eventData.toUntypedEvent()); + Map<String, Object> untyped = eventData.tryUntypedEvent(); + if (untyped == null) { + _log.debug("The event for topic {} has no untyped form and is not delivered to the unhandled event handler", topic); + return; + } + eventProcessor.notifyUnhandled(topic, untyped); } } \ No newline at end of file diff --git a/org.apache.aries.typedevent.bus/src/main/java/org/apache/aries/typedevent/bus/impl/UntypedEventTask.java b/org.apache.aries.typedevent.bus/src/main/java/org/apache/aries/typedevent/bus/impl/UntypedEventTask.java index 0be4fe8..89997fd 100644 --- a/org.apache.aries.typedevent.bus/src/main/java/org/apache/aries/typedevent/bus/impl/UntypedEventTask.java +++ b/org.apache.aries.typedevent.bus/src/main/java/org/apache/aries/typedevent/bus/impl/UntypedEventTask.java @@ -17,6 +17,8 @@ package org.apache.aries.typedevent.bus.impl; +import java.util.Map; + import org.osgi.service.typedevent.UntypedEventHandler; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -41,6 +43,11 @@ class UntypedEventTask extends EventTask { if(_log.isDebugEnabled()) { _log.debug("Distributing event data {} to the untyped handler {}", eventData, eventProcessor); } - eventProcessor.notifyUntyped(topic, eventData.toUntypedEvent()); + Map<String, Object> untyped = eventData.tryUntypedEvent(); + if (untyped == null) { + _log.debug("The event for topic {} has no untyped form and is not delivered to the untyped handler", topic); + return; + } + eventProcessor.notifyUntyped(topic, untyped); } } \ No newline at end of file diff --git a/org.apache.aries.typedevent.bus/src/test/java/org/apache/aries/typedevent/bus/impl/EventConverterCycleTest.java b/org.apache.aries.typedevent.bus/src/test/java/org/apache/aries/typedevent/bus/impl/EventConverterCycleTest.java new file mode 100644 index 0000000..dc8ced1 --- /dev/null +++ b/org.apache.aries.typedevent.bus/src/test/java/org/apache/aries/typedevent/bus/impl/EventConverterCycleTest.java @@ -0,0 +1,352 @@ +/* + * 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.aries.typedevent.bus.impl; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotSame; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.IdentityHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +import org.apache.aries.typedevent.bus.spi.CustomEventConverter; +import org.apache.aries.typedevent.bus.spi.TypeData; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; +import org.osgi.framework.Filter; +import org.osgi.util.converter.ConversionException; + +/** + * Events that are not trees, because they contain a cycle, must not break the + * event bus + */ +public class EventConverterCycleTest { + + public static class CyclicEvent { + public String name; + public CyclicEvent other; + public List<Object> list; + public Set<Object> set; + public Map<String, Object> map; + public Object[] array; + } + + public static class OtherCyclicEvent { + public String name; + public OtherCyclicEvent other; + } + + public static class Inner { + public String message; + } + + public static class SharedEvent { + public Inner first; + public Inner second; + public List<Inner> list; + } + + public static class InnerCopy { + public String message; + } + + public static class SharedEventCopy { + public InnerCopy first; + public InnerCopy second; + public List<InnerCopy> list; + } + + private static CyclicEvent event(String name) { + CyclicEvent event = new CyclicEvent(); + event.name = name; + return event; + } + + private static TypeData typeData(Class<?> type) { + return new TypeData(type, Collections.emptyMap()); + } + + @SuppressWarnings("unchecked") + private static Map<String, Object> asMap(Object o) { + return (Map<String, Object>) o; + } + + @Test + public void testSelfReference() { + CyclicEvent a = event("a"); + a.other = a; + + Map<String, Object> flat = EventConverter.forTypedEvent(a, null).toUntypedEvent(); + + assertEquals("a", flat.get("name")); + assertSame(flat, flat.get("other")); + } + + @Test + public void testTwoObjectCycle() { + CyclicEvent a = event("a"); + CyclicEvent b = event("b"); + a.other = b; + b.other = a; + + Map<String, Object> flat = EventConverter.forTypedEvent(a, null).toUntypedEvent(); + + Map<String, Object> flatB = asMap(flat.get("other")); + assertEquals("b", flatB.get("name")); + assertSame(flat, flatB.get("other")); + } + + @Test + public void testCycleThroughList() { + CyclicEvent a = event("a"); + a.list = new ArrayList<>(); + a.list.add(a); + a.list.add("end"); + + Map<String, Object> flat = EventConverter.forTypedEvent(a, null).toUntypedEvent(); + + List<?> flatList = (List<?>) flat.get("list"); + assertSame(flat, flatList.get(0)); + assertEquals("end", flatList.get(1)); + } + + @Test + public void testCycleThroughSet() { + CyclicEvent a = event("a"); + a.set = new LinkedHashSet<>(); + a.set.add(a); + + Map<String, Object> flat = EventConverter.forTypedEvent(a, null).toUntypedEvent(); + + Set<?> flatSet = (Set<?>) flat.get("set"); + assertEquals(1, flatSet.size()); + assertSame(flat, flatSet.iterator().next()); + } + + @Test + public void testCycleThroughMap() { + CyclicEvent a = event("a"); + a.map = new HashMap<>(); + a.map.put("self", a); + + Map<String, Object> flat = EventConverter.forTypedEvent(a, null).toUntypedEvent(); + + assertSame(flat, asMap(flat.get("map")).get("self")); + } + + @Test + public void testCycleThroughArray() { + CyclicEvent a = event("a"); + a.array = new Object[] { a, "end" }; + + Map<String, Object> flat = EventConverter.forTypedEvent(a, null).toUntypedEvent(); + + Object[] flatArray = (Object[]) flat.get("array"); + assertSame(flat, flatArray[0]); + assertEquals("end", flatArray[1]); + } + + @Test + public void testCycleNotInvolvingTheRoot() { + CyclicEvent a = event("a"); + CyclicEvent b = event("b"); + CyclicEvent c = event("c"); + a.other = b; + b.other = c; + c.other = b; + + Map<String, Object> flat = EventConverter.forTypedEvent(a, null).toUntypedEvent(); + + Map<String, Object> flatB = asMap(flat.get("other")); + Map<String, Object> flatC = asMap(flatB.get("other")); + assertEquals("c", flatC.get("name")); + assertSame(flatB, flatC.get("other")); + } + + @Test + public void testUntypedFormIsOnlyCreatedOnce() { + CyclicEvent a = event("a"); + a.other = a; + + EventConverter converter = EventConverter.forTypedEvent(a, null); + + assertSame(converter.toUntypedEvent(), converter.toUntypedEvent()); + } + + @Test + public void testConcurrentFlattening() throws Exception { + CyclicEvent a = event("a"); + a.other = a; + + EventConverter converter = EventConverter.forTypedEvent(a, null); + + int threads = 8; + ExecutorService executor = Executors.newFixedThreadPool(threads); + try { + CountDownLatch start = new CountDownLatch(1); + List<Future<Map<String, Object>>> results = new ArrayList<>(); + for (int i = 0; i < threads; i++) { + results.add(executor.submit(() -> { + start.await(); + return converter.toUntypedEvent(); + })); + } + start.countDown(); + + Map<Object, Boolean> distinct = new IdentityHashMap<>(); + for (Future<Map<String, Object>> result : results) { + distinct.put(result.get(10, TimeUnit.SECONDS), Boolean.TRUE); + } + assertEquals(1, distinct.size()); + } finally { + executor.shutdownNow(); + } + } + + @Test + public void testSharedObjectIsNotACycle() { + Inner inner = new Inner(); + inner.message = "shared"; + SharedEvent shared = new SharedEvent(); + shared.first = inner; + shared.second = inner; + shared.list = new ArrayList<>(); + shared.list.add(inner); + + EventConverter converter = EventConverter.forTypedEvent(shared, null); + Map<String, Object> flat = converter.toUntypedEvent(); + + assertEquals("shared", asMap(flat.get("first")).get("message")); + assertSame(flat.get("first"), flat.get("second")); + + // A tree-like event can still be converted to another type + SharedEventCopy copy = converter.toTypedEvent(typeData(SharedEventCopy.class)); + assertEquals("shared", copy.first.message); + assertEquals("shared", copy.second.message); + assertEquals("shared", copy.list.get(0).message); + } + + @Test + public void testNestedCollectionsAreStillCopied() { + CyclicEvent a = event("a"); + a.list = new ArrayList<>(); + a.list.add(event("in list")); + + Map<String, Object> flat = EventConverter.forTypedEvent(a, null).toUntypedEvent(); + + Object element = ((List<?>) flat.get("list")).get(0); + assertEquals("in list", asMap(element).get("name")); + assertNotSame(a.list.get(0), element); + } + + @Test + public void testTypedHandlerOfTheSameTypeKeepsTheCycle() { + CyclicEvent a = event("a"); + a.other = a; + + EventConverter converter = EventConverter.forTypedEvent(a, null); + // Flattening first must make no difference + converter.toUntypedEvent(); + + CyclicEvent delivered = converter.toTypedEvent(typeData(CyclicEvent.class)); + + assertSame(a, delivered); + assertSame(delivered, delivered.other); + } + + @Test + public void testConversionOfCycleToOtherTypeFailsFastOnceFlattened() { + CyclicEvent a = event("a"); + a.other = a; + + EventConverter converter = EventConverter.forTypedEvent(a, null); + converter.toUntypedEvent(); + + assertThrows(IllegalArgumentException.class, () -> converter.toTypedEvent(typeData(OtherCyclicEvent.class))); + assertThrows(IllegalArgumentException.class, () -> converter.toTypedEvent(typeData(OtherCyclicEvent.class))); + } + + @Test + public void testConversionOfCycleToOtherTypeFailsCleanly() { + CyclicEvent a = event("a"); + a.other = a; + + EventConverter converter = EventConverter.forTypedEvent(a, null); + + assertThrows(IllegalArgumentException.class, () -> converter.toTypedEvent(typeData(OtherCyclicEvent.class))); + // The failure is remembered, and the event can still be flattened + assertThrows(IllegalArgumentException.class, () -> converter.toTypedEvent(typeData(OtherCyclicEvent.class))); + assertSame(converter.toUntypedEvent(), converter.toUntypedEvent()); + } + + @Test + public void testFailedFlatteningIsNotRepeated() { + CyclicEvent a = event("a"); + CustomEventConverter custom = Mockito.mock(CustomEventConverter.class); + Mockito.when(custom.toUntypedEvent(a)).thenThrow(new IllegalStateException("Cannot flatten")); + + EventConverter converter = EventConverter.forTypedEvent(a, custom); + + ConversionException first = assertThrows(ConversionException.class, converter::toUntypedEvent); + ConversionException second = assertThrows(ConversionException.class, converter::toUntypedEvent); + + assertSame(first, second); + assertEquals(IllegalStateException.class, first.getCause().getClass()); + Mockito.verify(custom, Mockito.times(1)).toUntypedEvent(a); + } + + @Test + public void testFilterDoesNotMatchAnEventWithNoUntypedForm() { + CyclicEvent a = event("a"); + CustomEventConverter custom = Mockito.mock(CustomEventConverter.class); + Mockito.when(custom.toUntypedEvent(a)).thenThrow(new IllegalStateException("Cannot flatten")); + Filter filter = Mockito.mock(Filter.class); + Mockito.when(filter.matches(Mockito.anyMap())).thenReturn(true); + + EventConverter converter = EventConverter.forTypedEvent(a, custom); + + assertFalse(converter.applyFilter(filter)); + Mockito.verifyNoInteractions(filter); + } + + @Test + public void testFilterMatchesAFlattenedCycle() { + CyclicEvent a = event("a"); + a.other = a; + Filter filter = Mockito.mock(Filter.class); + Mockito.when(filter.matches(Mockito.anyMap())).thenReturn(true); + + EventConverter converter = EventConverter.forTypedEvent(a, null); + + assertTrue(converter.applyFilter(filter)); + } +} diff --git a/org.apache.aries.typedevent.bus/src/test/java/org/apache/aries/typedevent/bus/impl/TypedEventBusImplTest.java b/org.apache.aries.typedevent.bus/src/test/java/org/apache/aries/typedevent/bus/impl/TypedEventBusImplTest.java index 4bb1470..bf2617d 100644 --- a/org.apache.aries.typedevent.bus/src/test/java/org/apache/aries/typedevent/bus/impl/TypedEventBusImplTest.java +++ b/org.apache.aries.typedevent.bus/src/test/java/org/apache/aries/typedevent/bus/impl/TypedEventBusImplTest.java @@ -868,4 +868,106 @@ public class TypedEventBusImplTest { }; } + public static class CyclicTestEvent { + public String message; + public CyclicTestEvent other; + } + + public static class CyclicTestEventCopy { + public String message; + public CyclicTestEventCopy other; + } + + /** + * An event that is not a tree must not affect the delivery of later events + */ + @Test + public void testCyclicEventDoesNotAffectLaterEvents() throws InterruptedException { + + String cyclicTopic = CyclicTestEvent.class.getName().replace(".", "/"); + + CyclicTestEvent cyclic = new CyclicTestEvent(); + cyclic.message = "boo"; + cyclic.other = cyclic; + + // A typed handler of the same type, which needs no conversion + Map<String, Object> serviceProperties = new HashMap<>(); + serviceProperties.put(TYPED_EVENT_TOPICS, cyclicTopic); + serviceProperties.put(TYPED_EVENT_TYPE, CyclicTestEvent.class.getName()); + serviceProperties.put(SERVICE_ID, 42L); + impl.addTypedEventHandler(registeringBundle, handlerA, serviceProperties); + + // A typed handler of another type, which cannot be given a cyclic event + serviceProperties = new HashMap<>(); + serviceProperties.put(TYPED_EVENT_TOPICS, cyclicTopic); + serviceProperties.put(TYPED_EVENT_TYPE, CyclicTestEventCopy.class.getName()); + serviceProperties.put(SERVICE_ID, 43L); + impl.addTypedEventHandler(registeringBundle, handlerB, serviceProperties); + + // An untyped handler with a filter, which is evaluated by the sending thread + serviceProperties = new HashMap<>(); + serviceProperties.put(TYPED_EVENT_TOPICS, cyclicTopic); + serviceProperties.put(TYPED_EVENT_FILTER, "(message=boo)"); + serviceProperties.put(SERVICE_ID, 44L); + impl.addUntypedEventHandler(untypedHandlerA, serviceProperties); + + // A handler for the later event + serviceProperties = new HashMap<>(); + serviceProperties.put(TYPED_EVENT_TOPICS, TEST_EVENT_TOPIC); + serviceProperties.put(SERVICE_ID, 45L); + impl.addUntypedEventHandler(untypedHandlerB, serviceProperties); + + impl.deliver(cyclic); + + assertTrue(semA.tryAcquire(1, TimeUnit.SECONDS)); + Mockito.verify(handlerA).notify(Mockito.eq(cyclicTopic), Mockito.same(cyclic)); + + assertTrue(untypedSemA.tryAcquire(1, TimeUnit.SECONDS)); + Mockito.verify(untypedHandlerA).notifyUntyped(Mockito.eq(cyclicTopic), + Mockito.argThat(m -> m != null && m.get("other") == m)); + + assertFalse(semB.tryAcquire(500, TimeUnit.MILLISECONDS)); + + TestEvent event = new TestEvent(); + event.message = "after"; + impl.deliver(event); + + assertTrue(untypedSemB.tryAcquire(1, TimeUnit.SECONDS)); + Mockito.verify(untypedHandlerB).notifyUntyped(Mockito.eq(TEST_EVENT_TOPIC), + Mockito.argThat(isUntypedTestEventWithMessage("after"))); + } + + /** + * A handler that overflows the stack must not stop delivery of later events + */ + @Test + public void testStackOverflowInHandlerDoesNotStopDelivery() throws InterruptedException { + + Mockito.doThrow(new StackOverflowError()).doAnswer(i -> { + semA.release(); + return null; + }).when(handlerA).notify(Mockito.anyString(), Mockito.any()); + + Map<String, Object> serviceProperties = new HashMap<>(); + serviceProperties.put(TYPED_EVENT_TOPICS, TEST_EVENT_TOPIC); + serviceProperties.put(TYPED_EVENT_TYPE, TestEvent.class.getName()); + serviceProperties.put(SERVICE_ID, 42L); + impl.addTypedEventHandler(registeringBundle, handlerA, serviceProperties); + + TestEvent event = new TestEvent(); + event.message = "first"; + impl.deliver(event); + + // The first delivery fails + assertFalse(semA.tryAcquire(500, TimeUnit.MILLISECONDS)); + + event = new TestEvent(); + event.message = "second"; + impl.deliver(event); + + assertTrue(semA.tryAcquire(1, TimeUnit.SECONDS)); + Mockito.verify(handlerA).notify(Mockito.eq(TEST_EVENT_TOPIC), + Mockito.argThat(isTestEventWithMessage("second"))); + } + } diff --git a/org.apache.aries.typedevent.bus/src/test/java/org/apache/aries/typedevent/bus/osgi/TypedEventMonitorIntegrationTest.java b/org.apache.aries.typedevent.bus/src/test/java/org/apache/aries/typedevent/bus/osgi/TypedEventMonitorIntegrationTest.java index 0b6ac31..3eba5bd 100644 --- a/org.apache.aries.typedevent.bus/src/test/java/org/apache/aries/typedevent/bus/osgi/TypedEventMonitorIntegrationTest.java +++ b/org.apache.aries.typedevent.bus/src/test/java/org/apache/aries/typedevent/bus/osgi/TypedEventMonitorIntegrationTest.java @@ -27,6 +27,7 @@ import java.util.Arrays; import java.util.Dictionary; import java.util.Hashtable; import java.util.List; +import java.util.Map; import java.util.stream.Collectors; import org.apache.aries.typedevent.bus.common.TestEvent; @@ -589,4 +590,42 @@ public class TypedEventMonitorIntegrationTest extends AbstractIntegrationTest { assertEquals("bar", events.get(0).eventData.get("message")); assertEquals("foobar", events.get(1).eventData.get("message")); } + + public static class CyclicEvent { + public String message; + public CyclicEvent other; + } + + /** + * Tests that an event which is not a tree is monitored, and does + * not prevent later events from being delivered + */ + @Test + public void testCyclicEventMonitoring(@InjectService TypedEventMonitor monitor, + @InjectService TypedEventBus eventBus) throws Exception { + + Promise<List<MonitorEvent>> eventsPromise = monitor.monitorEvents() + .limit(2) + .collect(Collectors.toList()); + + CyclicEvent cyclic = new CyclicEvent(); + cyclic.message = "cyclic"; + cyclic.other = cyclic; + eventBus.deliver(cyclic); + + TestEvent event = new TestEvent(); + event.message = "after"; + eventBus.deliver(event); + + List<MonitorEvent> events = eventsPromise.timeout(5000).getValue(); + + assertEquals(2, events.size()); + + assertEquals(CyclicEvent.class.getName().replace('.', '/'), events.get(0).topic); + assertEquals("cyclic", events.get(0).eventData.get("message")); + assertEquals("cyclic", ((Map<?, ?>) events.get(0).eventData.get("other")).get("message")); + + assertEquals(TEST_EVENT_TOPIC, events.get(1).topic); + assertEquals("after", events.get(1).eventData.get("message")); + } } diff --git a/org.apache.aries.typedevent.remote/org.apache.aries.typedevent.remote.spi/pom.xml b/org.apache.aries.typedevent.remote/org.apache.aries.typedevent.remote.spi/pom.xml index 6a384cb..a9bbe3e 100644 --- a/org.apache.aries.typedevent.remote/org.apache.aries.typedevent.remote.spi/pom.xml +++ b/org.apache.aries.typedevent.remote/org.apache.aries.typedevent.remote.spi/pom.xml @@ -24,7 +24,19 @@ <properties> <no.test.jar>true</no.test.jar> </properties> - + + <dependencyManagement> + <dependencies> + <dependency> + <groupId>org.apache.aries.typedevent</groupId> + <artifactId>typedevent-test-bom</artifactId> + <version>${project.version}</version> + <type>pom</type> + <scope>import</scope> + </dependency> + </dependencies> + </dependencyManagement> + <dependencies> <dependency> <groupId>org.osgi</groupId> @@ -52,6 +64,11 @@ <artifactId>org.apache.aries.typedevent.remote.api</artifactId> <version>1.1.0-SNAPSHOT</version> </dependency> + <dependency> + <groupId>org.junit.jupiter</groupId> + <artifactId>junit-jupiter</artifactId> + <scope>test</scope> + </dependency> </dependencies> <build> diff --git a/org.apache.aries.typedevent.remote/org.apache.aries.typedevent.remote.spi/src/main/java/org/apache/aries/typedevent/remote/spi/RemoteEventMonitorImpl.java b/org.apache.aries.typedevent.remote/org.apache.aries.typedevent.remote.spi/src/main/java/org/apache/aries/typedevent/remote/spi/RemoteEventMonitorImpl.java index 61ef87f..8f05dc6 100644 --- a/org.apache.aries.typedevent.remote/org.apache.aries.typedevent.remote.spi/src/main/java/org/apache/aries/typedevent/remote/spi/RemoteEventMonitorImpl.java +++ b/org.apache.aries.typedevent.remote/org.apache.aries.typedevent.remote.spi/src/main/java/org/apache/aries/typedevent/remote/spi/RemoteEventMonitorImpl.java @@ -16,19 +16,20 @@ */ package org.apache.aries.typedevent.remote.spi; +import static java.util.Collections.newSetFromMap; import static org.apache.aries.typedevent.remote.api.RemoteEventConstants.REMOTE_EVENT_MARKER; import java.time.Instant; -import java.util.AbstractMap; import java.util.Arrays; +import java.util.IdentityHashMap; import java.util.List; import java.util.Map; import java.util.Map.Entry; +import java.util.Set; import java.util.SortedMap; import java.util.TreeMap; import java.util.regex.Pattern; import java.util.stream.Collectors; -import java.util.stream.Stream; import org.apache.aries.typedevent.remote.api.FilterDTO; import org.apache.aries.typedevent.remote.api.RemoteEventMonitor; @@ -112,55 +113,71 @@ public class RemoteEventMonitorImpl implements RemoteEventMonitor { // We use a TreeMap to ensure predictable ordering of keys // This is important for the regex matching contract. - SortedMap<String, Object> toFilter = new TreeMap<>(); - - // Using a collector blew up with null values, even though they are - // supported by the TreeMap - event.eventData.entrySet().stream() - .flatMap(e -> flatten("", e)) - .forEach(e -> toFilter.put(e.getKey(), e.getValue())); + SortedMap<String, Object> toFilter = flatten(event.eventData); toFilter.put("-topic", event.topic); toFilter.put("-publishType", event.publishType); - StringBuilder eventText = new StringBuilder(); + // Event data that is not a tree can have values that print or match + // endlessly. Such an event cannot match any filter that looks at them. + try { + StringBuilder eventText = new StringBuilder(); - if (filterPairs.stream().anyMatch(p -> p.regex != null)) { - toFilter.forEach((k, v) -> { - eventText.append(k).append(':').append(v).append(','); - }); - } + if (filterPairs.stream().anyMatch(p -> p.regex != null)) { + toFilter.forEach((k, v) -> { + eventText.append(k).append(':').append(v).append(','); + }); + } - // If a FilterDTO contains both LDAP and regular expressions, then both must match. - return filterPairs.stream().anyMatch(p -> - (p.ldap == null || p.ldap.matches(toFilter)) && - (p.regex == null || p.regex.matcher(eventText).find()) - ); + // If a FilterDTO contains both LDAP and regular expressions, then both must match. + return filterPairs.stream().anyMatch(p -> + (p.ldap == null || p.ldap.matches(toFilter)) && + (p.regex == null || p.regex.matcher(eventText).find()) + ); + } catch (StackOverflowError so) { + return false; + } }; } - private Stream<Entry<String, Object>> flatten(String parentScope, - Entry<String, Object> entry) { - - if (entry.getValue() instanceof Map) { - - String keyPrefix = parentScope + entry.getKey() + "."; + /** + * Flatten nested maps into a single map with keys of the form x.y.z. The + * entries are sorted to give a predictable ordering of keys, which is + * important for the regex matching contract. + * <p> + * A nested map that refers back to one of the maps containing it (which the + * event bus allows for events that are not trees) contributes no further entries. + */ + static SortedMap<String, Object> flatten(Map<String, Object> eventData) { + SortedMap<String, Object> result = new TreeMap<>(); + flatten("", eventData, newSetFromMap(new IdentityHashMap<>()), result); + return result; + } - @SuppressWarnings("unchecked") - Map<String, Object> subMap = (Map<String, Object>) entry.getValue(); + private static void flatten(String scope, Map<String, Object> map, + Set<Object> enclosing, Map<String, Object> result) { - // Recursively flatten maps that are inside our map - return subMap.entrySet().stream() - .flatMap(e -> flatten(keyPrefix, e)); - } else if(parentScope.isEmpty()) { - // Fast path for top-level entries - return Stream.of(entry); - } else { - // Map the key of a nested entry into x.y.z - return Stream.of(new AbstractMap.SimpleEntry<>( - parentScope + entry.getKey(), entry.getValue())); + if (!enclosing.add(map)) { + return; } + try { + for (Entry<String, Object> entry : map.entrySet()) { + Object value = entry.getValue(); + if (value instanceof Map) { + @SuppressWarnings("unchecked") + Map<String, Object> subMap = (Map<String, Object>) value; + + // Recursively flatten maps that are inside our map + flatten(scope + entry.getKey() + ".", subMap, enclosing, result); + } else { + // Map the key of a nested entry into x.y.z + result.put(scope + entry.getKey(), value); + } + } + } finally { + enclosing.remove(map); + } } } diff --git a/org.apache.aries.typedevent.remote/org.apache.aries.typedevent.remote.spi/src/test/java/org/apache/aries/typedevent/remote/spi/RemoteEventMonitorImplFlattenTest.java b/org.apache.aries.typedevent.remote/org.apache.aries.typedevent.remote.spi/src/test/java/org/apache/aries/typedevent/remote/spi/RemoteEventMonitorImplFlattenTest.java new file mode 100644 index 0000000..26800e7 --- /dev/null +++ b/org.apache.aries.typedevent.remote/org.apache.aries.typedevent.remote.spi/src/test/java/org/apache/aries/typedevent/remote/spi/RemoteEventMonitorImplFlattenTest.java @@ -0,0 +1,92 @@ +/* + * 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.aries.typedevent.remote.spi; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.SortedMap; + +import org.junit.jupiter.api.Test; + +public class RemoteEventMonitorImplFlattenTest { + + @Test + public void testNestedMapsAreFlattened() { + Map<String, Object> inner = new HashMap<>(); + inner.put("x", 1); + Map<String, Object> eventData = new LinkedHashMap<>(); + eventData.put("a", "top"); + eventData.put("b", inner); + eventData.put("c", null); + + SortedMap<String, Object> flat = RemoteEventMonitorImpl.flatten(eventData); + + assertEquals(3, flat.size()); + assertEquals("top", flat.get("a")); + assertEquals(1, flat.get("b.x")); + assertEquals(null, flat.get("c")); + } + + @Test + public void testMapReferringBackToItselfTerminates() { + Map<String, Object> eventData = new LinkedHashMap<>(); + eventData.put("a", "top"); + eventData.put("self", eventData); + + SortedMap<String, Object> flat = RemoteEventMonitorImpl.flatten(eventData); + + assertEquals(1, flat.size()); + assertEquals("top", flat.get("a")); + } + + @Test + public void testCycleBetweenNestedMapsTerminates() { + Map<String, Object> first = new LinkedHashMap<>(); + Map<String, Object> second = new LinkedHashMap<>(); + first.put("name", "first"); + first.put("next", second); + second.put("name", "second"); + second.put("next", first); + + Map<String, Object> eventData = new LinkedHashMap<>(); + eventData.put("start", first); + + SortedMap<String, Object> flat = RemoteEventMonitorImpl.flatten(eventData); + + assertEquals(2, flat.size()); + assertEquals("first", flat.get("start.name")); + assertEquals("second", flat.get("start.next.name")); + } + + @Test + public void testSharedMapIsNotACycle() { + Map<String, Object> shared = new HashMap<>(); + shared.put("x", 1); + Map<String, Object> eventData = new LinkedHashMap<>(); + eventData.put("first", shared); + eventData.put("second", shared); + + SortedMap<String, Object> flat = RemoteEventMonitorImpl.flatten(eventData); + + assertEquals(2, flat.size()); + assertEquals(1, flat.get("first.x")); + assertEquals(1, flat.get("second.x")); + } +}
