Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Switched to generic implementation of MQTT with backpressure.
* Legacy MQTT implementation still has to be removed after testing. * Refactored `GenericMqttClientFactory` to be a utility class instead of an interface because it did not make sense to have an instance of this factory. This entailed some static mocking in `GenericMqttClientActorTest` and `ConnectionTesterTest`. * Provide `MqttSpecificConfig` to `HiveMqttClientProperties` because this config is required in `GenericMqttActor` anyway. This prevents duplicate creation of `MqttSpecificConfig`. * Implemented possibility of reconnecting of consumer client to `GenericMqttActor`. Signed-off-by: Juergen Fickel <juergen.fickel@bosch.io>
- Loading branch information
Juergen Fickel
committed
May 31, 2022
1 parent
f967c0b
commit 2253bd4
Showing
25 changed files
with
939 additions
and
459 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
94 changes: 94 additions & 0 deletions
94
...e/src/main/java/org/eclipse/ditto/connectivity/service/messaging/mqtt/ReconnectDelay.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,94 @@ | ||
| /* | ||
| * Copyright (c) 2022 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.connectivity.service.messaging.mqtt; | ||
|
|
||
| import java.time.Duration; | ||
| import java.util.Objects; | ||
|
|
||
| import javax.annotation.Nullable; | ||
| import javax.annotation.concurrent.Immutable; | ||
|
|
||
| import org.eclipse.ditto.base.model.common.ConditionChecker; | ||
|
|
||
| /** | ||
| * Delay how long to wait before reconnecting an MQTT client. | ||
| * This delay is guaranteed | ||
| */ | ||
| @Immutable | ||
| public final class ReconnectDelay implements Comparable<ReconnectDelay> { | ||
|
|
||
| /** | ||
| * The lower boundary a {@code ReconnectDelay} a never falls below. | ||
| */ | ||
| static final Duration LOWER_BOUNDARY = Duration.ofSeconds(1L); | ||
|
|
||
| private final Duration duration; | ||
|
|
||
| private ReconnectDelay(final Duration duration) { | ||
| this.duration = duration; | ||
| } | ||
|
|
||
| /** | ||
| * Returns an instance of {@code ReconnectDelay} for the specified {@code Duration} argument or | ||
| * {@link #LOWER_BOUNDARY} if the argument is less than the lower boundary. | ||
| * | ||
| * @param duration the duration of the returned {@code ReconnectDelay}. | ||
| * @return the instance. | ||
| * @throws NullPointerException if {@code duration} is {@code null}. | ||
| */ | ||
| static ReconnectDelay ofOrLowerBoundary(final Duration duration) { | ||
| final Duration durationWithinLowerBoundary; | ||
| if (0 < LOWER_BOUNDARY.compareTo(ConditionChecker.checkNotNull(duration, "duration"))) { | ||
| durationWithinLowerBoundary = LOWER_BOUNDARY; | ||
| } else { | ||
| durationWithinLowerBoundary = duration; | ||
| } | ||
| return new ReconnectDelay(durationWithinLowerBoundary); | ||
| } | ||
|
|
||
| public Duration getDuration() { | ||
| return duration; | ||
| } | ||
|
|
||
| @Override | ||
| public int compareTo(final ReconnectDelay o) { | ||
| ConditionChecker.checkNotNull(o, "o"); | ||
| return duration.compareTo(o.getDuration()); | ||
| } | ||
|
|
||
| @Override | ||
| public boolean equals(@Nullable final Object o) { | ||
| if (this == o) { | ||
| return true; | ||
| } | ||
| if (o == null || getClass() != o.getClass()) { | ||
| return false; | ||
| } | ||
| final var that = (ReconnectDelay) o; | ||
| return Objects.equals(duration, that.duration); | ||
| } | ||
|
|
||
| @Override | ||
| public int hashCode() { | ||
| return Objects.hash(duration); | ||
| } | ||
|
|
||
| /** | ||
| * @return the string representation of the wrapped {@code Duration}. | ||
| */ | ||
| @Override | ||
| public String toString() { | ||
| return duration.toString(); | ||
| } | ||
|
|
||
| } |
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
Oops, something went wrong.