Simplified MqttPublishFlowableAckLink linking
@@ -70,7 +70,6 @@ public class MqttPublishFlowableAckLink extends Flowable<MqttPublishWithFlow> {
private final @NotNull Subscriber<? super MqttPublishWithFlow> subscriber;
private final @NotNull MqttAckFlowableFlow ackFlow;
private @Nullable Subscription subscription;
- private boolean linked;
private final @NotNull AtomicInteger state = new AtomicInteger();
private long published;
@@ -86,7 +85,7 @@ public class MqttPublishFlowableAckLink extends Flowable<MqttPublishWithFlow> {
public void onSubscribe(final @NotNull Subscription subscription) {
this.subscription = subscription;
subscriber.onSubscribe(this);
- link();
+ ackFlow.link(this);
}
@Override
@@ -121,7 +120,6 @@ public class MqttPublishFlowableAckLink extends Flowable<MqttPublishWithFlow> {
@Override
public void request(final long n) {
assert subscription != null;
- link();
subscription.request(n);
}
@@ -132,13 +130,6 @@ public class MqttPublishFlowableAckLink extends Flowable<MqttPublishWithFlow> {
subscription.cancel();
}
- private void link() {
- if (!linked) {
- linked = true;
- ackFlow.link(this);
- }
- }
-
@Override
public void cancelLink() {
if (state.getAndSet(STATE_CANCEL) == STATE_NONE) {
