Merge pull request eclipse-ditto#5 from bosch-iot-things/feature/ditt…

…o-client-completion-stage

Replace return of CompletableFuture with CompletionStage

java/src/main/java/org/eclipse/ditto/client/DittoClient.java

@@ -12,7 +12,7 @@
  */
 package org.eclipse.ditto.client;
 
-import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
 
 import org.eclipse.ditto.client.live.Live;
 import org.eclipse.ditto.client.policies.Policies;
@@ -52,10 +52,10 @@ public interface DittoClient {
      * Directly sends a Ditto Protocol {@link Adaptable} message to the established Ditto backend connection.
      *
      * @param dittoProtocolAdaptable the adaptable to send
-     * @return a CompletableFuture containing the correlated response to the sent {@code dittoProtocolAdaptable}
+     * @return a CompletionStage containing the correlated response to the sent {@code dittoProtocolAdaptable}
      * @throws IllegalStateException when no twin/live connection was configured for this client
      */
-    CompletableFuture<Adaptable> sendDittoProtocol(Adaptable dittoProtocolAdaptable);
+    CompletionStage<Adaptable> sendDittoProtocol(Adaptable dittoProtocolAdaptable);
 
     /**
      * Returns the client's {@link Policies} singleton which provides the necessary functionality to manage and monitor

java/src/main/java/org/eclipse/ditto/client/internal/DefaultDittoClient.java

@@ -14,7 +14,6 @@
 
 import java.text.MessageFormat;
 import java.util.Optional;
-import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import java.util.function.Consumer;
 
@@ -151,7 +150,7 @@ public Policies policies() {
     }
 
     @Override
-    public CompletableFuture<Adaptable> sendDittoProtocol(final Adaptable dittoProtocolAdaptable) {
+    public CompletionStage<Adaptable> sendDittoProtocol(final Adaptable dittoProtocolAdaptable) {
 
         final TopicPath.Group group = dittoProtocolAdaptable.getTopicPath().getGroup();
         switch (group) {
@@ -164,7 +163,7 @@ public CompletableFuture<Adaptable> sendDittoProtocol(final Adaptable dittoProto
         }
     }
 
-    private CompletableFuture<Adaptable> sendDittoProtocolForThingsGroup(final Adaptable dittoProtocolAdaptable) {
+    private CompletionStage<Adaptable> sendDittoProtocolForThingsGroup(final Adaptable dittoProtocolAdaptable) {
         final TopicPath.Channel channel = dittoProtocolAdaptable.getTopicPath().getChannel();
         switch (channel) {
             case TWIN:
@@ -176,7 +175,7 @@ private CompletableFuture<Adaptable> sendDittoProtocolForThingsGroup(final Adapt
         }
     }
 
-    private CompletableFuture<Adaptable> sendDittoProtocolForPoliciesGroup(final Adaptable dittoProtocolAdaptable) {
+    private CompletionStage<Adaptable> sendDittoProtocolForPoliciesGroup(final Adaptable dittoProtocolAdaptable) {
         final TopicPath.Channel channel = dittoProtocolAdaptable.getTopicPath().getChannel();
         if (TopicPath.Channel.NONE.equals(channel)) {
             return policies.getMessagingProvider().sendAdaptable(dittoProtocolAdaptable);

java/src/main/java/org/eclipse/ditto/client/live/Live.java

@@ -12,7 +12,7 @@
  */
 package org.eclipse.ditto.client.live;
 
-import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
 
 import org.eclipse.ditto.client.live.commands.FeaturePropertiesCommandHandling;
 import org.eclipse.ditto.client.live.commands.FeaturesCommandHandling;
@@ -60,11 +60,11 @@ public interface Live extends CommonManagement<LiveThingHandle, LiveFeatureHandl
     /**
      * Start consuming changes, messages and commands on this {@code live()} channel.
      *
-     * @return a CompletableFuture that terminates when the start operation was successful.
+     * @return a CompletionStage that terminates when the start operation was successful.
      */
     @Override
     // overwritten in order to display a better suiting javadoc for the user
-    CompletableFuture<Void> startConsumption();
+    CompletionStage<Void> startConsumption();
 
     /**
      * Start consuming changes, messages and commands on this {@code live()} channel with the passed {@code
@@ -75,10 +75,10 @@ public interface Live extends CommonManagement<LiveThingHandle, LiveFeatureHandl
      * <pre>{@code Options.Consumption.namespaces("org.eclipse.ditto.namespace1","org.eclipse.ditto.namespace2");
      * Options.Consumption.filter("gt(attributes/counter,42)");}
      * </pre>
-     * @return a CompletableFuture that terminates when the start operation was successful.
+     * @return a CompletionStage that terminates when the start operation was successful.
      */
     @Override
     // overwritten in order to display a better suiting javadoc for the user
-    CompletableFuture<Void> startConsumption(Option<?>... consumptionOptions);
+    CompletionStage<Void> startConsumption(Option<?>... consumptionOptions);
 
 }

java/src/main/java/org/eclipse/ditto/client/live/internal/LiveImpl.java

@@ -17,6 +17,7 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Consumer;
 import java.util.function.Function;
@@ -126,7 +127,7 @@ protected LiveFeatureHandleImpl createFeatureHandle(final ThingId thingId, final
     }
 
     @Override
-    protected CompletableFuture<Void> doStartConsumption(final Map<String, String> consumptionConfig) {
+    protected CompletionStage<Void> doStartConsumption(final Map<String, String> consumptionConfig) {
         final CompletableFuture<Void> completableFutureEvents = new CompletableFuture<>();
         final CompletableFuture<Void> completableFutureMessages = new CompletableFuture<>();
         final CompletableFuture<Void> completableFutureLiveCommands = new CompletableFuture<>();
@@ -175,7 +176,7 @@ protected CompletableFuture<Void> doStartConsumption(final Map<String, String> c
     }
 
     @Override
-    public CompletableFuture<Void> suspendConsumption() {
+    public CompletionStage<Void> suspendConsumption() {
         return CompletableFuture.allOf(
                 subscriptionIds.entrySet()
                         .stream()