improve: mamed event sources and equality of event sources and their implications

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

Author: csviri

caffeine-bounded-cache-support/src/test/java/io/javaoperatorsdk/operator/processing/event/source/cache/sample/AbstractTestReconciler.java

@@ -1,6 +1,7 @@
 package io.javaoperatorsdk.operator.processing.event.source.cache.sample;
 
 import java.time.Duration;
+import java.util.List;
 import java.util.Map;
 
 import org.slf4j.Logger;
@@ -69,7 +70,7 @@ protected void createConfigMap(P resource, Context<P> context) {
   }
 
   @Override
-  public Map<String, EventSource> prepareEventSources(
+  public List<EventSource> prepareEventSources(
       EventSourceContext<P> context) {
 
     var boundedItemStore =
@@ -82,7 +83,7 @@ public Map<String, EventSource> prepareEventSources(
             Mappers.fromOwnerReference(this instanceof BoundedCacheClusterScopeTestReconciler))
         .build(), context);
 
-    return EventSourceUtils.nameEventSources(es);
+    return List.of(es);
   }
 
   private void ensureStatus(P resource) {

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/EventSourceUtils.java

@@ -5,10 +5,20 @@
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
 import io.javaoperatorsdk.operator.processing.dependent.workflow.Workflow;
+import io.javaoperatorsdk.operator.processing.event.Event;
 import io.javaoperatorsdk.operator.processing.event.source.EventSource;
 import io.javaoperatorsdk.operator.processing.event.source.ResourceEventSource;
+import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource;
 
+// todo cleanup / delete
 public class EventSourceUtils {
+
+  @SuppressWarnings("unchecked")
+  public static  <R extends HasMetadata> List<EventSource> dependentEventSources(EventSourceContext<R> eventSourceContext,DependentResource... dependentResources) {
+      return Arrays.stream(dependentResources)
+              .flatMap(dr-> dr.eventSource(eventSourceContext).stream()).toList();
+  }
+
   /**
    * Utility method to easily create map with generated name for event sources. This is for the use
    * case when the event sources are not access explicitly by name in the reconciler.
@@ -25,15 +35,11 @@ public static Map<String, EventSource> nameEventSources(EventSource... eventSour
   }
 
   @SuppressWarnings("unchecked")
-  public static <K extends HasMetadata> Map<String, EventSource> eventSourcesFromWorkflow(
+  public static <K extends HasMetadata> List<EventSource> eventSourcesFromWorkflow(
       EventSourceContext<K> context,
       Workflow<K> workflow) {
-    Map<String, EventSource> result = new HashMap<>();
-    for (var e : workflow.getDependentResourcesByNameWithoutActivationCondition().entrySet()) {
-      var eventSource = e.getValue().eventSource(context);
-      eventSource.ifPresent(es -> result.put(e.getKey(), (EventSource) es));
-    }
-    return result;
+    return workflow.getDependentResourcesByNameWithoutActivationCondition().stream()
+            .flatMap(dr->dr.eventSource(context).stream()).toList();
   }
 
   @SuppressWarnings("rawtypes")

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/Reconciler.java

@@ -19,17 +19,17 @@ public interface Reconciler<P extends HasMetadata> {
    */
   UpdateControl<P> reconcile(P resource, Context<P> context) throws Exception;
 
-
   /**
    * Prepares a map of {@link EventSource} implementations keyed by the name with which they need to
    * be registered by the SDK.
    *
    * @param context a {@link EventSourceContext} providing access to information useful to event
    *        sources
-   * @return a map of event sources to register
+   * @return a list of event sources
    */
-  default Map<String, EventSource> prepareEventSources(EventSourceContext<P> context) {
-    return Map.of();
+  // todo should be ? extends EventSource
+  default List<EventSource> prepareEventSources(EventSourceContext<P> context) {
+    return Collections.emptyList();
   }
 
 }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/DependentResource.java

@@ -49,6 +49,8 @@ default Optional<? extends ResourceEventSource<R, P>> eventSource(
     return Optional.empty();
   }
 
+
+
   /**
    * Retrieves the secondary resource (if it exists) associated with the specified primary resource
    * for this DependentResource.

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

@@ -238,19 +238,19 @@ public void initAndRegisterEventSources(EventSourceContext<P> context) {
         managedWorkflow.getDependentResourcesByNameWithoutActivationCondition();
     final var size = dependentResourcesByName.size();
     if (size > 0) {
-      dependentResourcesByName.forEach((key, dependentResource) -> {
+      dependentResourcesByName.forEach(dependentResource -> {
         if (dependentResource instanceof EventSourceProvider provider) {
           final var source = provider.initEventSource(context);
-          eventSourceManager.registerEventSource(key, source);
+          eventSourceManager.registerEventSource(source);
         } else {
           Optional<ResourceEventSource> eventSource = dependentResource.eventSource(context);
-          eventSource.ifPresent(es -> eventSourceManager.registerEventSource(key, es));
+          eventSource.ifPresent(es -> eventSourceManager.registerEventSource(es));
         }
       });
 
       // resolve event sources referenced by name for dependents that reuse an existing event source
       final Map<String, List<EventSourceReferencer>> unresolvable = new HashMap<>(size);
-      dependentResourcesByName.values().stream()
+      dependentResourcesByName.stream()
           .filter(EventSourceReferencer.class::isInstance)
           .map(EventSourceReferencer.class::cast)
           .forEach(dr -> {

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/AbstractWorkflowExecutor.java

@@ -139,8 +139,7 @@ protected <R> void registerOrDeregisterEventSourceBasedOnActivation(
         var eventSource =
             dr.eventSource(eventSourceRetriever.eventSourceContextForDynamicRegistration());
         var es = eventSource.orElseThrow();
-        eventSourceRetriever.dynamicallyRegisterEventSource(dr.name(), es);
-
+        eventSourceRetriever.dynamicallyRegisterEventSource(es);
       } else {
         eventSourceRetriever.dynamicallyDeRegisterEventSource(dr.name());
       }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/DefaultWorkflow.java

@@ -148,14 +148,8 @@ public Map<String, DependentResource> getDependentResourcesByName() {
     return resources;
   }
 
-  public Map<String, DependentResource> getDependentResourcesByNameWithoutActivationCondition() {
-    final var resources = new HashMap<String, DependentResource>(dependentResourceNodes.size());
-    dependentResourceNodes
-        .forEach((name, node) -> {
-          if (node.getActivationCondition().isEmpty()) {
-            resources.put(name, node.getDependentResource());
-          }
-        });
-    return resources;
+  public List<DependentResource> getDependentResourcesByNameWithoutActivationCondition() {
+    return dependentResourceNodes.values().stream().filter(n->n.getActivationCondition().isEmpty())
+            .map(DependentResourceNode::getDependentResource).toList();
   }
 }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/workflow/Workflow.java

@@ -1,6 +1,7 @@
 package io.javaoperatorsdk.operator.processing.dependent.workflow;
 
 import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 
@@ -44,7 +45,7 @@ default Map<String, DependentResource> getDependentResourcesByName() {
   }
 
   @SuppressWarnings("rawtypes")
-  default Map<String, DependentResource> getDependentResourcesByNameWithoutActivationCondition() {
-    return Collections.emptyMap();
+  default List<DependentResource> getDependentResourcesByNameWithoutActivationCondition() {
+    return Collections.emptyList();
   }
 }

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

@@ -144,29 +144,23 @@ private Void stopEventSource(NamedEventSource eventSource) {
     return null;
   }
 
-  public final void registerEventSource(EventSource eventSource) throws OperatorException {
-    registerEventSource(null, eventSource);
-  }
-
   @SuppressWarnings("rawtypes")
-  public final synchronized void registerEventSource(String name, EventSource eventSource)
+  public final synchronized void registerEventSource(EventSource eventSource)
       throws OperatorException {
     Objects.requireNonNull(eventSource, "EventSource must not be null");
     try {
-      if (name == null || name.isBlank()) {
-        name = EventSourceUtils.generateNameFor(eventSource);
-      }
       if (eventSource instanceof ManagedInformerEventSource managedInformerEventSource) {
+
         managedInformerEventSource.setConfigurationService(
             controller.getConfiguration().getConfigurationService());
       }
-      final var named = new NamedEventSource(eventSource, name);
+      final var named = new NamedEventSource(eventSource, eventSource.name());
       eventSources.add(named);
       named.setEventHandler(controller.getEventProcessor());
     } catch (IllegalStateException | MissingCRDException e) {
       throw e; // leave untouched
     } catch (Exception e) {
-      throw new OperatorException("Couldn't register event source: " + name + " for "
+      throw new OperatorException("Couldn't register event source: " + eventSource.name() + " for "
           + controller.getConfiguration().getName() + " controller", e);
     }
   }
@@ -232,14 +226,13 @@ public <R> List<ResourceEventSource<R, P>> getResourceEventSourcesFor(Class<R> d
   }
 
   @Override
-  public EventSource dynamicallyRegisterEventSource(String name,
-      EventSource eventSource) {
+  public EventSource dynamicallyRegisterEventSource(EventSource eventSource) {
     synchronized (this) {
-      var actual = eventSources.existing(name, eventSource);
+      var actual = eventSources.existing(eventSource.name(), eventSource);
       if (actual != null) {
         eventSource = actual.eventSource();
       } else {
-        registerEventSource(name, eventSource);
+        registerEventSource(eventSource);
       }
     }
     // The start itself is blocking thus blocking only the threads which are attempt to start the

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

@@ -39,11 +39,10 @@ default <R> ResourceEventSource<R, P> getResourceEventSourceFor(Class<R> depende
    * name will ever be registered.
    * </p>
    *
-   * @param name of the event source
    * @param eventSource to register
    * @return the actual event source registered. Might not be the same as the parameter.
    */
-  EventSource dynamicallyRegisterEventSource(String name, EventSource eventSource);
+  EventSource dynamicallyRegisterEventSource(EventSource eventSource);
 
 
   /**