refactor: simplify handling of reused event sources

DependentResources that provide an EventSource should implement
EventSourceAware. If they want to reuse an already created event source
instead of providing their own, they need to select the appropriate
event source identified by the name specified by the useEventSourceNamed
method.

Author: metacosm

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/AnnotationControllerConfiguration.java

@@ -248,7 +248,8 @@ public List<DependentResourceSpec> getDependentResources() {
             instantiateIfNotDefault(dependent.readyPostcondition(), Condition.class, context),
             instantiateIfNotDefault(dependent.reconcilePrecondition(), Condition.class, context),
             instantiateIfNotDefault(dependent.deletePostcondition(), Condition.class, context),
-            dependent.provideEventSource());
+            Constants.NO_VALUE_SET.equals(dependent.eventSource()) ? null
+                : dependent.eventSource());
         specsMap.put(name, spec);
       }
 
@@ -314,14 +315,12 @@ private Object createKubernetesResourceConfig(Class<? extends DependentResource>
       resourceDiscriminator =
           instantiateIfNotDefault(kubeDependent.resourceDiscriminator(),
               ResourceDiscriminator.class, context);
-      eventSourceNameToUse = Constants.NO_VALUE_SET.equals(kubeDependent.eventSourceToUse()) ? null
-          : kubeDependent.eventSourceToUse();
     }
 
     config =
         new KubernetesDependentResourceConfig(namespaces, labelSelector, configuredNS,
             resourceDiscriminator, onAddFilter,
-            onUpdateFilter, onDeleteFilter, genericFilter, eventSourceNameToUse);
+            onUpdateFilter, onDeleteFilter, genericFilter);
 
     return config;
   }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ControllerConfigurationOverrider.java

@@ -174,7 +174,7 @@ private void replaceConfig(String name, Object newConfig, DependentResourceSpec<
     namedDependentResourceSpecs.put(name,
         new DependentResourceSpec<>(current.getDependentResourceClass(), newConfig, name,
             current.getDependsOn(), current.getReadyCondition(), current.getReconcileCondition(),
-            current.getDeletePostCondition(), current.provideEventSource()));
+            current.getDeletePostCondition(), current.getEventSourceName().orElse(null)));
   }
 
   @SuppressWarnings("unchecked")
