remove of NamedEventSource

Signed-off-by: Attila Mészáros <csviri@gmail.com>

Author: csviri

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/health/ControllerHealthInfo.java

@@ -4,6 +4,7 @@
 import java.util.stream.Collectors;
 
 import io.javaoperatorsdk.operator.processing.event.EventSourceManager;
+import io.javaoperatorsdk.operator.processing.event.source.EventSource;
 
 @SuppressWarnings("rawtypes")
 public class ControllerHealthInfo {
@@ -15,21 +16,21 @@ public ControllerHealthInfo(EventSourceManager eventSourceManager) {
   }
 
   public Map<String, EventSourceHealthIndicator> eventSourceHealthIndicators() {
-    return eventSourceManager.allEventSources().entrySet().stream()
-        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+    return eventSourceManager.allEventSources().stream()
+        .collect(Collectors.toMap(EventSource::name, e -> e));
   }
 
   public Map<String, EventSourceHealthIndicator> unhealthyEventSources() {
-    return eventSourceManager.allEventSources().entrySet().stream()
-        .filter(e -> e.getValue().getStatus() == Status.UNHEALTHY)
-        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+    return eventSourceManager.allEventSources().stream()
+        .filter(e -> e.getStatus() == Status.UNHEALTHY)
+        .collect(Collectors.toMap(EventSource::name, e -> e));
   }
 
   public Map<String, InformerWrappingEventSourceHealthIndicator> informerEventSourceHealthIndicators() {
-    return eventSourceManager.allEventSources().entrySet().stream()
-        .filter(e -> e.getValue() instanceof InformerWrappingEventSourceHealthIndicator)
-        .collect(Collectors.toMap(Map.Entry::getKey,
-            e -> (InformerWrappingEventSourceHealthIndicator) e.getValue()));
+    return eventSourceManager.allEventSources().stream()
+        .filter(e -> e instanceof InformerWrappingEventSourceHealthIndicator)
+        .collect(Collectors.toMap(EventSource::name,
+            e -> (InformerWrappingEventSourceHealthIndicator) e));
 
   }
 
@@ -40,11 +41,11 @@ public Map<String, InformerWrappingEventSourceHealthIndicator> informerEventSour
    *         {@link io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource}.
    */
   public Map<String, InformerWrappingEventSourceHealthIndicator> unhealthyInformerEventSourceHealthIndicators() {
-    return eventSourceManager.allEventSources().entrySet().stream()
-        .filter(e -> e.getValue().getStatus() == Status.UNHEALTHY)
-        .filter(e -> e.getValue() instanceof InformerWrappingEventSourceHealthIndicator)
-        .collect(Collectors.toMap(Map.Entry::getKey,
-            e -> (InformerWrappingEventSourceHealthIndicator) e.getValue()));
+    return eventSourceManager.allEventSources().stream()
+        .filter(e -> e.getStatus() == Status.UNHEALTHY)
+        .filter(e -> e instanceof InformerWrappingEventSourceHealthIndicator)
+        .collect(Collectors.toMap(EventSource::name,
+            e -> (InformerWrappingEventSourceHealthIndicator) e));
   }
 
 }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSourceManager.java

@@ -2,13 +2,11 @@
 
 import java.util.LinkedHashSet;
 import java.util.List;
-import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 import java.util.function.Function;
 import java.util.stream.Collectors;
