-
Notifications
You must be signed in to change notification settings - Fork 25
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #105 from bosch-io/bugfix/handle-consumption-error
Ditto-java-client bugfix: When starting consumption with invalid filter, wrongly timeout exception is propagate to the user
- Loading branch information
Showing
14 changed files
with
596 additions
and
73 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
30 changes: 30 additions & 0 deletions
30
...rc/main/java/org/eclipse/ditto/client/internal/ConcurrentConsumptionRequestException.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,30 @@ | ||
| /* | ||
| * Copyright (c) 2020 Contributors to the Eclipse Foundation | ||
| * | ||
| * See the NOTICE file(s) distributed with this work for additional | ||
| * information regarding copyright ownership. | ||
| * | ||
| * This program and the accompanying materials are made available under the | ||
| * terms of the Eclipse Public License 2.0 which is available at | ||
| * http://www.eclipse.org/legal/epl-2.0 | ||
| * | ||
| * SPDX-License-Identifier: EPL-2.0 | ||
| */ | ||
| package org.eclipse.ditto.client.internal; | ||
|
|
||
| /** | ||
| * This exception may be thrown if concurrent consumption requests are detected. | ||
| */ | ||
| public class ConcurrentConsumptionRequestException extends RuntimeException { | ||
|
|
||
| private static final long serialVersionUID = -565137801315595348L; | ||
| private static final String MESSAGE = "Concurrent consumption requests are not allowed on one channel."; | ||
|
|
||
| /** | ||
| * Constructs a new {@code UncompletedTwinConsumptionRequestException} object. | ||
| */ | ||
| public ConcurrentConsumptionRequestException() { | ||
| super(MESSAGE, null); | ||
| } | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
117 changes: 117 additions & 0 deletions
117
java/src/test/java/org/eclipse/ditto/client/AbstractConsumptionDittoClientTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,117 @@ | ||
| /* | ||
| * Copyright (c) 2020 Contributors to the Eclipse Foundation | ||
| * | ||
| * See the NOTICE file(s) distributed with this work for additional | ||
| * information regarding copyright ownership. | ||
| * | ||
| * This program and the accompanying materials are made available under the | ||
| * terms of the Eclipse Public License 2.0 which is available at | ||
| * http://www.eclipse.org/legal/epl-2.0 | ||
| * | ||
| * SPDX-License-Identifier: EPL-2.0 | ||
| */ | ||
| package org.eclipse.ditto.client; | ||
|
|
||
| import static org.assertj.core.api.Assertions.assertThat; | ||
|
|
||
| import java.util.ArrayList; | ||
| import java.util.Collections; | ||
| import java.util.List; | ||
| import java.util.concurrent.CompletableFuture; | ||
| import java.util.concurrent.ExecutionException; | ||
| import java.util.concurrent.TimeUnit; | ||
| import java.util.stream.Collectors; | ||
|
|
||
| import org.eclipse.ditto.client.internal.AbstractDittoClientTest; | ||
| import org.eclipse.ditto.client.internal.ConcurrentConsumptionRequestException; | ||
| import org.junit.Test; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| /** | ||
| * Tests different sequences of failing/succeeding startConsumption invocations. | ||
| */ | ||
| abstract class AbstractConsumptionDittoClientTest extends AbstractDittoClientTest { | ||
|
|
||
| private static final Logger LOGGER = LoggerFactory.getLogger(AbstractDittoClientTest.class); | ||
|
|
||
| @Test | ||
| public void concurrentStartConsumptionFails() { | ||
| try { | ||
| final CompletableFuture<Void> firstRequest = startConsumptionRequest(); | ||
| final CompletableFuture<Void> concurrentRequest = startConsumptionRequest(); | ||
|
|
||
| replyToConsumptionRequest(); | ||
|
|
||
| assertCompletion(firstRequest); | ||
| concurrentRequest.get(1, TimeUnit.SECONDS); | ||
| } catch (final Exception e) { | ||
| assertThat(e) | ||
| .isInstanceOf(ExecutionException.class) | ||
| .hasCauseInstanceOf(ConcurrentConsumptionRequestException.class); | ||
| } | ||
| } | ||
|
|
||
| protected abstract CompletableFuture<Void> startConsumptionRequest(); | ||
|
|
||
| protected abstract void replyToConsumptionRequest(); | ||
|
|
||
| @Test | ||
| public void testStartConsumptionCombinations() { | ||
| // test all combinations of startConsumption invocations expecting success or failure of the given length | ||
| testStartConsumptionSequence(4, Collections.emptyList()); | ||
| } | ||
|
|
||
| protected void testStartConsumptionSequence(int remaining, final List<Boolean> sequence) { | ||
| testSequence(sequence); | ||
| if (remaining > 0) { | ||
| testStartConsumptionSequence(remaining - 1, addElement(sequence, true)); | ||
| testStartConsumptionSequence(remaining - 1, addElement(sequence, false)); | ||
| } | ||
| } | ||
|
|
||
| private void testSequence(final List<Boolean> sequence) { | ||
| if (sequence.isEmpty()) { | ||
| return; | ||
| } | ||
| LOGGER.info("Testing startConsumption sequence: {}", | ||
| sequence.stream() | ||
| .map(success -> success ? "expect success" : "expect failure") | ||
| .collect(Collectors.joining(" -> "))); | ||
| sequence.forEach(expectSuccess -> { | ||
| messaging.clearEmitted(); | ||
| if (expectSuccess) { | ||
| startConsumptionSucceeds(); | ||
| } else { | ||
| startConsumptionAndExpectError(); | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| private List<Boolean> addElement(final List<Boolean> sequence, final boolean b) { | ||
| final ArrayList<Boolean> success = new ArrayList<>(sequence); | ||
| success.add(b); | ||
| return success; | ||
| } | ||
|
|
||
| protected void assertFailedCompletion(final CompletableFuture<Void> future, | ||
| final Class<? extends Exception> exception) { | ||
| try { | ||
| future.get(1L, TimeUnit.SECONDS); | ||
| } catch (final Exception e) { | ||
| assertThat(e) | ||
| .isInstanceOf(ExecutionException.class) | ||
| .hasCauseInstanceOf(exception); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Tests a succeeding startConsumption invocation. | ||
| */ | ||
| protected abstract void startConsumptionSucceeds(); | ||
|
|
||
| /** | ||
| * Tests a failing startConsumption invocation. | ||
| */ | ||
| protected abstract void startConsumptionAndExpectError(); | ||
| } |
Oops, something went wrong.