once event source

Author: csviri

operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/event/internal/TimerEvent.java

@@ -7,4 +7,5 @@ public class TimerEvent extends AbstractEvent {
     public TimerEvent(String relatedCustomResourceUid, TimerEventSource eventSource) {
         super(relatedCustomResourceUid, eventSource);
     }
+
 }

operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/event/internal/TimerEventSource.java

@@ -16,6 +16,7 @@ public class TimerEventSource extends AbstractEventSource {
 
     private final Timer timer = new Timer();
 
+    private final Map<String, EventProducerTimeTask> onceTasks = new ConcurrentHashMap<>();
     private final Map<String, EventProducerTimeTask> timerTasks = new ConcurrentHashMap<>();
 
     public void schedule(CustomResource customResource, long delay, long period) {
@@ -28,14 +29,34 @@ public void schedule(CustomResource customResource, long delay, long period) {
         timer.schedule(task, delay, period);
     }
 
+    public void scheduleOnce(CustomResource customResource, long delay) {
+        String resourceUid = KubernetesResourceUtils.getUID(customResource);
+        if (onceTasks.containsKey(resourceUid)) {
+            cancelOnceSchedule(resourceUid);
+        }
+        EventProducerTimeTask task = new EventProducerTimeTask(resourceUid);
+        onceTasks.put(resourceUid, task);
+        timer.schedule(task, delay);
+    }
+
     @Override
     public void eventSourceDeRegisteredForResource(String customResourceUid) {
         cancelSchedule(customResourceUid);
+        cancelOnceSchedule(customResourceUid);
     }
 
     public void cancelSchedule(String customResourceUid) {
         TimerTask timerTask = timerTasks.remove(customResourceUid);
-        timerTask.cancel();
+        if (timerTask != null) {
+            timerTask.cancel();
+        }
+    }
+
+    public void cancelOnceSchedule(String customResourceUid) {
+        TimerTask timerTask = onceTasks.remove(customResourceUid);
+        if (timerTask != null) {
+            timerTask.cancel();
+        }
     }
 
     public class EventProducerTimeTask extends TimerTask {

operator-framework/src/test/java/io/javaoperatorsdk/operator/processing/event/internal/TimerEventSourceTest.java

@@ -23,7 +23,6 @@ class TimerEventSourceTest {
 
     private TimerEventSource timerEventSource;
     private EventHandler eventHandlerMock = mock(EventHandler.class);
-    private EventSourceManager eventSourceManagerMock = mock(EventSourceManager.class);
 
     @BeforeEach
     public void setup() {
@@ -58,4 +57,53 @@ public void deRegistersPeriodicalEventSources() throws InterruptedException {
                 .handleEvent(any());
     }
 
+    @Test
+    public void schedulesOnce() throws InterruptedException {
+        CustomResource customResource = TestUtils.testCustomResource();
+
+        timerEventSource.scheduleOnce(customResource, PERIOD);
+
+        Thread.sleep(2 * PERIOD + TESTING_TIME_SLACK);
+        verify(eventHandlerMock, times(1))
+                .handleEvent(any());
+    }
+
+    @Test
+    public void canCancelOnce() throws InterruptedException {
+        CustomResource customResource = TestUtils.testCustomResource();
+
+        timerEventSource.scheduleOnce(customResource, PERIOD);
+        timerEventSource.cancelOnceSchedule(KubernetesResourceUtils.getUID(customResource));
+
+        Thread.sleep(PERIOD + TESTING_TIME_SLACK);
+        verify(eventHandlerMock, never())
+                .handleEvent(any());
+    }
+
+    @Test
+    public void canRescheduleOnceEvent() throws InterruptedException {
+        CustomResource customResource = TestUtils.testCustomResource();
+
+        timerEventSource.scheduleOnce(customResource, PERIOD);
+        timerEventSource.scheduleOnce(customResource, 2 * PERIOD);
+
+        Thread.sleep(PERIOD + TESTING_TIME_SLACK);
+        verify(eventHandlerMock, never())
+                .handleEvent(any());
+        Thread.sleep(PERIOD + TESTING_TIME_SLACK);
+        verify(eventHandlerMock, times(1))
+                .handleEvent(any());
+    }
+
+    @Test
+    public void deRegistersOnceEventSources() throws InterruptedException {
+        CustomResource customResource = TestUtils.testCustomResource();
+
+        timerEventSource.scheduleOnce(customResource, PERIOD);
+        timerEventSource.eventSourceDeRegisteredForResource(getUID(customResource));
+        Thread.sleep(PERIOD + TESTING_TIME_SLACK);
+
+        verify(eventHandlerMock, never())
+                .handleEvent(any());
+    }
 }