fix in-vm events

apache · Nov 29, 2023 · 22c6248 · 22c6248
1 parent 631b203
commit 22c6248
Show file tree

Hide file tree

Showing 19 changed files with 323 additions and 277 deletions.
diff --git a/...-index-persistence/kogito-addons-quarkus-data-index-persistence-common/runtime/.gitignore b/...-index-persistence/kogito-addons-quarkus-data-index-persistence-common/runtime/.gitignore
@@ -0,0 +1 @@
+/target/
diff --git a/...gito-addons-quarkus-data-index/kogito-addons-quarkus-data-index-common/runtime/.gitignore b/...gito-addons-quarkus-data-index/kogito-addons-quarkus-data-index-common/runtime/.gitignore
@@ -0,0 +1 @@
+/target/
diff --git a/...ervice/jobs-service-common/src/main/java/org/kie/kogito/jobs/service/job/DelegateJob.java b/...ervice/jobs-service-common/src/main/java/org/kie/kogito/jobs/service/job/DelegateJob.java
@@ -24,7 +24,7 @@
 import org.kie.kogito.jobs.service.executor.JobExecutorResolver;
 import org.kie.kogito.jobs.service.model.JobDetailsContext;
 import org.kie.kogito.jobs.service.model.JobExecutionResponse;
