improve: PerResourcePollingEventSource init improvements

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

@@ -1,5 +1,7 @@
 package io.javaoperatorsdk.operator.processing.dependent.external;
 
+import java.time.Duration;
+
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
 import io.javaoperatorsdk.operator.api.reconciler.Ignore;
@@ -23,8 +25,8 @@ public PerResourcePollingDependentResource(Class<R> resourceType, long pollingPe
   @Override
   protected ExternalResourceCachingEventSource<R, P> createEventSource(
       EventSourceContext<P> context) {
-    return new PerResourcePollingEventSource<>(this, context.getPrimaryCache(),
-        getPollingPeriod(), resourceType(), this);
+    return new PerResourcePollingEventSource<>(this, context,
+        Duration.ofMillis(getPollingPeriod()), resourceType(), this);
   }
 
 }

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

@@ -1,15 +1,24 @@
 package io.javaoperatorsdk.operator.processing.event.source.polling;
 
 import java.time.Duration;
-import java.util.*;
-import java.util.concurrent.*;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Predicate;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.javaoperatorsdk.operator.OperatorException;
+import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
 import io.javaoperatorsdk.operator.processing.event.source.Cache;
 import io.javaoperatorsdk.operator.processing.event.source.CacheKeyMapper;
@@ -43,19 +52,82 @@ public class PerResourcePollingEventSource<R, P extends HasMetadata>
   private final long period;
   private final Set<ResourceID> fetchedForPrimaries = ConcurrentHashMap.newKeySet();
 
+  public PerResourcePollingEventSource(ResourceFetcher<R, P> resourceFetcher,
+      EventSourceContext<P> context, Duration defaultPollingPeriod,
+      Class<R> resourceClass) {
+    this(resourceFetcher, context.getPrimaryCache(), defaultPollingPeriod.toMillis(),
+        null, resourceClass,
+        CacheKeyMapper.singleResourceCacheKeyMapper());
+  }
 
+  /**
+   * @deprecated use the variant which uses {@link EventSourceContext} instead of {@link Cache} and
+   *             {@link Duration} for period parameter as it provides a more intuitive API.
+   *
+   * @param resourceFetcher fetches resource related to a primary resource
+   * @param resourceCache cache of the primary resource
+   * @param period default polling period
+   * @param resourceClass class of the target resource
+   */
+  @Deprecated(forRemoval = true)
   public PerResourcePollingEventSource(ResourceFetcher<R, P> resourceFetcher,
       Cache<P> resourceCache, long period, Class<R> resourceClass) {
     this(resourceFetcher, resourceCache, period, null, resourceClass,
         CacheKeyMapper.singleResourceCacheKeyMapper());
   }
 
+  public PerResourcePollingEventSource(ResourceFetcher<R, P> resourceFetcher,
+      EventSourceContext<P> context,
+      Duration defaultPollingPeriod,
+      Class<R> resourceClass,
+      CacheKeyMapper<R> cacheKeyMapper) {
+    this(resourceFetcher, context.getPrimaryCache(), defaultPollingPeriod.toMillis(),
+        null, resourceClass, cacheKeyMapper);
+  }
+
+  /**
+   * @deprecated use the variant which uses {@link EventSourceContext} instead of {@link Cache} and
+   *             {@link Duration} for period parameter as it provides a more intuitive API.
+   *
+   * @param resourceFetcher fetches resource related to a primary resource
+   * @param resourceCache cache of the primary resource
+   * @param period default polling period
+   * @param resourceClass class of the target resource
+   * @param cacheKeyMapper use to distinguish resource in case more resources are handled for a
+   *        single primary resource
+   */
+  @Deprecated(forRemoval = true)
   public PerResourcePollingEventSource(ResourceFetcher<R, P> resourceFetcher,
       Cache<P> resourceCache, long period, Class<R> resourceClass,
       CacheKeyMapper<R> cacheKeyMapper) {
     this(resourceFetcher, resourceCache, period, null, resourceClass, cacheKeyMapper);
   }
 
