feat: merge EventSource and EventSourceMetadata

Signed-off-by: Chris Laprun <claprun@redhat.com>

Author: metacosm

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

@@ -7,6 +7,7 @@
 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;
@@ -81,6 +82,7 @@ public synchronized void start() {
         getThreadNamer("start"));
   }
 
+  @SuppressWarnings("rawtypes")
   private static Function<EventSource, String> getThreadNamer(String stage) {
     return es -> es.priority() + " " + stage + " -> " + es.name();
   }
@@ -106,7 +108,7 @@ private void logEventSourceEvent(EventSource eventSource, String event) {
     }
   }
 
-  private Void startEventSource(EventSource eventSource) {
+  private <R> Void startEventSource(EventSource<R, P> eventSource) {
     try {
       logEventSourceEvent(eventSource, "Starting");
       eventSource.start();
@@ -119,7 +121,7 @@ private Void startEventSource(EventSource eventSource) {
     return null;
   }
 
-  private Void stopEventSource(EventSource eventSource) {
+  private <R> Void stopEventSource(EventSource<R, P> eventSource) {
     try {
       logEventSourceEvent(eventSource, "Stopping");
       eventSource.stop();
@@ -131,7 +133,7 @@ private Void stopEventSource(EventSource eventSource) {
   }
 
   @SuppressWarnings("rawtypes")
-  public final synchronized void registerEventSource(EventSource eventSource)
+  public final synchronized <R> void registerEventSource(EventSource<R, P> eventSource)
       throws OperatorException {
     Objects.requireNonNull(eventSource, "EventSource must not be null");
     try {
@@ -184,16 +186,22 @@ public void changeNamespaces(Set<String> namespaces) {
         getEventSourceThreadNamer("changeNamespace"));
   }
 
-  public Set<EventSource> getRegisteredEventSources() {
+  public Set<EventSource<?, P>> getRegisteredEventSources() {
     return eventSources.flatMappedSources()
-
         .collect(Collectors.toCollection(LinkedHashSet::new));
   }
 
+  @SuppressWarnings("rawtypes")
   public List<EventSource> allEventSources() {
     return eventSources.allEventSources().toList();
   }
 
+
+  @SuppressWarnings("unused")
+  public Stream<? extends EventSource<?, P>> getEventSourcesStream() {
+    return eventSources.flatMappedSources();
+  }
+
   public ControllerEventSource<P> getControllerEventSource() {
     return eventSources.controllerEventSource();
   }
@@ -203,7 +211,7 @@ public <R> List<EventSource<R, P>> getEventSourcesFor(Class<R> dependentType) {
   }
 
   @Override
-  public EventSource dynamicallyRegisterEventSource(EventSource eventSource) {
+  public <R> EventSource<R, P> dynamicallyRegisterEventSource(EventSource<R, P> eventSource) {
     synchronized (this) {
       var actual = eventSources.existingEventSourceOfSameNameAndType(eventSource);
       if (actual != null) {
@@ -219,8 +227,10 @@ public EventSource dynamicallyRegisterEventSource(EventSource eventSource) {
   }
 
   @Override
-  public synchronized Optional<EventSource> dynamicallyDeRegisterEventSource(String name) {
-    EventSource es = eventSources.remove(name);
+  public synchronized <R> Optional<EventSource<R, P>> dynamicallyDeRegisterEventSource(
+      String name) {
+    @SuppressWarnings("unchecked")
+    EventSource<R, P> es = eventSources.remove(name);
     if (es != null) {
       es.stop();
     }

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

@@ -15,56 +15,60 @@
 import io.javaoperatorsdk.operator.processing.event.source.controller.ControllerEventSource;
 import io.javaoperatorsdk.operator.processing.event.source.timer.TimerEventSource;
 
-class EventSources<R extends HasMetadata> {
+class EventSources<P extends HasMetadata> {
 
-  private final ConcurrentNavigableMap<String, Map<String, EventSource>> sources =
+  private final ConcurrentNavigableMap<String, Map<String, EventSource<?, P>>> sources =
       new ConcurrentSkipListMap<>();
-  private final TimerEventSource<R> retryAndRescheduleTimerEventSource =
+  private final TimerEventSource<P> retryAndRescheduleTimerEventSource =
       new TimerEventSource<>("RetryAndRescheduleTimerEventSource");
-  private ControllerEventSource<R> controllerEventSource;
+  private ControllerEventSource<P> controllerEventSource;
 
 
-  void createControllerEventSource(Controller<R> controller) {
+  void createControllerEventSource(Controller<P> controller) {
     controllerEventSource = new ControllerEventSource<>(controller);
   }
 
-  public ControllerEventSource<R> controllerEventSource() {
+  public ControllerEventSource<P> controllerEventSource() {
     return controllerEventSource;
   }
 
-  TimerEventSource<R> retryEventSource() {
+  TimerEventSource<P> retryEventSource() {
     return retryAndRescheduleTimerEventSource;
   }
 
+  @SuppressWarnings("rawtypes")
   public Stream<EventSource> allEventSources() {
     return Stream.concat(
         Stream.of(controllerEventSource(), retryAndRescheduleTimerEventSource),
         flatMappedSources());
   }
 
+  @SuppressWarnings("rawtypes")
   Stream<EventSource> additionalEventSources() {
     return Stream.concat(
         Stream.of(retryEventSource()).filter(Objects::nonNull),
         flatMappedSources());
   }
 
-  Stream<EventSource> flatMappedSources() {
+  Stream<EventSource<?, P>> flatMappedSources() {
     return sources.values().stream().flatMap(c -> c.values().stream());
   }
 
   public void clear() {
     sources.clear();
   }
 
-  public EventSource existingEventSourceOfSameNameAndType(EventSource source) {
-    return existingEventSourceOfSameType(source).get(source.name());
+  @SuppressWarnings("unchecked")
+  public <R> EventSource<R, P> existingEventSourceOfSameNameAndType(EventSource<R, P> source) {
+    return (EventSource<R, P>) existingEventSourceOfSameType(source).get(source.name());
   }
 
-  public Map<String, EventSource> existingEventSourceOfSameType(EventSource source) {
+  private <R> Map<String, EventSource<?, P>> existingEventSourceOfSameType(
+      EventSource<R, P> source) {
     return sources.getOrDefault(keyFor(source), Collections.emptyMap());
   }
 
-  public void add(EventSource eventSource) {
+  public <R> void add(EventSource<R, P> eventSource) {
     final var name = eventSource.name();
     final var existing = existingEventSourceOfSameType(eventSource);
     if (existing.get(name) != null) {
@@ -75,7 +79,7 @@ public void add(EventSource eventSource) {
     sources.computeIfAbsent(keyFor(eventSource), k -> new HashMap<>()).put(name, eventSource);
   }
 
-  private String keyFor(EventSource source) {
+  private <R> String keyFor(EventSource<R, P> source) {
     return keyFor(source.resourceType());
   }
 
@@ -84,7 +88,7 @@ private String keyFor(Class<?> dependentType) {
   }
 
   @SuppressWarnings("unchecked")
-  public <S> EventSource<S, R> get(Class<S> dependentType, String name) {
+  public <S> EventSource<S, P> get(Class<S> dependentType, String name) {
     if (dependentType == null) {
       throw new IllegalArgumentException("Must pass a dependent type to retrieve event sources");
     }
@@ -96,17 +100,17 @@ public <S> EventSource<S, R> get(Class<S> dependentType, String name) {
     }
 
     final var size = sourcesForType.size();
-    EventSource source;
+    EventSource<S, P> source;
     if (size == 1 && name == null) {
-      source = sourcesForType.values().stream().findFirst().orElseThrow();
+      source = (EventSource<S, P>) sourcesForType.values().stream().findFirst().orElseThrow();
     } else {
       if (name == null || name.isBlank()) {
         throw new IllegalArgumentException("There are multiple EventSources registered for type "
             + dependentType.getCanonicalName()
             + ", you need to provide a name to specify which EventSource you want to query. Known names: "
             + String.join(",", sourcesForType.keySet()));
       }
-      source = sourcesForType.get(name);
+      source = (EventSource<S, P>) sourcesForType.get(name);
 
       if (source == null) {
         throw new IllegalArgumentException("There is no event source found for class:" +
@@ -132,16 +136,17 @@ private String keyAsString(Class dependentType, String name) {
   }
 
   @SuppressWarnings("unchecked")
-  public <S> List<EventSource<S, R>> getEventSources(Class<S> dependentType) {
+  public <S> List<EventSource<S, P>> getEventSources(Class<S> dependentType) {
     final var sourcesForType = sources.get(keyFor(dependentType));
     if (sourcesForType == null) {
       return Collections.emptyList();
     }
 
     return sourcesForType.values().stream()
-        .map(es -> (EventSource<S, R>) es).toList();
+        .map(es -> (EventSource<S, P>) es).toList();
   }
 
+  @SuppressWarnings("rawtypes")
   public EventSource remove(String name) {
     var optionalMap = sources.values().stream().filter(m -> m.containsKey(name)).findFirst();
     return optionalMap.map(m -> m.remove(name)).orElse(null);

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

@@ -44,6 +44,20 @@ default EventSourceStartPriority priority() {
    */
   Class<R> resourceType();
 
+  /**
+   * Retrieves this EventSource's configuration if it exists.
+   *
+   * @return this EventSource's configuration if it exists
+   * @since 5.0.0
+   */
+  @SuppressWarnings({"rawtypes", "unused"})
+  default Optional<?> optionalConfiguration() {
+    if (this instanceof Configurable configurable) {
+      return Optional.ofNullable(configurable.configuration());
+    }
+    return Optional.empty();
+  }
+
   default Optional<R> getSecondaryResource(P primary) {
     var resources = getSecondaryResources(primary);
     if (resources.isEmpty()) {

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

@@ -39,7 +39,7 @@ public void registersEventSource() {
 
     eventSourceManager.registerEventSource(eventSource);
 
-    Set<EventSource> registeredSources = eventSourceManager.getRegisteredEventSources();
+    final var registeredSources = eventSourceManager.getRegisteredEventSources();
     assertThat(registeredSources).contains(eventSource);
 
     verify(eventSource, times(1)).setEventHandler(any());