Improved MqttBlockingClient
@@ -176,19 +176,9 @@ public class MqttBlockingClient implements Mqtt5BlockingClient {
return;
}
}
- }
- synchronized (this) {
- final boolean retry;
- synchronized (entries) {
- retry = !entries.isEmpty();
- }
- if (retry) {
- onNext(publish);
- } else {
queuedPublish = publish;
}
}
- }
@Override
public void onComplete() {
@@ -212,12 +202,16 @@ public class MqttBlockingClient implements Mqtt5BlockingClient {
@Override
public @NotNull Mqtt5Publish receive() throws InterruptedException {
final Entry entry;
- synchronized (this) {
+ synchronized (entries) {
+ if (cancelled) {
+ throw new CancellationException();
+ }
final Mqtt5Publish publish = receiveNowUnsafe();
if (publish != null) {
return publish;
}
- entry = addEntry();
+ entry = new Entry();
+ entries.offer(entry);
}
InterruptedException interruptedException = null;
@@ -252,12 +246,16 @@ public class MqttBlockingClient implements Mqtt5BlockingClient {
Objects.requireNonNull(timeUnit, "Time unit must not be null.");
final Entry entry;
- synchronized (this) {
+ synchronized (entries) {
+ if (cancelled) {
+ throw new CancellationException();
+ }
final Mqtt5Publish publish = receiveNowUnsafe();
if (publish != null) {
return Optional.of(publish);
}
- entry = addEntry();
+ entry = new Entry();
+ entries.offer(entry);
}
InterruptedException interruptedException = null;
@@ -285,7 +283,10 @@ public class MqttBlockingClient implements Mqtt5BlockingClient {
@Override
public @NotNull Optional<Mqtt5Publish> receiveNow() {
final Mqtt5Publish publish;
- synchronized (this) {
+ synchronized (entries) {
+ if (cancelled) {
+ throw new CancellationException();
+ }
publish = receiveNowUnsafe();
}
return Optional.ofNullable(publish);
@@ -300,17 +301,6 @@ public class MqttBlockingClient implements Mqtt5BlockingClient {
return null;
}
- private @NotNull Entry addEntry() {
- synchronized (entries) {
- if (cancelled) {
- throw new CancellationException();
- }
- final Entry entry = new Entry();
- entries.offer(entry);
- return entry;
- }
- }
-
@Override
public void close() {
final Subscription subscription = this.subscription.getAndSet(SubscriptionHelper.CANCELLED);
