Avoid double executor on non-persistent topic send operation
@@ -188,32 +188,24 @@ public class NonPersistentTopic implements Topic {
@Override
public void publishMessage(ByteBuf data, PublishContext callback) {
- AtomicInteger msgDeliveryCount = new AtomicInteger(2);
+ callback.completed(null, 0L, 0L);
ENTRIES_ADDED_COUNTER_UPDATER.incrementAndGet(this);
- // retain data for sub/replication because io-thread will release actual payload
- data.retain(2);
- this.executor.executeOrdered(topic, SafeRun.safeRun(() -> {
subscriptions.forEach((name, subscription) -> {
ByteBuf duplicateBuffer = data.retainedDuplicate();
Entry entry = create(0L, 0L, duplicateBuffer);
// entry internally retains data so, duplicateBuffer should be release here
duplicateBuffer.release();
if (subscription.getDispatcher() != null) {
- subscription.getDispatcher().sendMessages(Lists.newArrayList(entry));
+ subscription.getDispatcher().sendMessages(Collections.singletonList(entry));
} else {
// it happens when subscription is created but dispatcher is not created as consumer is not added
// yet
entry.release();
}
});
- data.release();
- if (msgDeliveryCount.decrementAndGet() == 0) {
- callback.completed(null, 0L, 0L);
- }
- }));
- this.executor.executeOrdered(topic, SafeRun.safeRun(() -> {
+ if (!replicators.isEmpty()) {
replicators.forEach((name, replicator) -> {
ByteBuf duplicateBuffer = data.retainedDuplicate();
Entry entry = create(0L, 0L, duplicateBuffer);
@@ -221,11 +213,7 @@ public class NonPersistentTopic implements Topic {
duplicateBuffer.release();
((NonPersistentReplicator) replicator).sendMessage(entry);
});
- data.release();
- if (msgDeliveryCount.decrementAndGet() == 0) {
- callback.completed(null, 0L, 0L);
}
- }));
}
@Override
