Merge pull request #286 from hivemq/develop

Release 1.1.0

Author: SgtSilvio

CODE_OF_CONDUCT.md

@@ -0,0 +1,3 @@
+# HiveMQ Code of Conduct
+
+Please refer to our [HiveMQ Code of Conduct](https://github.com/hivemq/hivemq-community/blob/master/code-of-conduct.md).

CONTRIBUTING.md

@@ -1,18 +1,24 @@
 # Contributing
+## Contributing to the HiveMQ Community Projects
 
-## External contributors
+Welcome to the HiveMQ Community! Glad to see your interest in contributing to HiveMQ MQTT Client. Please checkout our [Contribution Guide](https://github.com/hivemq/hivemq-community/blob/master/CONTRIBUTING.adoc) to make sure your contribution will be accepted by the HiveMQ team.
+
+For information on how the HiveMQ Community is organized and how contributions will be accepted please have a look at our [HiveMQ Community Repo](https://github.com/hivemq/hivemq-community). 
+
+##Contributing to HiveMQ MQTT Client
+### External contributors
 If you would like to contribute code, do the following:
 - Fork the repository on GitHub
 - Open a pull request targeting the `develop` branch
 
-## License
+### License
 By contributing your code, you agree to license your contribution under the terms of the
 [Apache License, Version 2.0](https://github.com/hivemq/hivemq-mqtt-client/blob/develop/LICENSE).
 
 All files must contain the license header from the
 [HEADER](https://github.com/hivemq/hivemq-mqtt-client/blob/develop/HEADER) file.
 
-## Branching model
+### Branching model
 
 - `master`: release branch, protected
   - `develop` is merged into `master` by creating a merge commit if a new version is released
@@ -25,22 +31,22 @@ All files must contain the license header from the
   - Pull request targeting the `develop` branch
   - Mandatory code review of the pull request
 
-## Branching guidelines
+### Branching guidelines
 
 - Branch types: feature, bugfix, improvement, cleanup (same as the label of a corresponding GitHub Issue)
 - Branch names:
   - Starting with type: `feature/`, `bugfix/`, `improvement/`, `cleanup/`
   - \+ task: lower case, spaces replaced with `-`
 
-## Commit guidelines
+### Commit guidelines
 
 - Commits should be as atomic as possible.
 - Commit messages should describe the changes clearly.
 - Commit messages should start with a capital letter for consistency.
 - Commit messages should avoid exceeding the line length limit. Instead use multiple lines, each describing one specific
 change.
 
-## Code style guidelines
+### Code style guidelines
 
 - The project uses Nullability annotations to avoid NullPointerExceptions: `@NotNull`, `@Nullable`.
 Every non-primitive parameter/return type/field should be annotated with one of them.

README.md

@@ -49,7 +49,7 @@ If you use Gradle, just include the following inside your `build.gradle` file.
 
 ```groovy
 dependencies {
-    compile group: 'com.hivemq', name: 'hivemq-mqtt-client', version: '1.0.0'
+    compile group: 'com.hivemq', name: 'hivemq-mqtt-client', version: '1.0.1'
 }
 ```
 
@@ -71,7 +71,7 @@ NOTE: You have to set the compiler version to `1.8` or higher.
         <dependency>
             <groupId>com.hivemq</groupId>
             <artifactId>hivemq-mqtt-client</artifactId>
-            <version>1.0.0</version>
+            <version>1.0.1</version>
         </dependency>
     </dependencies>
     ...
@@ -88,7 +88,7 @@ To use the shaded version just append `-shaded` to the artifact name.
 
 ```groovy
 dependencies {
-    compile group: 'com.hivemq', name: 'hivemq-mqtt-client-shaded', version: '1.0.0'
+    compile group: 'com.hivemq', name: 'hivemq-mqtt-client-shaded', version: '1.0.1'
 }
 ```
 
@@ -101,7 +101,7 @@ dependencies {
         <dependency>
             <groupId>com.hivemq</groupId>
             <artifactId>hivemq-mqtt-client-shaded</artifactId>
-            <version>1.0.0</version>
+            <version>1.0.1</version>
         </dependency>
     </dependencies>
     ...
@@ -125,7 +125,7 @@ repositories {
 }
 
 dependencies {
-    compile group: 'com.hivemq', name: 'hivemq-mqtt-client', version: '1.0.0-SNAPSHOT'
+    compile group: 'com.hivemq', name: 'hivemq-mqtt-client', version: '1.0.1-SNAPSHOT'
 }
 ```
 
@@ -145,7 +145,7 @@ dependencies {
         <dependency>
             <groupId>com.hivemq</groupId>
             <artifactId>hivemq-mqtt-client</artifactId>
-            <version>1.0.0-SNAPSHOT</version>
+            <version>1.0.1-SNAPSHOT</version>
         </dependency>
     </dependencies>
     ...
@@ -590,3 +590,7 @@ part of the public API.
 # Contributing
 
 See [CONTRIBUTING.md](CONTRIBUTING.md)
+
+# License
+
+See [LICENSE](LICENSE)

build.gradle

@@ -13,7 +13,7 @@ plugins {
 }
 
 group 'com.hivemq'
-version '1.0.1' + (Boolean.valueOf(System.getProperty("snapshot")) ? "-SNAPSHOT" : "")
+version '1.1.0' + (Boolean.valueOf(System.getProperty("snapshot")) ? "-SNAPSHOT" : "")
 description 'HiveMQ MQTT Client is a MQTT 5.0 and MQTT 3.1.1 compatible and feature-rich high-performance Java client ' +
         'library with different API flavours and backpressure support'
 
@@ -28,6 +28,7 @@ ext {
     licenseReadableName = 'The Apache License, Version 2.0'
     licenseUrl = 'http://www.apache.org/licenses/LICENSE-2.0.txt'
     shadedAppendix = 'shaded'
+    prevVersion = '1.0.1'
 }
 
 sourceCompatibility = 1.8

japicc.gradle

@@ -8,7 +8,7 @@ task japicc {
     def workingDir = new File(project.buildDir, 'japicc')
     def executable = new File(workingDir, 'japi-compliance-checker-' + japiccVersion + '/japi-compliance-checker.pl')
 
-    def lastSemVer = lastSemVer()
+    def lastSemVer = project.prevVersion == null ? lastSemVer() : project.prevVersion
     def shadedName = project.name + '-' + shadedAppendix
     def lastJar = new File(workingDir, project.name + '-' + lastSemVer + '.jar')
     def lastShadedJar = new File(workingDir, shadedName + '-' + lastSemVer + '.jar')

src/main/java/com/hivemq/client/internal/mqtt/MqttAsyncClient.java

@@ -17,7 +17,13 @@
 
 package com.hivemq.client.internal.mqtt;
 
+import com.hivemq.client.internal.mqtt.message.connect.MqttConnect;
+import com.hivemq.client.internal.mqtt.message.disconnect.MqttDisconnect;
+import com.hivemq.client.internal.mqtt.message.publish.MqttPublish;
+import com.hivemq.client.internal.mqtt.message.subscribe.MqttSubscribe;
 import com.hivemq.client.internal.mqtt.message.subscribe.MqttSubscribeBuilder;
+import com.hivemq.client.internal.mqtt.message.unsubscribe.MqttUnsubscribe;
+import com.hivemq.client.internal.mqtt.util.MqttChecks;
 import com.hivemq.client.internal.rx.RxFutureConverter;
 import com.hivemq.client.internal.util.Checks;
 import com.hivemq.client.mqtt.MqttGlobalPublishFilter;
@@ -31,7 +37,6 @@
 import com.hivemq.client.mqtt.mqtt5.message.subscribe.suback.Mqtt5SubAck;
 import com.hivemq.client.mqtt.mqtt5.message.unsubscribe.Mqtt5Unsubscribe;
 import com.hivemq.client.mqtt.mqtt5.message.unsubscribe.unsuback.Mqtt5UnsubAck;
-import io.reactivex.Flowable;
 import io.reactivex.FlowableSubscriber;
 import io.reactivex.schedulers.Schedulers;
 import org.jetbrains.annotations.NotNull;
@@ -51,8 +56,6 @@ public class MqttAsyncClient implements Mqtt5AsyncClient {
     private static final @NotNull Function<Mqtt5SubAck, Mqtt5SubAck> SUBACK_HANDLER = MqttBlockingClient::handleSubAck;
     private static final @NotNull Function<Mqtt5UnsubAck, Mqtt5UnsubAck> UNSUBACK_HANDLER =
             MqttBlockingClient::handleUnsubAck;
-    private static final @NotNull Function<Mqtt5PublishResult, Mqtt5PublishResult> PUBLISH_HANDLER =
-            MqttBlockingClient::handlePublish;
 
     private final @NotNull MqttRxClient delegate;
 
@@ -62,21 +65,26 @@ public class MqttAsyncClient implements Mqtt5AsyncClient {
 
     @Override
     public @NotNull CompletableFuture<@NotNull Mqtt5ConnAck> connect(final @Nullable Mqtt5Connect connect) {
-        return RxFutureConverter.toFuture(delegate.connect(connect));
+        final MqttConnect mqttConnect = MqttChecks.connect(connect);
+
+        return RxFutureConverter.toFuture(delegate.connect(mqttConnect));
     }
 
     @Override
     public @NotNull CompletableFuture<@NotNull Mqtt5SubAck> subscribe(final @Nullable Mqtt5Subscribe subscribe) {
-        return RxFutureConverter.toFuture(delegate.subscribe(subscribe));
+        final MqttSubscribe mqttSubscribe = MqttChecks.subscribe(subscribe);
+
+        return RxFutureConverter.toFuture(delegate.subscribe(mqttSubscribe)).thenApply(SUBACK_HANDLER);
     }
 
     @Override
     public @NotNull CompletableFuture<@NotNull Mqtt5SubAck> subscribe(
             final @Nullable Mqtt5Subscribe subscribe, final @Nullable Consumer<@NotNull Mqtt5Publish> callback) {
 
+        final MqttSubscribe mqttSubscribe = MqttChecks.subscribe(subscribe);
         Checks.notNull(callback, "Callback");
 
-        return delegate.subscribeStream(subscribe)
+        return delegate.subscribeStream(mqttSubscribe)
                 .subscribeSingleFuture(new CallbackSubscriber(callback))
                 .thenApply(SUBACK_HANDLER);
     }
@@ -86,10 +94,11 @@ public class MqttAsyncClient implements Mqtt5AsyncClient {
             final @Nullable Mqtt5Subscribe subscribe, final @Nullable Consumer<@NotNull Mqtt5Publish> callback,
             final @Nullable Executor executor) {
 
+        final MqttSubscribe mqttSubscribe = MqttChecks.subscribe(subscribe);
         Checks.notNull(callback, "Callback");
         Checks.notNull(executor, "Executor");
 
-        return delegate.subscribeStreamUnsafe(subscribe)
+        return delegate.subscribeStreamUnsafe(mqttSubscribe)
                 .observeOnBoth(Schedulers.from(executor), true)
                 .subscribeSingleFuture(new CallbackSubscriber(callback))
                 .thenApply(SUBACK_HANDLER);
@@ -99,6 +108,7 @@ public class MqttAsyncClient implements Mqtt5AsyncClient {
     public void publishes(
             final @Nullable MqttGlobalPublishFilter filter, final @Nullable Consumer<@NotNull Mqtt5Publish> callback) {
 
+        Checks.notNull(filter, "Global publish filter");
         Checks.notNull(callback, "Callback");
 
         delegate.publishes(filter).subscribe(new CallbackSubscriber(callback));
@@ -109,6 +119,7 @@ public void publishes(
             final @Nullable MqttGlobalPublishFilter filter, final @Nullable Consumer<@NotNull Mqtt5Publish> callback,
             final @Nullable Executor executor) {
 
+        Checks.notNull(filter, "Global publish filter");
         Checks.notNull(callback, "Callback");
         Checks.notNull(executor, "Executor");
 
@@ -120,15 +131,17 @@ public void publishes(
     @Override
     public @NotNull CompletableFuture<@NotNull Mqtt5UnsubAck> unsubscribe(
             final @Nullable Mqtt5Unsubscribe unsubscribe) {
-        return RxFutureConverter.toFuture(delegate.unsubscribe(unsubscribe)).thenApply(UNSUBACK_HANDLER);
+
+        final MqttUnsubscribe mqttUnsubscribe = MqttChecks.unsubscribe(unsubscribe);
+
+        return RxFutureConverter.toFuture(delegate.unsubscribe(mqttUnsubscribe)).thenApply(UNSUBACK_HANDLER);
     }
 
     @Override
     public @NotNull CompletableFuture<@NotNull Mqtt5PublishResult> publish(final @Nullable Mqtt5Publish publish) {
-        Checks.notNull(publish, "Publish");
+        final MqttPublish mqttPublish = MqttChecks.publish(publish);
 
-        return RxFutureConverter.toFuture(delegate.publishHalfSafe(Flowable.just(publish)).singleOrError())
-                .thenApply(PUBLISH_HANDLER);
+        return RxFutureConverter.toFuture(delegate.publish(mqttPublish));
     }
 
     @Override
@@ -138,7 +151,9 @@ public void publishes(
 
     @Override
     public @NotNull CompletableFuture<Void> disconnect(final @Nullable Mqtt5Disconnect disconnect) {
-        return RxFutureConverter.toFuture(delegate.disconnect(disconnect));
+        final MqttDisconnect mqttDisconnect = MqttChecks.disconnect(disconnect);
+
+        return RxFutureConverter.toFuture(delegate.disconnect(mqttDisconnect));
     }
 
     @Override

src/main/java/com/hivemq/client/internal/mqtt/MqttBlockingClient.java

@@ -17,6 +17,12 @@
 
 package com.hivemq.client.internal.mqtt;
 
+import com.hivemq.client.internal.mqtt.message.connect.MqttConnect;
+import com.hivemq.client.internal.mqtt.message.disconnect.MqttDisconnect;
+import com.hivemq.client.internal.mqtt.message.publish.MqttPublish;
+import com.hivemq.client.internal.mqtt.message.subscribe.MqttSubscribe;
+import com.hivemq.client.internal.mqtt.message.unsubscribe.MqttUnsubscribe;
+import com.hivemq.client.internal.mqtt.util.MqttChecks;
 import com.hivemq.client.internal.util.AsyncRuntimeException;
 import com.hivemq.client.internal.util.Checks;
 import com.hivemq.client.mqtt.MqttGlobalPublishFilter;
@@ -70,18 +76,6 @@ public class MqttBlockingClient implements Mqtt5BlockingClient {
         return unsubAck;
     }
 
-    static @NotNull Mqtt5PublishResult handlePublish(final @NotNull Mqtt5PublishResult publishResult) {
-        final Optional<Throwable> error = publishResult.getError();
-        if (error.isPresent()) {
-            final Throwable throwable = error.get();
-            if (throwable instanceof RuntimeException) {
-                throw (RuntimeException) throwable;
-            }
-            throw new RuntimeException(throwable);
-        }
-        return publishResult;
-    }
-
     private final @NotNull MqttRxClient delegate;
 
     MqttBlockingClient(final @NotNull MqttRxClient delegate) {
@@ -90,41 +84,46 @@ public class MqttBlockingClient implements Mqtt5BlockingClient {
 
     @Override
     public @NotNull Mqtt5ConnAck connect(final @Nullable Mqtt5Connect connect) {
+        final MqttConnect mqttConnect = MqttChecks.connect(connect);
         try {
-            return delegate.connectUnsafe(connect).blockingGet();
+            return delegate.connectUnsafe(mqttConnect).blockingGet();
         } catch (final RuntimeException e) {
             throw AsyncRuntimeException.fillInStackTrace(e);
         }
     }
 
     @Override
     public @NotNull Mqtt5SubAck subscribe(final @Nullable Mqtt5Subscribe subscribe) {
+        final MqttSubscribe mqttSubscribe = MqttChecks.subscribe(subscribe);
         try {
-            return handleSubAck(delegate.subscribeUnsafe(subscribe).blockingGet());
+            return handleSubAck(delegate.subscribeUnsafe(mqttSubscribe).blockingGet());
         } catch (final RuntimeException e) {
             throw AsyncRuntimeException.fillInStackTrace(e);
         }
     }
 
     @Override
     public @NotNull Mqtt5Publishes publishes(final @Nullable MqttGlobalPublishFilter filter) {
+        Checks.notNull(filter, "Global publish filter");
+
         return new MqttPublishes(delegate.publishesUnsafe(filter));
     }
 
     @Override
     public @NotNull Mqtt5UnsubAck unsubscribe(final @Nullable Mqtt5Unsubscribe unsubscribe) {
+        final MqttUnsubscribe mqttUnsubscribe = MqttChecks.unsubscribe(unsubscribe);
         try {
-            return handleUnsubAck(delegate.unsubscribeUnsafe(unsubscribe).blockingGet());
+            return handleUnsubAck(delegate.unsubscribeUnsafe(mqttUnsubscribe).blockingGet());
         } catch (final RuntimeException e) {
             throw AsyncRuntimeException.fillInStackTrace(e);
         }
     }
 
     @Override
     public @NotNull Mqtt5PublishResult publish(final @Nullable Mqtt5Publish publish) {
-        Checks.notNull(publish, "Publish");
+        final MqttPublish mqttPublish = MqttChecks.publish(publish);
         try {
-            return handlePublish(delegate.publishUnsafe(Flowable.just(publish)).singleOrError().blockingGet());
+            return delegate.publishUnsafe(mqttPublish).blockingGet();
         } catch (final RuntimeException e) {
             throw AsyncRuntimeException.fillInStackTrace(e);
         }
@@ -141,8 +140,9 @@ public void reauth() {
 
     @Override
     public void disconnect(final @NotNull Mqtt5Disconnect disconnect) {
+        final MqttDisconnect mqttDisconnect = MqttChecks.disconnect(disconnect);
         try {
-            delegate.disconnectUnsafe(disconnect).blockingAwait();
+            delegate.disconnectUnsafe(mqttDisconnect).blockingAwait();
         } catch (final RuntimeException e) {
             throw AsyncRuntimeException.fillInStackTrace(e);
         }