fix: make dependent sources proper ResourceEventSources

Author: metacosm

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

@@ -24,9 +24,9 @@
 import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
 import io.javaoperatorsdk.operator.processing.event.EventSourceManager;
 import io.javaoperatorsdk.operator.processing.event.source.AbstractResourceEventSource;
+import io.javaoperatorsdk.operator.processing.event.source.DependentResourceEventSource;
 import io.javaoperatorsdk.operator.processing.event.source.EventSourceRegistry;
 import io.javaoperatorsdk.operator.processing.event.source.EventSourceWrapper;
-import io.javaoperatorsdk.operator.processing.event.source.InformerEventSourceEventSourceWrapper;
 import io.javaoperatorsdk.operator.processing.event.source.ResourceEventFilter;
 
 public class Controller<R extends HasMetadata> implements Reconciler<R>,
@@ -143,7 +143,7 @@ public UpdateControl<R> execute() {
   public void prepareEventSources(EventSourceRegistry<R> eventSourceRegistry, Cloner cloner) {
     configuration.getDependentResources().forEach(dependent -> {
       final var dependentConfiguration = dependent.getConfiguration();
-      final var source = new AbstractResourceEventSource<>(dependentConfiguration,
+      new AbstractResourceEventSource<>(dependentConfiguration,
           kubernetesClient.resources(dependentConfiguration.getResourceClass()), cloner,
           eventSourceRegistry) {
         @Override
@@ -154,14 +154,12 @@ protected ResourceEventFilter initFilter(ResourceConfiguration configuration) {
         @Override
         protected EventSourceWrapper wrapEventSource(
             FilterWatchListDeletable filteredBySelectorClient, Cloner cloner) {
-          return new InformerEventSourceEventSourceWrapper(filteredBySelectorClient, cloner,
-              dependentConfiguration.getPrimaryResourcesRetriever(),
-              dependentConfiguration.getAssociatedResourceRetriever(),
-              dependentConfiguration.skipUpdateIfUnchanged(),
-              eventSourceRegistry);
+          final var source = new DependentResourceEventSource(filteredBySelectorClient, cloner,
+              dependentConfiguration);
+          eventSourceRegistry.registerEventSource(source);
+          return source;
         }
       };
-      eventSourceRegistry.registerEventSource(source);
     });
 
     if (reconciler instanceof EventSourceInitializer) {

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

@@ -0,0 +1,54 @@
+package io.javaoperatorsdk.operator.processing.event.source;
+
+import java.util.Optional;
+import java.util.function.Predicate;
+import java.util.stream.Stream;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.fabric8.kubernetes.api.model.KubernetesResourceList;
+import io.fabric8.kubernetes.client.dsl.FilterWatchListDeletable;
+import io.javaoperatorsdk.operator.api.config.Cloner;
+import io.javaoperatorsdk.operator.api.config.ResourceConfiguration;
+import io.javaoperatorsdk.operator.api.config.dependent.DependentResourceConfiguration;
+import io.javaoperatorsdk.operator.processing.event.ResourceID;
+
+public class DependentResourceEventSource<T extends HasMetadata, P extends HasMetadata>
+    extends InformerEventSource<T, P>
+    implements EventSourceWrapper<T>, ResourceEventSource<T, P> {
+  private final DependentResourceConfiguration<T, P> configuration;
+
+  public DependentResourceEventSource(
+      FilterWatchListDeletable<T, KubernetesResourceList<T>> client,
+      Cloner cloner, DependentResourceConfiguration<T, P> dependentConfiguration) {
+    super(client.runnableInformer(0),
+        dependentConfiguration.getPrimaryResourcesRetriever(),
+        dependentConfiguration.getAssociatedResourceRetriever(),
+        dependentConfiguration.skipUpdateIfUnchanged(), cloner);
+    this.configuration = dependentConfiguration;
+  }
+
+  @Override
+  public Optional<T> get(ResourceID resourceID) {
+    return getCache().get(resourceID);
+  }
+
+  @Override
+  public Stream<T> list(Predicate<T> predicate) {
+    return getCache().list(predicate);
+  }
+
+  @Override
+  public Stream<T> list(String namespace, Predicate<T> predicate) {
+    return getCache().list(namespace, predicate);
+  }
+
+  @Override
+  public ResourceCache<T> getResourceCache() {
+    return getCache();
+  }
+
+  @Override
+  public ResourceConfiguration<T, ? extends ResourceConfiguration> getConfiguration() {
+    return configuration;
+  }
+}