@@ -220,7 +220,8 @@ public ControllerConfiguration<R> build() {
       KubernetesDependentResourceConfig c) {
     return new DependentResourceSpec(spec.getDependentResourceClass(),
         c.setNamespaces(namespaces), name, spec.getDependsOn(), spec.getReadyCondition(),
-        spec.getReconcileCondition(), spec.getDeletePostCondition(), spec.provideEventSource());
+        spec.getReconcileCondition(), spec.getDeletePostCondition(),
+        (String) spec.getEventSourceName().orElse(null));
   }
 
   public static <R extends HasMetadata> ControllerConfigurationOverrider<R> override(

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/dependent/DependentResourceSpec.java

@@ -23,20 +23,20 @@ public class DependentResourceSpec<T extends DependentResource<?, ?>, C> {
 
   private final Condition<?, ?> deletePostCondition;
 
-  private final boolean provideEventSource;
+  private final String eventSourceName;
 
   public DependentResourceSpec(Class<T> dependentResourceClass, C dependentResourceConfig,
       String name, Set<String> dependsOn, Condition<?, ?> readyCondition,
       Condition<?, ?> reconcileCondition, Condition<?, ?> deletePostCondition,
-      boolean provideEventSource) {
+      String eventSourceName) {
     this.dependentResourceClass = dependentResourceClass;
     this.dependentResourceConfig = dependentResourceConfig;
     this.name = name;
     this.dependsOn = dependsOn;
     this.readyCondition = readyCondition;
     this.reconcileCondition = reconcileCondition;
     this.deletePostCondition = deletePostCondition;
-    this.provideEventSource = provideEventSource;
+    this.eventSourceName = eventSourceName;
   }
 
   public Class<T> getDependentResourceClass() {
@@ -94,7 +94,7 @@ public Condition getDeletePostCondition() {
     return deletePostCondition;
   }
 
-  public boolean provideEventSource() {
-    return provideEventSource;
+  public Optional<String> getEventSourceName() {
+    return Optional.ofNullable(eventSourceName);
   }
 }

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

@@ -3,6 +3,8 @@
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.client.KubernetesClient;
 import io.javaoperatorsdk.operator.api.config.ControllerConfiguration;
+import io.javaoperatorsdk.operator.processing.event.EventSourceManager;
+import io.javaoperatorsdk.operator.processing.event.EventSourceRetriever;
 import io.javaoperatorsdk.operator.processing.event.source.EventSource;
 import io.javaoperatorsdk.operator.processing.event.source.IndexerResourceCache;
 
@@ -13,14 +15,14 @@
  */
 public class EventSourceContext<P extends HasMetadata> {
 
-  private final IndexerResourceCache<P> primaryCache;
+  private final EventSourceManager<P> eventSourceManager;
   private final ControllerConfiguration<P> controllerConfiguration;
   private final KubernetesClient client;
 
-  public EventSourceContext(IndexerResourceCache<P> primaryCache,
+  public EventSourceContext(EventSourceManager<P> eventSourceManager,
       ControllerConfiguration<P> controllerConfiguration,
       KubernetesClient client) {
-    this.primaryCache = primaryCache;
+    this.eventSourceManager = eventSourceManager;
     this.controllerConfiguration = controllerConfiguration;
     this.client = client;
   }
@@ -31,7 +33,7 @@ public EventSourceContext(IndexerResourceCache<P> primaryCache,
    * @return the primary resource cache
    */
   public IndexerResourceCache<P> getPrimaryCache() {
-    return primaryCache;
+    return eventSourceManager.getControllerResourceEventSource();
   }
 
   /**
@@ -54,4 +56,8 @@ public ControllerConfiguration<P> getControllerConfiguration() {
   public KubernetesClient getClient() {
     return client;
   }
+
+  public EventSourceRetriever<P> getEventSourceRetriever() {
+    return eventSourceManager;
+  }
 }

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

@@ -1,14 +1,14 @@
 package io.javaoperatorsdk.operator.api.reconciler;
 
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
-import java.util.Optional;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.EventSourceAware;
 import io.javaoperatorsdk.operator.processing.event.source.EventSource;
-import io.javaoperatorsdk.operator.processing.event.source.ResourceEventSource;
 
 /**
  * An interface that a {@link Reconciler} can implement to have the SDK register the provided
@@ -46,13 +46,12 @@ static Map<String, EventSource> nameEventSources(EventSource... eventSources) {
   @SuppressWarnings("unchecked,rawtypes")
   static <K extends HasMetadata> Map<String, EventSource> nameEventSourcesFromDependentResource(
       EventSourceContext<K> context, DependentResource... dependentResources) {
-
     if (dependentResources != null) {
       Map<String, EventSource> eventSourceMap = new HashMap<>(dependentResources.length);
-      for (DependentResource dependentResource : dependentResources) {
-        Optional<ResourceEventSource> es = dependentResource.eventSource(context);
-        es.ifPresent(e -> eventSourceMap.put(generateNameFor(e), e));
-      }
+      Arrays.stream(dependentResources)
+          .filter(EventSourceAware.class::isInstance)
+          .forEach(esa -> ((EventSourceAware<?, K>) esa).eventSource(context)
+              .ifPresent(es -> eventSourceMap.put(generateNameFor(es), es)));
       return eventSourceMap;
     } else {
       return Collections.emptyMap();

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

@@ -65,5 +65,5 @@
    *
    * @return if the event source (if any) provided by the dependent resource should be used or not.
    */
-  boolean provideEventSource() default true;
+  String eventSource() default NO_VALUE_SET;
 }

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

@@ -4,8 +4,6 @@
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
-import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
-import io.javaoperatorsdk.operator.processing.event.source.ResourceEventSource;
 
 /**
  * An interface to implement and provide dependent resource support.
@@ -31,31 +29,6 @@ public interface DependentResource<R, P extends HasMetadata> {
    */
   Class<R> resourceType();
 
-  /**
-   * Dependent resources are designed to by default provide event sources. There are cases where it
-   * might not:
-   * <ul>
-   * <li>If an event source is shared between multiple dependent resources. In this case only one or
-   * none of the dependent resources sharing the event source should provide one.</li>
-   * <li>Some special implementation of an event source. That just execute some action might not
-   * provide one.</li>
-   * </ul>
-   *
-   * @param eventSourceContext context of event source initialization
-   * @return an optional event source
-   */
-  default Optional<ResourceEventSource<R, P>> eventSource(
-      EventSourceContext<P> eventSourceContext) {
-    return Optional.empty();
-  }
-
-  /**
-   * Calling this method, instructs the implementation to not provide an event source, even if it
-   * normally does.
-   */
-  void doNotProvideEventSource();
-
-
   default Optional<R> getSecondaryResource(P primary, Context<P> context) {
     return Optional.empty();
   }

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

@@ -1,10 +1,36 @@
 package io.javaoperatorsdk.operator.api.reconciler.dependent;
 
+import java.util.Optional;
+
 import io.fabric8.kubernetes.api.model.HasMetadata;
-import io.javaoperatorsdk.operator.processing.event.EventSourceRetriever;
+import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
+import io.javaoperatorsdk.operator.processing.event.source.ResourceEventSource;
+
+public interface EventSourceAware<R, P extends HasMetadata> {
+
+  /**
+   * Dependent resources are designed to by default provide event sources. There are cases where it
+   * might not:
+   * <ul>
+   * <li>If an event source is shared between multiple dependent resources. In this case only one or
+   * none of the dependent resources sharing the event source should provide one.</li>
+   * <li>Some special implementation of an event source. That just execute some action might not
+   * provide one.</li>
+   * </ul>
+   *
+   * @param context context of event source initialization
+   * @return an optional event source
+   */
+  default Optional<ResourceEventSource<R, P>> eventSource(EventSourceContext<P> context) {
+    return getUsedEventSourceName().map(
+        name -> context.getEventSourceRetriever().getResourceEventSourceFor(resourceType(), name));
+  }
 
-public interface EventSourceAware<P extends HasMetadata> {
+  Class<R> resourceType();
 
-  void selectEventSources(EventSourceRetriever<P> eventSourceRetriever);
+  void useEventSourceNamed(String eventSourceName);
 
+  default Optional<String> getUsedEventSourceName() {
+    return Optional.empty();
+  }
 }

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

@@ -40,7 +40,6 @@
 import io.javaoperatorsdk.operator.processing.event.EventProcessor;
 import io.javaoperatorsdk.operator.processing.event.EventSourceManager;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
-import io.javaoperatorsdk.operator.processing.event.source.ResourceEventSource;
 
 import static io.javaoperatorsdk.operator.api.reconciler.Constants.WATCH_CURRENT_NAMESPACE;
 
@@ -214,25 +213,21 @@ public void initAndRegisterEventSources(EventSourceContext<P> context) {
       final var ownSources = provider.prepareEventSources(context);
       ownSources.forEach(eventSourceManager::registerEventSource);
     }
-    managedWorkflow
-        .getDependentResourcesByName().entrySet().stream()
-        .forEach(drEntry -> {
-          if (drEntry.getValue() instanceof EventSourceProvider) {
-            final var provider = (EventSourceProvider) drEntry.getValue();
-            final var source = provider.initEventSource(context);
-            eventSourceManager.registerEventSource(drEntry.getKey(), source);
-          } else {
-            Optional<ResourceEventSource> eventSource =
-                drEntry.getValue().eventSource(context);
-            eventSource.ifPresent(es -> {
-              eventSourceManager.registerEventSource(drEntry.getKey(), es);
-            });
-          }
-        });
-    managedWorkflow.getDependentResourcesByName().entrySet().stream().map(Map.Entry::getValue)
-        .filter(EventSourceAware.class::isInstance)
-        .forEach(dr -> ((EventSourceAware) dr)
-            .selectEventSources(eventSourceManager));
+    managedWorkflow.getDependentResourcesByName().entrySet().stream().filter(entry -> {
+      final var value = entry.getValue();
+      return value instanceof EventSourceProvider || value instanceof EventSourceAware;
+    }).forEach(entry -> {
+      final var value = entry.getValue();
+      final var key = entry.getKey();
+      if (value instanceof EventSourceProvider) {
+        final var provider = (EventSourceProvider) value;
+        final var source = provider.initEventSource(context);
+        eventSourceManager.registerEventSource(key, source);
+      } else {
+        ((EventSourceAware<?, P>) value).eventSource(context)
+            .ifPresent(es -> eventSourceManager.registerEventSource(key, es));
+      }
+    });
   }
 
   @Override
@@ -299,8 +294,8 @@ public synchronized void start(boolean startEventProcessor) throws OperatorExcep
     try {
       // check that the custom resource is known by the cluster if configured that way
       validateCRDWithLocalModelIfRequired(resClass, controllerName, crdName, specVersion);
-      final var context = new EventSourceContext<>(
-          eventSourceManager.getControllerResourceEventSource(), configuration, kubernetesClient);
+      final var context =
+          new EventSourceContext<>(eventSourceManager, configuration, kubernetesClient);
 
       initAndRegisterEventSources(context);
       eventSourceManager.start();

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

@@ -9,13 +9,11 @@
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.reconciler.Context;
-import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
 import io.javaoperatorsdk.operator.api.reconciler.Ignore;
 import io.javaoperatorsdk.operator.api.reconciler.ResourceDiscriminator;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource;
 import io.javaoperatorsdk.operator.api.reconciler.dependent.ReconcileResult;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
-import io.javaoperatorsdk.operator.processing.event.source.ResourceEventSource;
 
 @Ignore
 public abstract class AbstractDependentResource<R, P extends HasMetadata>
@@ -29,7 +27,6 @@ public abstract class AbstractDependentResource<R, P extends HasMetadata>
   protected Creator<R, P> creator;
   protected Updater<R, P> updater;
   protected BulkDependentResource<R, P> bulkDependentResource;
-  private boolean returnEventSource = true;
 
   protected List<ResourceDiscriminator<R, P>> resourceDiscriminator = new ArrayList<>(1);
 
@@ -41,23 +38,6 @@ public AbstractDependentResource() {
     bulkDependentResource = bulk ? (BulkDependentResource<R, P>) this : null;
   }
 
-  @Override
-  public void doNotProvideEventSource() {
-    this.returnEventSource = false;
-  }
-
-  @Override
-  public Optional<ResourceEventSource<R, P>> eventSource(EventSourceContext<P> eventSourceContext) {
-    if (!returnEventSource) {
-      return Optional.empty();
-    } else {
-      return Optional.of(provideEventSource(eventSourceContext));
-    }
-  }
-
-  protected abstract ResourceEventSource<R, P> provideEventSource(
-      EventSourceContext<P> eventSourceContext);
-
   @Override
   public ReconcileResult<R> reconcile(P primary, Context<P> context) {
     if (bulk) {
@@ -238,8 +218,4 @@ public ResourceDiscriminator<R, P> getResourceDiscriminator() {
   protected int lastKnownBulkSize() {
     return resourceDiscriminator.size();
   }
-
-  protected boolean getReturnEventSource() {
-    return returnEventSource;
-  }
 }