-import org.kie.kogito.jobs.service.stream.JobStreams;
+import org.kie.kogito.jobs.service.stream.JobEventPublisher;
 import org.kie.kogito.timer.Job;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -40,9 +40,9 @@ public class DelegateJob implements Job<JobDetailsContext> {
 
     private final JobExecutorResolver jobExecutorResolver;
 
-    private final JobStreams jobStreams;
+    private final JobEventPublisher jobStreams;
 
-    public DelegateJob(JobExecutorResolver executorResolver, JobStreams jobStreams) {
+    public DelegateJob(JobExecutorResolver executorResolver, JobEventPublisher jobStreams) {
         this.jobExecutorResolver = executorResolver;
         this.jobStreams = jobStreams;
     }

diff --git a/.../src/main/java/org/kie/kogito/jobs/service/repository/impl/BaseReactiveJobRepository.java b/.../src/main/java/org/kie/kogito/jobs/service/repository/impl/BaseReactiveJobRepository.java
@@ -30,17 +30,17 @@
 import org.kie.kogito.jobs.service.model.JobDetails;
 import org.kie.kogito.jobs.service.model.JobStatus;
 import org.kie.kogito.jobs.service.repository.ReactiveJobRepository;
-import org.kie.kogito.jobs.service.stream.JobStreams;
+import org.kie.kogito.jobs.service.stream.JobEventPublisher;
 
 import io.vertx.core.Vertx;
 
 public abstract class BaseReactiveJobRepository implements ReactiveJobRepository {
 
     private Vertx vertx;
 
-    private JobStreams jobStreams;
+    private JobEventPublisher jobStreams;
 
-    protected BaseReactiveJobRepository(Vertx vertx, JobStreams jobStreams) {
+    protected BaseReactiveJobRepository(Vertx vertx, JobEventPublisher jobStreams) {
         this.vertx = vertx;
         this.jobStreams = jobStreams;
     }

diff --git a/...mmon/src/main/java/org/kie/kogito/jobs/service/repository/impl/InMemoryJobRepository.java b/...mmon/src/main/java/org/kie/kogito/jobs/service/repository/impl/InMemoryJobRepository.java
@@ -36,7 +36,7 @@
 import org.kie.kogito.jobs.service.model.JobDetails;
 import org.kie.kogito.jobs.service.model.JobStatus;
 import org.kie.kogito.jobs.service.repository.ReactiveJobRepository;
-import org.kie.kogito.jobs.service.stream.JobStreams;
+import org.kie.kogito.jobs.service.stream.JobEventPublisher;
 import org.kie.kogito.jobs.service.utils.DateUtil;
 
 import io.quarkus.arc.DefaultBean;
@@ -53,7 +53,7 @@ public InMemoryJobRepository() {
     }
 
     @Inject
-    public InMemoryJobRepository(Vertx vertx, JobStreams jobStreams) {
+    public InMemoryJobRepository(Vertx vertx, JobEventPublisher jobStreams) {
         super(vertx, jobStreams);
     }
 

diff --git a/...n/src/main/java/org/kie/kogito/jobs/service/scheduler/impl/TimerDelegateJobScheduler.java b/...n/src/main/java/org/kie/kogito/jobs/service/scheduler/impl/TimerDelegateJobScheduler.java
@@ -20,27 +20,21 @@
 
 import java.util.Objects;
 import java.util.Optional;
-import java.util.concurrent.CompletionStage;
 
 import javax.enterprise.context.ApplicationScoped;
 import javax.inject.Inject;
 
 import org.eclipse.microprofile.config.inject.ConfigProperty;
-import org.eclipse.microprofile.reactive.messaging.Acknowledgment;
-import org.eclipse.microprofile.reactive.messaging.Incoming;
 import org.eclipse.microprofile.reactive.streams.operators.PublisherBuilder;
 import org.eclipse.microprofile.reactive.streams.operators.ReactiveStreams;
 import org.kie.kogito.jobs.service.executor.JobExecutorResolver;
 import org.kie.kogito.jobs.service.job.DelegateJob;
 import org.kie.kogito.jobs.service.model.JobDetails;
 import org.kie.kogito.jobs.service.model.JobDetailsContext;
-import org.kie.kogito.jobs.service.model.JobExecutionResponse;
 import org.kie.kogito.jobs.service.model.ManageableJobHandle;
 import org.kie.kogito.jobs.service.repository.ReactiveJobRepository;
 import org.kie.kogito.jobs.service.scheduler.BaseTimerJobScheduler;
-import org.kie.kogito.jobs.service.stream.AvailableStreams;
-import org.kie.kogito.jobs.service.stream.JobStreams;
-import org.kie.kogito.jobs.service.utils.ErrorHandling;
+import org.kie.kogito.jobs.service.stream.JobEventPublisher;
 import org.kie.kogito.timer.Trigger;
 import org.reactivestreams.Publisher;
 import org.slf4j.Logger;
@@ -58,7 +52,7 @@ public class TimerDelegateJobScheduler extends BaseTimerJobScheduler {
 
     private VertxTimerServiceScheduler delegate;
 
-    private JobStreams jobStreams;
+    private JobEventPublisher jobStreams;
 
     protected TimerDelegateJobScheduler() {
     }
@@ -70,7 +64,7 @@ public TimerDelegateJobScheduler(ReactiveJobRepository jobRepository,
             @ConfigProperty(name = "kogito.jobs-service.schedulerChunkInMinutes", defaultValue = "10") long schedulerChunkInMinutes,
             @ConfigProperty(name = "kogito.jobs-service.forceExecuteExpiredJobs", defaultValue = "true") boolean forceExecuteExpiredJobs,
             JobExecutorResolver jobExecutorResolver, VertxTimerServiceScheduler delegate,
-            JobStreams jobStreams) {
+            JobEventPublisher jobStreams) {
         super(jobRepository, backoffRetryMillis, maxIntervalLimitToRetryMillis, schedulerChunkInMinutes, forceExecuteExpiredJobs);
         this.jobExecutorResolver = jobExecutorResolver;
         this.delegate = delegate;
@@ -100,33 +94,4 @@ public Publisher<ManageableJobHandle> doCancel(JobDetails scheduledJob) {
                 .buildRs();
     }
 
-    //Stream Processors
-
-    @Incoming(AvailableStreams.JOB_ERROR_EVENTS)
-    @Acknowledgment(Acknowledgment.Strategy.PRE_PROCESSING)
-    public CompletionStage<Boolean> jobErrorProcessor(JobExecutionResponse response) {
-        LOGGER.warn("Error received {}", response);
-        return ErrorHandling.skipErrorPublisherBuilder(this::handleJobExecutionError, response)
-                .findFirst()
-                .run()
-                .thenApply(Optional::isPresent)
-                .exceptionally(e -> {
-                    LOGGER.error("Error handling error {}", response, e);
-                    return false;
-                });
-    }
-
-    @Incoming(AvailableStreams.JOB_SUCCESS_EVENTS)
-    @Acknowledgment(Acknowledgment.Strategy.PRE_PROCESSING)
-    public CompletionStage<Boolean> jobSuccessProcessor(JobExecutionResponse response) {
-        LOGGER.debug("Success received to be processed {}", response);
-        return ErrorHandling.skipErrorPublisherBuilder(this::handleJobExecutionSuccess, response)
-                .findFirst()
-                .run()
-                .thenApply(Optional::isPresent)
-                .exceptionally(e -> {
-                    LOGGER.error("Error handling error {}", response, e);
-                    return false;
-                });
-    }
 }
diff --git a/...bs-service-common/src/main/java/org/kie/kogito/jobs/service/stream/JobEventPublisher.java b/...bs-service-common/src/main/java/org/kie/kogito/jobs/service/stream/JobEventPublisher.java
@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.kie.kogito.jobs.service.stream;
+
+import org.kie.kogito.jobs.service.model.JobDetails;
+import org.kie.kogito.jobs.service.model.JobExecutionResponse;
+
+public interface JobEventPublisher {
+
+    JobExecutionResponse publishJobError(JobExecutionResponse response);
+
+    JobExecutionResponse publishJobSuccess(JobExecutionResponse response);
+
+    JobDetails publishJobStatusChange(JobDetails scheduledJob);
+}
diff --git a/...ogito/jobs/service/stream/JobStreams.java → ...vice/stream/JobStreamsEventPublisher.java b/...ogito/jobs/service/stream/JobStreams.java → ...vice/stream/JobStreamsEventPublisher.java
@@ -18,6 +18,9 @@
  */
 package org.kie.kogito.jobs.service.stream;
 
+import java.util.Optional;
+import java.util.concurrent.CompletionStage;
+
 import javax.enterprise.context.ApplicationScoped;
 import javax.inject.Inject;
 
@@ -29,6 +32,8 @@
 import org.eclipse.microprofile.reactive.messaging.Outgoing;
 import org.kie.kogito.jobs.service.model.JobDetails;
 import org.kie.kogito.jobs.service.model.JobExecutionResponse;
+import org.kie.kogito.jobs.service.scheduler.ReactiveJobScheduler;
+import org.kie.kogito.jobs.service.utils.ErrorHandling;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -39,9 +44,12 @@
  * received item.
  */
 @ApplicationScoped
-public class JobStreams {
+public class JobStreamsEventPublisher implements JobEventPublisher {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(JobStreamsEventPublisher.class);
 
-    private static final Logger LOGGER = LoggerFactory.getLogger(JobStreams.class);
+    @Inject
+    ReactiveJobScheduler scheduler;
 
     /**
      * Publish on Stream of Job Error events
@@ -82,8 +90,36 @@ public JobDetails publishJobStatusChange(JobDetails scheduledJob) {
         return scheduledJob;
     }
 
-    //Broadcast Events from Emitter to Streams
+    //Stream Processors
+    @Incoming(AvailableStreams.JOB_ERROR_EVENTS)
+    @Acknowledgment(Acknowledgment.Strategy.PRE_PROCESSING)
+    public CompletionStage<Boolean> jobErrorProcessor(JobExecutionResponse response) {
+        LOGGER.warn("Error received {}", response);
+        return ErrorHandling.skipErrorPublisherBuilder(scheduler::handleJobExecutionError, response)
+                .findFirst()
+                .run()
+                .thenApply(Optional::isPresent)
+                .exceptionally(e -> {
+                    LOGGER.error("Error handling error {}", response, e);
+                    return false;
+                });
+    }
+
+    @Incoming(AvailableStreams.JOB_SUCCESS_EVENTS)
+    @Acknowledgment(Acknowledgment.Strategy.PRE_PROCESSING)
+    public CompletionStage<Boolean> jobSuccessProcessor(JobExecutionResponse response) {
+        LOGGER.debug("Success received to be processed {}", response);
+        return ErrorHandling.skipErrorPublisherBuilder(scheduler::handleJobExecutionSuccess, response)
+                .findFirst()
+                .run()
+                .thenApply(Optional::isPresent)
+                .exceptionally(e -> {
+                    LOGGER.error("Error handling error {}", response, e);
+                    return false;
+                });
+    }
 
+    // Broadcast Events from Emitter to Streams
     @Incoming(AvailableStreams.JOB_ERROR)
     @Outgoing(AvailableStreams.JOB_ERROR_EVENTS)
     @Acknowledgment(Acknowledgment.Strategy.PRE_PROCESSING)

diff --git a/...mmon/src/test/java/org/kie/kogito/jobs/service/repository/impl/BaseJobRepositoryTest.java b/...mmon/src/test/java/org/kie/kogito/jobs/service/repository/impl/BaseJobRepositoryTest.java
@@ -34,7 +34,7 @@
 import org.kie.kogito.jobs.service.model.Recipient;
 import org.kie.kogito.jobs.service.model.RecipientInstance;
 import org.kie.kogito.jobs.service.repository.ReactiveJobRepository;
-import org.kie.kogito.jobs.service.stream.JobStreams;
+import org.kie.kogito.jobs.service.stream.JobEventPublisher;
 import org.kie.kogito.jobs.service.utils.DateUtil;
 import org.kie.kogito.jobs.service.utils.FunctionsUtil;
 import org.kie.kogito.timer.impl.PointInTimeTrigger;
@@ -56,8 +56,8 @@ public void setUp() throws Exception {
         createAndSaveJob(ID);
     }
 
-    public JobStreams mockJobStreams() {
-        final JobStreams mock = mock(JobStreams.class);
+    public JobEventPublisher mockJobStreams() {
+        final JobEventPublisher mock = mock(JobEventPublisher.class);
         lenient().when(mock.publishJobStatusChange(any(JobDetails.class))).thenAnswer(a -> a.getArgument(0));
         lenient().when(mock.publishJobSuccess(any(JobExecutionResponse.class))).thenAnswer(a -> a.getArgument(0));
         lenient().when(mock.publishJobError(any(JobExecutionResponse.class))).thenAnswer(a -> a.getArgument(0));

diff --git a/...c/test/java/org/kie/kogito/jobs/service/scheduler/impl/TimerDelegateJobSchedulerTest.java b/...c/test/java/org/kie/kogito/jobs/service/scheduler/impl/TimerDelegateJobSchedulerTest.java
@@ -22,7 +22,6 @@
 import java.util.UUID;
 
 import org.eclipse.microprofile.reactive.streams.operators.PublisherBuilder;
-import org.eclipse.microprofile.reactive.streams.operators.ReactiveStreams;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -45,10 +44,8 @@
 
 import io.smallrye.mutiny.Multi;
 
-import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.eq;
-import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.lenient;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
@@ -98,42 +95,6 @@ void testDoCancelNullId() {
         verify(timer, never()).removeJob(any(ManageableJobHandle.class));
     }
 
-    @Test
-    void testJobSuccessProcessor() {
-        JobExecutionResponse response = getJobResponse();
-        doReturn(ReactiveStreams.of(JobDetails.builder().build()))
-                .when(tested).handleJobExecutionSuccess(response);
-        tested.jobSuccessProcessor(response).thenAccept(r -> assertThat(r).isTrue());
-        verify(tested).handleJobExecutionSuccess(response);
-    }
-
-    @Test
-    void testJobSuccessProcessorFail() {
-        JobExecutionResponse response = getJobResponse();
-        doReturn(ReactiveStreams.failed(new RuntimeException()))
-                .when(tested).handleJobExecutionSuccess(response);
-        tested.jobSuccessProcessor(response).thenAccept(r -> assertThat(r).isFalse());
-        verify(tested).handleJobExecutionSuccess(response);
-    }
-
-    @Test
-    void testJobErrorProcessor() {
-        JobExecutionResponse response = getJobResponse();
-        doReturn(ReactiveStreams.of(JobDetails.builder().build()))
-                .when(tested).handleJobExecutionError(response);
-        tested.jobErrorProcessor(response).thenAccept(r -> assertThat(r).isTrue());
-        verify(tested).handleJobExecutionError(response);
-    }
-
-    @Test
-    void testJobErrorProcessorFail() {
-        JobExecutionResponse response = getJobResponse();
-        doReturn(ReactiveStreams.failed(new RuntimeException()))
-                .when(tested).handleJobExecutionError(response);
-        tested.jobErrorProcessor(response).thenAccept(r -> assertThat(r).isFalse());
-        verify(tested).handleJobExecutionError(response);
-    }
-
     private JobExecutionResponse getJobResponse() {
         return JobExecutionResponse.builder()
                 .jobId(UUID.randomUUID().toString())

diff --git a/.../test/java/org/kie/kogito/jobs/service/scheduler/impl/VertxTimerServiceSchedulerTest.java b/.../test/java/org/kie/kogito/jobs/service/scheduler/impl/VertxTimerServiceSchedulerTest.java
@@ -29,7 +29,7 @@
 import org.kie.kogito.jobs.service.model.JobDetails;
 import org.kie.kogito.jobs.service.model.JobDetailsContext;
 import org.kie.kogito.jobs.service.model.ManageableJobHandle;
-import org.kie.kogito.jobs.service.stream.JobStreams;
+import org.kie.kogito.jobs.service.stream.JobEventPublisher;
 import org.kie.kogito.jobs.service.utils.DateUtil;
 import org.kie.kogito.timer.Job;
 import org.kie.kogito.timer.JobContext;
@@ -62,7 +62,7 @@ class VertxTimerServiceSchedulerTest {
     private JobExecutorResolver jobExecutorResolver;
 
     @Mock
-    private JobStreams jobStreams;
+    private JobEventPublisher jobStreams;
 
     @Captor
     private ArgumentCaptor<JobDetails> jobCaptor;

diff --git a/.../main/java/org/kie/kogito/jobs/service/repository/infinispan/InfinispanJobRepository.java b/.../main/java/org/kie/kogito/jobs/service/repository/infinispan/InfinispanJobRepository.java
@@ -39,7 +39,7 @@
 import org.kie.kogito.jobs.service.model.JobStatus;
 import org.kie.kogito.jobs.service.repository.ReactiveJobRepository;
 import org.kie.kogito.jobs.service.repository.impl.BaseReactiveJobRepository;
-import org.kie.kogito.jobs.service.stream.JobStreams;
+import org.kie.kogito.jobs.service.stream.JobEventPublisher;
 
 import io.vertx.core.Vertx;
 
@@ -58,7 +58,7 @@ public class InfinispanJobRepository extends BaseReactiveJobRepository implement
 
     @Inject
     public InfinispanJobRepository(Vertx vertx,
-            JobStreams jobStreams,
+            JobEventPublisher jobStreams,
             RemoteCacheManager remoteCacheManager) {
         super(vertx, jobStreams);
         this.remoteCacheManager = remoteCacheManager;

diff --git a/...db/src/main/java/org/kie/kogito/jobs/service/repository/mongodb/MongoDBJobRepository.java b/...db/src/main/java/org/kie/kogito/jobs/service/repository/mongodb/MongoDBJobRepository.java
@@ -34,7 +34,7 @@
 import org.kie.kogito.jobs.service.repository.ReactiveJobRepository;
 import org.kie.kogito.jobs.service.repository.impl.BaseReactiveJobRepository;
 import org.kie.kogito.jobs.service.repository.marshaller.JobDetailsMarshaller;
-import org.kie.kogito.jobs.service.stream.JobStreams;
+import org.kie.kogito.jobs.service.stream.JobStreamsEventPublisher;
 
 import com.mongodb.client.model.FindOneAndReplaceOptions;
 
@@ -86,7 +86,7 @@ public class MongoDBJobRepository extends BaseReactiveJobRepository implements R
     }
 
     @Inject
-    public MongoDBJobRepository(Vertx vertx, JobStreams jobStreams, ReactiveMongoClient mongoClient,
+    public MongoDBJobRepository(Vertx vertx, JobStreamsEventPublisher jobStreams, ReactiveMongoClient mongoClient,
             @ConfigProperty(name = DATABASE_PROPERTY) String database,
             JobDetailsMarshaller jobDetailsMarshaller) {
         super(vertx, jobStreams);