+  public PerResourcePollingEventSource(ResourceFetcher<R, P> resourceFetcher,
+      EventSourceContext<P> context,
+      Duration defaultPollingPeriod,
+      Predicate<P> registerPredicate,
+      Class<R> resourceClass,
+      CacheKeyMapper<R> cacheKeyMapper) {
+    this(resourceFetcher, context.getPrimaryCache(), defaultPollingPeriod.toMillis(),
+        registerPredicate, resourceClass, cacheKeyMapper,
+        new ScheduledThreadPoolExecutor(DEFAULT_EXECUTOR_THREAD_NUMBER));
+  }
+
+  /**
+   * @deprecated use the variant which uses {@link EventSourceContext} instead of {@link Cache} and
+   *             {@link Duration} for period parameter as it provides a more intuitive API.
+   *
+   * @param resourceFetcher fetches resource related to a primary resource
+   * @param resourceCache cache of the primary resource
+   * @param period default polling period
+   * @param resourceClass class of the target resource
+   * @param cacheKeyMapper use to distinguish resource in case more resources are handled for a
+   *        single primary resource
+   * @param registerPredicate used to determine if the related resource for a custom resource should
+   *        be polled or not.
+   */
+  @Deprecated(forRemoval = true)
   public PerResourcePollingEventSource(ResourceFetcher<R, P> resourceFetcher,
       Cache<P> resourceCache, long period,
       Predicate<P> registerPredicate, Class<R> resourceClass,
@@ -64,7 +136,35 @@ public PerResourcePollingEventSource(ResourceFetcher<R, P> resourceFetcher,
         new ScheduledThreadPoolExecutor(DEFAULT_EXECUTOR_THREAD_NUMBER));
   }
 
