fix: cleanup & beta

kestra-io · May 15, 2024 · 6a441ff · 6a441ff
1 parent e84dae5
commit 6a441ff
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 17 deletions.
diff --git a/src/main/java/io/kestra/plugin/gcp/pubsub/Consume.java b/src/main/java/io/kestra/plugin/gcp/pubsub/Consume.java
@@ -25,16 +25,12 @@
 import java.net.URI;
 import java.time.Duration;
 import java.time.ZonedDateTime;
-import java.util.Objects;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
 import jakarta.validation.constraints.NotNull;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
-import reactor.core.publisher.FluxSink;
-import reactor.core.scheduler.Scheduler;
-import reactor.core.scheduler.Schedulers;
 
 @SuperBuilder
 @ToString
@@ -136,10 +132,9 @@ public Publisher<Message> stream(RunContext runContext) throws Exception {
 	    ProjectSubscriptionName subscriptionName = this.createSubscription(runContext, subscription, autoCreateSubscription);
         GoogleCredentials credentials = this.credentials(runContext);
 
-        return Flux.<Message>create(
+        return Flux.create(
                 sink -> {
 	                AtomicInteger total = new AtomicInteger();
-	                ZonedDateTime started = ZonedDateTime.now();
 
                     MessageReceiver receiver = (message, consumer) -> {
                         try {
@@ -168,7 +163,7 @@ public void failed(ApiService.State from, Throwable failure) {
                         }, MoreExecutors.directExecutor()
                     );
 
-                    while (!this.ended(total, started)) {
+                    while (true) {
                         if (sink.isCancelled()) {
                             subscriber.stopAsync().awaitTerminated();
                             return;
@@ -177,16 +172,11 @@ public void failed(ApiService.State from, Throwable failure) {
                             Thread.sleep(100);
                         } catch (InterruptedException exception) {
                             Thread.currentThread().interrupt();
+                            subscriber.stopAsync().awaitTerminated();
                             sink.error(exception);
                         }
                     }
-
-                    subscriber.stopAsync().awaitTerminated();
-                    sink.complete();
-                },
-                FluxSink.OverflowStrategy.BUFFER
-            )
-            .subscribeOn(Schedulers.boundedElastic());
+                });
     }
 
     private boolean ended(AtomicInteger count, ZonedDateTime start) {

diff --git a/src/main/java/io/kestra/plugin/gcp/pubsub/RealtimeTrigger.java b/src/main/java/io/kestra/plugin/gcp/pubsub/RealtimeTrigger.java
@@ -38,7 +38,8 @@
                 "maxRecords: 10"
             }
         )
-    }
+    },
+    beta = true
 )
 public class RealtimeTrigger extends AbstractTrigger implements RealtimeTriggerInterface, TriggerOutput<Consume.Output>, PubSubConnectionInterface {
 
@@ -97,7 +98,6 @@ public Publisher<Execution> evaluate(ConditionContext conditionContext, TriggerC
             .build();
 
         return Flux.from(task.stream(conditionContext.getRunContext()))
-            .map(message -> TriggerService.generateRealtimeExecution(this, context, message))
-            .next();
+            .map(message -> TriggerService.generateRealtimeExecution(this, context, message));
     }
 }