-import java.util.stream.Stream;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -69,27 +67,23 @@ public void postProcessDefaultEventSourcesAfterProcessorInitializer() {
    */
   @Override
   public synchronized void start() {
-    startEventSource(eventSources.namedControllerResourceEventSource());
+    startEventSource(eventSources.controllerResourceEventSource());
 
     executorServiceManager.boundedExecuteAndWaitForAllToComplete(
-        eventSources.additionalNamedEventSources()
+        eventSources.additionalEventSources()
             .filter(es -> es.priority().equals(EventSourceStartPriority.RESOURCE_STATE_LOADER)),
         this::startEventSource,
         getThreadNamer("start"));
 
     executorServiceManager.boundedExecuteAndWaitForAllToComplete(
-        eventSources.additionalNamedEventSources()
+        eventSources.additionalEventSources()
             .filter(es -> es.priority().equals(EventSourceStartPriority.DEFAULT)),
         this::startEventSource,
         getThreadNamer("start"));
   }
 
-  private static Function<NamedEventSource, String> getThreadNamer(String stage) {
-    return es -> {
-      final var name = es.name();
-      return es.priority() + " " + stage + " -> "
-          + (es.isNameSet() ? name + " " + es.original().getClass() : es.original());
-    };
+  private static Function<EventSource, String> getThreadNamer(String stage) {
+    return es -> es.priority() + " " + stage + " -> " + es.name();
   }
 
   private static Function<NamespaceChangeable, String> getEventSourceThreadNamer(String stage) {
@@ -98,28 +92,26 @@ private static Function<NamespaceChangeable, String> getEventSourceThreadNamer(S
 
   @Override
   public synchronized void stop() {
-    stopEventSource(eventSources.namedControllerResourceEventSource());
+    stopEventSource(eventSources.controllerResourceEventSource());
     executorServiceManager.boundedExecuteAndWaitForAllToComplete(
-        eventSources.additionalNamedEventSources(),
+        eventSources.additionalEventSources(),
         this::stopEventSource,
         getThreadNamer("stop"));
   }
 
   @SuppressWarnings("rawtypes")
-  private void logEventSourceEvent(NamedEventSource eventSource, String event) {
+  private void logEventSourceEvent(EventSource eventSource, String event) {
     if (log.isDebugEnabled()) {
-      if (eventSource.original() instanceof ResourceEventSource source) {
-        log.debug("{} event source {} for {}", event,
-            eventSource.isNameSet() ? eventSource.name() : eventSource,
+      if (eventSource instanceof ResourceEventSource source) {
+        log.debug("{} event source {} for {}", event, eventSource.name(),
             source.resourceType());
       } else {
-        log.debug("{} event source {}", event,
-            eventSource.isNameSet() ? eventSource.name() : eventSource);
+        log.debug("{} event source {}", event, eventSource.name());
       }
     }
   }
 
-  private Void startEventSource(NamedEventSource eventSource) {
+  private Void startEventSource(EventSource eventSource) {
     try {
       logEventSourceEvent(eventSource, "Starting");
       eventSource.start();
@@ -132,7 +124,7 @@ private Void startEventSource(NamedEventSource eventSource) {
     return null;
   }
 
-  private Void stopEventSource(NamedEventSource eventSource) {
+  private Void stopEventSource(EventSource eventSource) {
     try {
       logEventSourceEvent(eventSource, "Stopping");
       eventSource.stop();
@@ -149,13 +141,11 @@ public final synchronized void registerEventSource(EventSource eventSource)
     Objects.requireNonNull(eventSource, "EventSource must not be null");
     try {
       if (eventSource instanceof ManagedInformerEventSource managedInformerEventSource) {
-
         managedInformerEventSource.setConfigurationService(
             controller.getConfiguration().getConfigurationService());
       }
-      final var named = new NamedEventSource(eventSource, eventSource.name());
-      eventSources.add(named);
-      named.setEventHandler(controller.getEventProcessor());
+      eventSources.add(eventSource);
+      eventSource.setEventHandler(controller.getEventProcessor());
     } catch (IllegalStateException | MissingCRDException e) {
       throw e; // leave untouched
     } catch (Exception e) {
@@ -166,8 +156,7 @@ public final synchronized void registerEventSource(EventSource eventSource)
 
   @SuppressWarnings("unchecked")
   public void broadcastOnResourceEvent(ResourceAction action, P resource, P oldResource) {
-    eventSources.additionalNamedEventSources()
-        .map(NamedEventSource::original)
+    eventSources.additionalEventSources()
         .forEach(source -> {
           if (source instanceof ResourceEventAware) {
             var lifecycleAwareES = ((ResourceEventAware<P>) source);
@@ -202,18 +191,12 @@ public void changeNamespaces(Set<String> namespaces) {
 
   public Set<EventSource> getRegisteredEventSources() {
     return eventSources.flatMappedSources()
-        .map(NamedEventSource::original)
-        .collect(Collectors.toCollection(LinkedHashSet::new));
-  }
 
-  public Map<String, EventSource> allEventSources() {
-    return eventSources.allNamedEventSources().collect(Collectors.toMap(NamedEventSource::name,
-        NamedEventSource::original));
+        .collect(Collectors.toCollection(LinkedHashSet::new));
   }
 
-  @SuppressWarnings("unused")
-  public Stream<? extends EventSourceMetadata> getNamedEventSourcesStream() {
-    return eventSources.flatMappedSources();
+  public List<EventSource> allEventSources() {
+    return eventSources.allEventSources().toList();
   }
 
   public ControllerResourceEventSource<P> getControllerResourceEventSource() {
@@ -227,9 +210,9 @@ public <R> List<ResourceEventSource<R, P>> getResourceEventSourcesFor(Class<R> d
   @Override
   public EventSource dynamicallyRegisterEventSource(EventSource eventSource) {
     synchronized (this) {
-      var actual = eventSources.existing(eventSource.name(), eventSource);
+      var actual = eventSources.existing(eventSource);
       if (actual != null) {
-        eventSource = actual.eventSource();
+        eventSource = actual;
       } else {
         registerEventSource(eventSource);
       }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSources.java

@@ -19,12 +19,10 @@
 
 class EventSources<R extends HasMetadata> {
 
-  public static final String CONTROLLER_RESOURCE_EVENT_SOURCE_NAME =
-      "ControllerResourceEventSource";
   public static final String RETRY_RESCHEDULE_TIMER_EVENT_SOURCE_NAME =
       "RetryAndRescheduleTimerEventSource";
 
-  private final ConcurrentNavigableMap<String, Map<String, NamedEventSource>> sources =
+  private final ConcurrentNavigableMap<String, Map<String, EventSource>> sources =
       new ConcurrentSkipListMap<>();
   private final TimerEventSource<R> retryAndRescheduleTimerEventSource = new TimerEventSource<>();
   private ControllerResourceEventSource<R> controllerResourceEventSource;
@@ -42,58 +40,44 @@ TimerEventSource<R> retryEventSource() {
     return retryAndRescheduleTimerEventSource;
   }
 
-  public Stream<NamedEventSource> additionalNamedEventSources() {
-    return Stream.concat(Stream.of(
-        new NamedEventSource(retryAndRescheduleTimerEventSource,
-            RETRY_RESCHEDULE_TIMER_EVENT_SOURCE_NAME)),
-        flatMappedSources());
-  }
-
-  public Stream<NamedEventSource> allNamedEventSources() {
-    return Stream.concat(Stream.of(namedControllerResourceEventSource(),
-        new NamedEventSource(retryAndRescheduleTimerEventSource,
-            RETRY_RESCHEDULE_TIMER_EVENT_SOURCE_NAME)),
+  public Stream<EventSource> allEventSources() {
+    return Stream.concat(
+        Stream.of(controllerResourceEventSource(), retryAndRescheduleTimerEventSource),
         flatMappedSources());
   }
 
   Stream<EventSource> additionalEventSources() {
     return Stream.concat(
         Stream.of(retryEventSource()).filter(Objects::nonNull),
-        flatMappedSources().map(NamedEventSource::original));
-  }
-
-  NamedEventSource namedControllerResourceEventSource() {
-    return new NamedEventSource(controllerResourceEventSource,
-        CONTROLLER_RESOURCE_EVENT_SOURCE_NAME);
+        flatMappedSources());
   }
 
-  Stream<NamedEventSource> flatMappedSources() {
+  Stream<EventSource> flatMappedSources() {
     return sources.values().stream().flatMap(c -> c.values().stream());
   }
 
   public void clear() {
     sources.clear();
   }
 
-  public NamedEventSource existing(String name, EventSource source) {
+  public EventSource existing(EventSource source) {
     final var eventSources = sources.get(keyFor(source));
     if (eventSources == null || eventSources.isEmpty()) {
       return null;
     }
-    return eventSources.get(name);
+    return eventSources.get(source.name());
   }
 
-  public void add(NamedEventSource eventSource) {
+  public void add(EventSource eventSource) {
     final var name = eventSource.name();
-    final var original = eventSource.original();
-    final var existing = existing(name, original);
+    final var existing = existing(eventSource);
     if (existing != null && !eventSource.scopeEquals(existing)) {
-      throw new IllegalArgumentException("Event source " + existing.original()
+      throw new IllegalArgumentException("Event source " + existing
           + " is already registered for the "
-          + keyAsString(getResourceType(original), name)
+          + keyAsString(getResourceType(eventSource), name)
           + " class/name combination");
     }
-    sources.computeIfAbsent(keyFor(original), k -> new HashMap<>()).put(name, eventSource);
+    sources.computeIfAbsent(keyFor(existing), k -> new HashMap<>()).put(name, eventSource);
   }
 
   @SuppressWarnings("rawtypes")
@@ -104,10 +88,6 @@ private Class<?> getResourceType(EventSource source) {
   }
 
   private String keyFor(EventSource source) {
-    if (source instanceof NamedEventSource) {
-      source = ((NamedEventSource) source).original();
-    }
-
     return keyFor(getResourceType(source));
   }
 
@@ -128,7 +108,7 @@ public <S> ResourceEventSource<S, R> get(Class<S> dependentType, String name) {
     }
 
     final var size = sourcesForType.size();
-    NamedEventSource source;
+    EventSource source;
     if (size == 1 && name == null) {
       source = sourcesForType.values().stream().findFirst().orElseThrow();
     } else {
@@ -146,16 +126,15 @@ public <S> ResourceEventSource<S, R> get(Class<S> dependentType, String name) {
       }
     }
 
-    EventSource original = source.original();
-    if (!(original instanceof ResourceEventSource)) {
+    if (!(source instanceof ResourceEventSource)) {
       throw new IllegalArgumentException(source + " associated with "
           + keyAsString(dependentType, name) + " is not a "
           + ResourceEventSource.class.getSimpleName());
     }
-    final var res = (ResourceEventSource<S, R>) original;
+    final var res = (ResourceEventSource<S, R>) source;
     final var resourceClass = res.resourceType();
     if (!resourceClass.isAssignableFrom(dependentType)) {
-      throw new IllegalArgumentException(original + " associated with "
+      throw new IllegalArgumentException(source + " associated with "
           + keyAsString(dependentType, name)
           + " is handling " + resourceClass.getName() + " resources but asked for "
           + dependentType.getName());
@@ -178,7 +157,6 @@ public <S> List<ResourceEventSource<S, R>> getEventSources(Class<S> dependentTyp
     }
 
     return sourcesForType.values().stream()
-        .map(NamedEventSource::original)
         .filter(ResourceEventSource.class::isInstance)
         .map(es -> (ResourceEventSource<S, R>) es)
         .collect(Collectors.toList());