Merge pull request #103 from bosch-io/bugfix/revert-consumption-fix

Revert consumption fix

Author: Stefan Maute

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

@@ -24,13 +24,10 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
-import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import java.util.function.Consumer;
 import java.util.function.Function;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
 
 import javax.annotation.Nullable;
 
@@ -52,7 +49,6 @@
 import org.eclipse.ditto.client.options.Option;
 import org.eclipse.ditto.client.options.OptionName;
 import org.eclipse.ditto.client.options.internal.OptionsEvaluator;
-import org.eclipse.ditto.client.twin.internal.UncompletedConsumptionRequestException;
 import org.eclipse.ditto.json.JsonFieldSelector;
 import org.eclipse.ditto.json.JsonObject;
 import org.eclipse.ditto.json.JsonPointer;
@@ -645,15 +641,6 @@ protected AdaptableBus.SubscriptionId subscribeAndPublishMessage(
 
         LOGGER.trace("Sending {} and waiting for {}", protocolCommand, protocolCommandAck);
         final AdaptableBus adaptableBus = messagingProvider.getAdaptableBus();
-
-        try {
-            if (previousSubscriptionId != null
-                    && checkIfTwinEventIsInsertedTwiceElseThrow(adaptableBus, futureToCompleteOrFailAfterAck)) {
-                return previousSubscriptionId;
-            }
-        } catch (UncompletedConsumptionRequestException e) {
-            LOGGER.error(e.getMessage());
-        }
         if (previousSubscriptionId != null) {
             // remove previous subscription without going through back-end because subscription will be replaced
             adaptableBus.unsubscribe(previousSubscriptionId);
@@ -664,28 +651,9 @@ && checkIfTwinEventIsInsertedTwiceElseThrow(adaptableBus, futureToCompleteOrFail
         final Classification tag = Classification.forString(protocolCommandAck);
         adjoin(adaptableBus.subscribeOnceForString(tag, getTimeout()), futureToCompleteOrFailAfterAck);
         messagingProvider.emit(protocolCommand);
-
         return subscriptionId;
     }
 
-    private boolean checkIfTwinEventIsInsertedTwiceElseThrow(final AdaptableBus adaptableBus,
-            final CompletableFuture<Void> futureToCompleteOrFailAfterAck) {
-
-        final Set<Classification> stringList = Stream.of(
-                Classification.forString(Classification.StreamingType.TWIN_EVENT.startAck()),
-                Classification.forString(Classification.StreamingType.LIVE_COMMAND.startAck()),
-                Classification.forString(Classification.StreamingType.LIVE_EVENT.startAck()),
-                Classification.forString(Classification.StreamingType.LIVE_MESSAGE.startAck()))
-                .collect(Collectors.toSet());
-
-        if (adaptableBus.getUnmodifiableOneTimeStringConsumers().keySet().stream().anyMatch(stringList::contains)) {
-            LOGGER.warn("First consumption request on this channel must be completed first");
-            futureToCompleteOrFailAfterAck.completeExceptionally(new UncompletedConsumptionRequestException());
-            return true;
-        }
-        return false;
-    }
-
     /**
      * Remove a subscription.
      *

java/src/main/java/org/eclipse/ditto/client/internal/bus/AdaptableBus.java

@@ -13,8 +13,6 @@
 package org.eclipse.ditto.client.internal.bus;
 
 import java.time.Duration;
-import java.util.Map;
-import java.util.Set;
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.function.Consumer;
@@ -25,7 +23,7 @@
 import org.eclipse.ditto.protocoladapter.Adaptable;
 
 /**
- * Event bus for messages that are either {@code String} or {@code Adaptable}.
+ * Event bus for messages that are either {@code String} or {@code} Adaptable.
  * On publication of a message as {@code String}, subscribers are notified as follows:
  * <ol>
  * <li>Message is classified as {@code String}. If a matching one-time subscriber is found, the subscriber is notified
@@ -55,14 +53,6 @@ public interface AdaptableBus {
      */
     AdaptableBus addAdaptableClassifier(Classifier<Adaptable> adaptableClassifier);
 
-
-    /**
-     * Get oneTimeStringConsumers but unmodifiable to only grant read access.
-     *
-     * @return a {@code UnmodifiableMap}
-     */
-    Map<Classification, Set<Entry<Consumer<String>>>> getUnmodifiableOneTimeStringConsumers();
-
     /**
      * Add a one-time subscriber for a string message.
      *

java/src/main/java/org/eclipse/ditto/client/internal/bus/DefaultAdaptableBus.java

@@ -15,7 +15,6 @@
 import java.time.Duration;
 import java.time.Instant;
 import java.util.Collection;
-import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -85,11 +84,6 @@ public AdaptableBus addAdaptableClassifier(final Classifier<Adaptable> adaptable
         return this;
     }
 
-    @Override
-    public final Map<Classification, Set<Entry<Consumer<String>>>> getUnmodifiableOneTimeStringConsumers() {
-        return Collections.unmodifiableMap(oneTimeStringConsumers);
-    }
-
     @Override
     public CompletionStage<String> subscribeOnceForString(final Classification tag, final Duration timeout) {
         return subscribeOnce(oneTimeStringConsumers, tag, timeout);
@@ -262,7 +256,7 @@ private boolean publishToPersistentAdaptableSubscribers(final Adaptable adaptabl
             if (persistentConsumers != null && !persistentConsumers.isEmpty()) {
                 publishedToPersistentSubscribers = true;
                 for (final Entry<Consumer<Adaptable>> entry : persistentConsumers) {
-                    runConsumerAsync(entry.getValue(), adaptable, tag);
+                    runConsumerAsync(entry.value, adaptable, tag);
                 }
             }
         }
@@ -314,7 +308,7 @@ private <T> void removeAfterIdle(
 
     private static <T> void addEntry(final Map<Classification, Set<Entry<T>>> registry,
             final Entry<T> entry) {
-        registry.compute(entry.getKey(), (key, previousSet) -> {
+        registry.compute(entry.key, (key, previousSet) -> {
             final Set<Entry<T>> concurrentHashSet =
                     previousSet != null ? previousSet : ConcurrentHashMap.newKeySet();
             concurrentHashSet.add(entry);
@@ -325,7 +319,7 @@ private static <T> void addEntry(final Map<Classification, Set<Entry<T>>> regist
     private static <T> void replaceEntry(final Map<Classification, Set<Entry<T>>> registry, final Entry<T> entry) {
         final Set<Entry<T>> set = ConcurrentHashMap.newKeySet();
         set.add(entry);
-        registry.put(entry.getKey(), set);
+        registry.put(entry.key, set);
     }
 
     private Optional<Adaptable> parseAsAdaptable(final String message) {
@@ -345,7 +339,7 @@ private Optional<Adaptable> parseAsAdaptable(final String message) {
     private <T> void removeEntry(final Map<Classification, Set<Entry<T>>> registry,
             final Entry<?> entry,
             final Runnable onRemove) {
-        registry.computeIfPresent(entry.getKey(), (key, set) -> {
+        registry.computeIfPresent(entry.key, (key, set) -> {
             if (set.remove(entry)) {
                 onRemove.run();
             }
@@ -365,7 +359,7 @@ private static <T> T removeOne(final Map<Classification, Set<Entry<T>>> registry
                 .findAny()
                 .map(entry -> {
                     if (set.remove(entry)) {
-                        result.set(entry.getValue());
+                        result.set(entry.value);
                     }
                     return set.isEmpty() ? null : set;
                 })
@@ -377,4 +371,18 @@ private static Throwable timeout(final Duration duration) {
         return new TimeoutException("Timed out after " + duration);
     }
 
+    /**
+     * Similar to Map.Entry but with object reference identity and fixed key type to act as identifier for
+     * a subscription.
+     */
+    private static final class Entry<T> implements SubscriptionId {
+
+        private final Classification key;
+        private final T value;
+
+        private Entry(final Classification key, final T value) {
+            this.key = key;
+            this.value = value;
+        }
+    }
 }

java/src/main/java/org/eclipse/ditto/client/internal/bus/Entry.java

@@ -13,18 +13,11 @@
 package org.eclipse.ditto.client;
 
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 import static org.eclipse.ditto.client.TestConstants.Thing.THING_ID;
 import static org.eclipse.ditto.model.base.acks.AcknowledgementRequest.parseAcknowledgementRequest;
 
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-
 import org.assertj.core.api.Assertions;
 import org.eclipse.ditto.client.internal.AbstractDittoClientTest;
-import org.eclipse.ditto.client.twin.internal.UncompletedConsumptionRequestException;
 import org.eclipse.ditto.json.JsonPointer;
 import org.eclipse.ditto.json.JsonValue;
 import org.eclipse.ditto.model.base.common.HttpStatusCode;
@@ -104,38 +97,6 @@ public void testAttributeEventAcknowledgement() {
                 .isEqualTo(HttpStatusCode.INTERNAL_SERVER_ERROR);
     }
 
-    @Test
-    public void startConsumptionParallelOnSameTwinChannelShouldThrowException()
-            throws InterruptedException, ExecutionException, TimeoutException {
-
-        final CompletableFuture<Void> voidCompletableFuture1 = client.twin().startConsumption();
-        final CompletableFuture<Void> voidCompletableFuture2 = client.twin().startConsumption();
-
-        messaging.receivePlainString("START-SEND-EVENTS:ACK");
-
-        voidCompletableFuture1.get(10, TimeUnit.SECONDS);
-        assertThatExceptionOfType(ExecutionException.class)
-                .isThrownBy(() -> voidCompletableFuture2.get(10, TimeUnit.SECONDS))
-                .withCauseInstanceOf(UncompletedConsumptionRequestException.class);
-    }
-
-    @Test
-    public void startConsumptionParallelOnSameLiveChannelShouldThrowException()
-            throws InterruptedException, ExecutionException, TimeoutException {
-
-        final CompletableFuture<Void> voidCompletableFuture1 = client.live().startConsumption();
-        final CompletableFuture<Void> voidCompletableFuture2 = client.live().startConsumption();
-
-        messaging.receivePlainString("START-SEND-LIVE-COMMANDS:ACK");
-        messaging.receivePlainString("START-SEND-LIVE-EVENTS:ACK");
-        messaging.receivePlainString("START-SEND-MESSAGES:ACK");
-
-        voidCompletableFuture1.get(10, TimeUnit.SECONDS);
-        assertThatExceptionOfType(ExecutionException.class)
-                .isThrownBy(() -> voidCompletableFuture2.get(10, TimeUnit.SECONDS))
-                .withCauseInstanceOf(UncompletedConsumptionRequestException.class);
-    }
-
     @Test
     public void testFeatureEventAcknowledgement() {
         client.twin().startConsumption();

java/src/main/java/org/eclipse/ditto/client/twin/internal/UncompletedConsumptionRequestException.java

@@ -150,10 +150,6 @@ public void receiveEvent(final Message<ThingEvent> message) {
         adaptableBus.publish(ProtocolFactory.wrapAsJsonifiableAdaptable(adaptable).toJsonString());
     }
 
-    public void receivePlainString(final String plain) {
-        adaptableBus.publish(plain);
-    }
-
     @Override
     public void close() {
         executor.shutdownNow();