-  public PerResourcePollingEventSource(ResourceFetcher<R, P> resourceFetcher,
+
+  public PerResourcePollingEventSource(
+      ResourceFetcher<R, P> resourceFetcher,
+      EventSourceContext<P> context, Duration defaultPollingPeriod,
+      Predicate<P> registerPredicate, Class<R> resourceClass,
+      CacheKeyMapper<R> cacheKeyMapper, ScheduledExecutorService executorService) {
+    this(resourceFetcher, context.getPrimaryCache(), defaultPollingPeriod.toMillis(),
+        registerPredicate,
+        resourceClass, cacheKeyMapper, executorService);
+  }
+
+  /**
+   * @deprecated use the variant which uses {@link EventSourceContext} instead of {@link Cache} and
+   *             {@link Duration} for period parameter as it provides a more intuitive API.
+   *
+   * @param resourceFetcher fetches resource related to a primary resource
+   * @param resourceCache cache of the primary resource
+   * @param period default polling period
+   * @param resourceClass class of the target resource
+   * @param cacheKeyMapper use to distinguish resource in case more resources are handled for a
+   *        single primary resource
+   * @param registerPredicate used to determine if the related resource for a custom resource should
+   *        be polled or not.
+   * @param executorService custom executor service
+   */
+
+  @Deprecated(forRemoval = true)
+  public PerResourcePollingEventSource(
+      ResourceFetcher<R, P> resourceFetcher,
       Cache<P> resourceCache, long period,
       Predicate<P> registerPredicate, Class<R> resourceClass,
       CacheKeyMapper<R> cacheKeyMapper, ScheduledExecutorService executorService) {

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/polling/PerResourcePollingEventSourceTest.java

@@ -9,11 +9,9 @@
 import org.junit.jupiter.api.Test;
 
 import io.javaoperatorsdk.operator.TestUtils;
+import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
 import io.javaoperatorsdk.operator.processing.event.EventHandler;
-import io.javaoperatorsdk.operator.processing.event.source.AbstractEventSourceTestBase;
-import io.javaoperatorsdk.operator.processing.event.source.Cache;
-import io.javaoperatorsdk.operator.processing.event.source.CacheKeyMapper;
-import io.javaoperatorsdk.operator.processing.event.source.SampleExternalResource;
+import io.javaoperatorsdk.operator.processing.event.source.*;
 import io.javaoperatorsdk.operator.sample.simple.TestCustomResource;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -35,16 +33,19 @@ class PerResourcePollingEventSourceTest extends
   private final PerResourcePollingEventSource.ResourceFetcher<SampleExternalResource, TestCustomResource> supplier =
       mock(PerResourcePollingEventSource.ResourceFetcher.class);
   @SuppressWarnings("unchecked")
-  private final Cache<TestCustomResource> resourceCache = mock(Cache.class);
+  private final IndexerResourceCache<TestCustomResource> resourceCache =
+      mock(IndexerResourceCache.class);
   private final TestCustomResource testCustomResource = TestUtils.testCustomResource();
+  private final EventSourceContext<TestCustomResource> context = mock(EventSourceContext.class);
 
   @BeforeEach
   public void setup() {
     when(resourceCache.get(any())).thenReturn(Optional.of(testCustomResource));
     when(supplier.fetchResources(any()))
         .thenReturn(Set.of(SampleExternalResource.testResource1()));
+    when(context.getPrimaryCache()).thenReturn(resourceCache);
 
-    setUpSource(new PerResourcePollingEventSource<>(supplier, resourceCache, PERIOD,
+    setUpSource(new PerResourcePollingEventSource<>(supplier, context, Duration.ofMillis(PERIOD),
         SampleExternalResource.class, r -> r.getName() + "#" + r.getValue()));
   }
 
@@ -61,7 +62,7 @@ void pollsTheResourceAfterAwareOfIt() {
 
   @Test
   void registeringTaskOnAPredicate() {
-    setUpSource(new PerResourcePollingEventSource<>(supplier, resourceCache, PERIOD,
+    setUpSource(new PerResourcePollingEventSource<>(supplier, context, Duration.ofMillis(PERIOD),
         testCustomResource -> testCustomResource.getMetadata().getGeneration() > 1,
         SampleExternalResource.class, CacheKeyMapper.singleResourceCacheKeyMapper()));
     source.onResourceCreated(testCustomResource);

operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/externalstate/ExternalStateReconciler.java

@@ -1,5 +1,6 @@
 package io.javaoperatorsdk.operator.sample.externalstate;
 
+import java.time.Duration;
 import java.util.Collections;
 import java.util.Map;
 import java.util.Set;
@@ -122,7 +123,7 @@ public Map<String, EventSource> prepareEventSources(
       var id = configMap.getData().get(ID_KEY);
       var externalResource = externalService.read(id);
       return externalResource.map(Set::of).orElseGet(Collections::emptySet);
-    }, context.getPrimaryCache(), 300L, ExternalResource.class);
+    }, context, Duration.ofMillis(300L), ExternalResource.class);
 
     return EventSourceInitializer.nameEventSources(configMapEventSource,
         externalResourceEventSource);

operator-framework/src/test/java/io/javaoperatorsdk/operator/sample/perresourceeventsource/PerResourcePollingEventSourceTestReconciler.java

@@ -1,5 +1,6 @@
 package io.javaoperatorsdk.operator.sample.perresourceeventsource;
 
+import java.time.Duration;
 import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
@@ -41,7 +42,7 @@ public Map<String, EventSource> prepareEventSources(
           numberOfFetchExecutions.compute(resource.getMetadata().getName(), (s, v) -> v + 1);
           return Set.of(UUID.randomUUID().toString());
         },
-            context.getPrimaryCache(), POLL_PERIOD, String.class);
+            context, Duration.ofMillis(POLL_PERIOD), String.class);
     return EventSourceInitializer.nameEventSources(eventSource);
   }