Merge pull request #839 from symphony-enrico/improve-datafeed-stop

Better handling stop datafeed

Author: Yannick-Malins

symphony-bdk-core/src/main/java/com/symphony/bdk/core/service/datafeed/impl/AbstractDatafeedLoop.java

@@ -77,7 +77,7 @@ public void unsubscribe(RealTimeEventListener listener) {
    */
   @Override
   public void start() throws AuthUnauthorizedException, ApiException {
-    if (this.started.get()) {
+    if (!this.started.compareAndSet(false, true)) {
       throw new IllegalStateException("The datafeed service is already started");
     }
 
@@ -104,8 +104,11 @@ public void start() throws AuthUnauthorizedException, ApiException {
    */
   @Override
   public void stop() {
-    log.info("Stopping the datafeed loop (will happen once the current read is finished)...");
-    this.started.set(false);
+    if (this.started.compareAndSet(true, false)) {
+      log.info("Stopping the datafeed loop (will happen once the current read is finished)...");
+    } else {
+      log.warn("Datafeed loop already stopping...");
+    }
   }
 
   private void updateLastPullTimestamp() {

symphony-bdk-core/src/main/java/com/symphony/bdk/core/service/datafeed/impl/DatafeedLoopV1.java

@@ -97,10 +97,9 @@ protected void runLoop() throws Throwable {
     this.datafeedId = this.datafeedId == null ? this.createDatafeed.execute() : this.datafeedId;
     log.info("Start reading events from datafeed {}", datafeedId);
 
-    this.started.set(true);
-    do {
+    while (this.started.get()) {
       this.readDatafeed.execute();
-    } while (this.started.get());
+    }
 
     log.info("Datafeed loop successfully stopped.");
   }

symphony-bdk-core/src/main/java/com/symphony/bdk/core/service/datafeed/impl/DatafeedLoopV2.java

@@ -97,12 +97,9 @@ protected void runLoop() throws Throwable {
     }
 
     log.info("Start reading events from datafeed {}", this.datafeed.getId());
-    this.started.set(true);
-    do {
-
+    while (this.started.get()) {
       this.readDatafeed.execute();
-
-    } while (this.started.get());
+    }
     log.info("Datafeed loop successfully stopped.");
   }
 

symphony-bdk-core/src/main/java/com/symphony/bdk/core/service/datafeed/impl/DatahoseLoopImpl.java

@@ -58,11 +58,10 @@ public DatahoseLoopImpl(DatafeedApi datafeedApi, AuthSession authSession, BdkCon
   @Override
   protected void runLoop() throws Throwable {
     log.info("Start reading events from datahose loop");
-    this.started.set(true);
 
-    do {
+    while (this.started.get()) {
       this.readEvents.execute();
-    } while (this.started.get());
+    }
 
     log.info("Datahose loop successfully stopped.");
   }