Skip to content

Introduce synchronous consumer #179

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
wants to merge 2 commits into
base: main
Choose a base branch
from
Draft

Introduce synchronous consumer #179

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
5658fc0
Introduce synchronous consumer
acogoluegnes Mar 27, 2025
e4da3f8
Remove unnecessary test class
acogoluegnes Mar 27, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 42 additions & 0 deletions src/main/java/com/rabbitmq/client/amqp/SynchronousConsumer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,42 @@
// Copyright (c) 2025 Broadcom. All Rights Reserved.
// The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// If you have any questions regarding licensing, please contact us at
// [email protected].
package com.rabbitmq.client.amqp;

import java.time.Duration;
import java.util.List;

public interface SynchronousConsumer extends AutoCloseable, Resource {

Response get(Duration timeout);

List<Response> get(int messageCount, Duration timeout);

// CompletableFuture<Response> getAsync(Duration timeout);
//
// CompletableFuture<List<Response>> getAsync(int messageCount, Duration timeout);

@Override
void close();

interface Response {

Consumer.Context context();

Message message();
}
}
211 changes: 37 additions & 174 deletions src/main/java/com/rabbitmq/client/amqp/impl/AmqpConsumer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,6 @@
import static com.rabbitmq.client.amqp.Resource.State.CLOSING;
import static com.rabbitmq.client.amqp.Resource.State.OPEN;
import static com.rabbitmq.client.amqp.impl.AmqpConsumerBuilder.*;
import static com.rabbitmq.client.amqp.metrics.MetricsCollector.ConsumeDisposition.*;
import static java.time.Duration.ofSeconds;
import static java.util.Optional.ofNullable;

Expand All @@ -38,7 +37,6 @@
import java.util.stream.IntStream;
import org.apache.qpid.protonj2.client.*;
import org.apache.qpid.protonj2.client.exceptions.*;
import org.apache.qpid.protonj2.client.impl.ClientConversionSupport;
import org.apache.qpid.protonj2.client.impl.ClientReceiver;
import org.apache.qpid.protonj2.client.util.DeliveryQueue;
import org.apache.qpid.protonj2.engine.EventHandler;
Expand All @@ -47,14 +45,10 @@
import org.apache.qpid.protonj2.engine.impl.ProtonReceiver;
import org.apache.qpid.protonj2.engine.impl.ProtonSessionIncomingWindow;
import org.apache.qpid.protonj2.types.DescribedType;
import org.apache.qpid.protonj2.types.messaging.Accepted;
import org.apache.qpid.protonj2.types.messaging.Modified;
import org.apache.qpid.protonj2.types.messaging.Rejected;
import org.apache.qpid.protonj2.types.messaging.Released;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

final class AmqpConsumer extends ResourceBase implements Consumer {
final class AmqpConsumer extends ResourceBase implements Consumer, ConsumerUtils.CloseableConsumer {

private static final AtomicLong ID_SEQUENCE = new AtomicLong(0);

Expand Down Expand Up @@ -114,7 +108,7 @@ final class AmqpConsumer extends ResourceBase implements Consumer {
.dispatch(
() -> {
// get result to make spotbugs happy
boolean ignored = maybeCloseConsumerOnException(this, e);
boolean ignored = ConsumerUtils.maybeCloseConsumerOnException(this, e);
});
};
this.consumerWorkService = connection.consumerWorkService();
Expand Down Expand Up @@ -301,7 +295,7 @@ void recoverAfterConnectionFailure() {
}
}

void close(Throwable cause) {
public void close(Throwable cause) {
if (this.closed.compareAndSet(false, true)) {
this.state(CLOSING, cause);
if (this.consumerWorkService != null) {
Expand Down Expand Up @@ -362,7 +356,8 @@ private void initStateFromNativeReceiver(ClientReceiver receiver) {
throw new AmqpException("Could not initialize consumer internal state");
}
} catch (InterruptedException e) {
throw new RuntimeException(e);
Thread.currentThread().interrupt();
throw new AmqpException(e);
}
}

Expand Down Expand Up @@ -400,17 +395,13 @@ enum PauseStatus {
PAUSED
}

private static class DeliveryContext implements Consumer.Context {
private static class DeliveryContext extends ConsumerUtils.DeliveryContextBase {

private static final DeliveryState REJECTED = DeliveryState.rejected(null, null);
private final AtomicBoolean settled = new AtomicBoolean(false);
private final Delivery delivery;
private final Scheduler protonExecutor;
private final ProtonReceiver protonReceiver;
private final MetricsCollector metricsCollector;
private final AtomicLong unsettledMessageCount;
private final Runnable replenishCreditOperation;
private final AmqpConsumer consumer;

private DeliveryContext(
Delivery delivery,
Expand All @@ -420,48 +411,17 @@ private DeliveryContext(
AtomicLong unsettledMessageCount,
Runnable replenishCreditOperation,
AmqpConsumer consumer) {
this.delivery = delivery;
super(delivery, consumer);
this.protonExecutor = protonExecutor;
this.protonReceiver = protonReceiver;
this.metricsCollector = metricsCollector;
this.unsettledMessageCount = unsettledMessageCount;
this.replenishCreditOperation = replenishCreditOperation;
this.consumer = consumer;
}

@Override
public void accept() {
this.settle(DeliveryState.accepted(), ACCEPTED, "accept");
}

@Override
public void discard() {
settle(REJECTED, DISCARDED, "discard");
}

@Override
public void discard(Map<String, Object> annotations) {
annotations = annotations == null ? Collections.emptyMap() : annotations;
Utils.checkMessageAnnotations(annotations);
this.settle(DeliveryState.modified(true, true, annotations), DISCARDED, "discard (modified)");
}

@Override
public void requeue() {
settle(DeliveryState.released(), REQUEUED, "requeue");
}

@Override
public void requeue(Map<String, Object> annotations) {
annotations = annotations == null ? Collections.emptyMap() : annotations;
Utils.checkMessageAnnotations(annotations);
this.settle(
DeliveryState.modified(false, false, annotations), REQUEUED, "requeue (modified)");
}

@Override
public BatchContext batch(int batchSizeHint) {
return new BatchDeliveryContext(
public Consumer.BatchContext batch(int batchSizeHint) {
return new BatchContext(
batchSizeHint,
protonExecutor,
protonReceiver,
Expand All @@ -471,18 +431,13 @@ public BatchContext batch(int batchSizeHint) {
consumer);
}

private void settle(
DeliveryState state, MetricsCollector.ConsumeDisposition disposition, String label) {
if (settled.compareAndSet(false, true)) {
try {
protonExecutor.execute(replenishCreditOperation);
delivery.disposition(state, true);
unsettledMessageCount.decrementAndGet();
metricsCollector.consumeDisposition(disposition);
} catch (Exception e) {
handleContextException(this.consumer, e, label);
}
}
@Override
protected void doSettle(DeliveryState state, MetricsCollector.ConsumeDisposition disposition)
throws Exception {
protonExecutor.execute(replenishCreditOperation);
delivery.disposition(state, true);
unsettledMessageCount.decrementAndGet();
metricsCollector.consumeDisposition(disposition);
}
}

Expand All @@ -491,142 +446,50 @@ public String toString() {
return "AmqpConsumer{" + "id=" + id + ", queue='" + queue + '\'' + '}';
}

private static final class BatchDeliveryContext implements BatchContext {
private static final class BatchContext extends ConsumerUtils.BatchContextBase {

private static final org.apache.qpid.protonj2.types.transport.DeliveryState REJECTED =
new Rejected();
private final List<DeliveryContext> contexts;
private final AtomicBoolean settled = new AtomicBoolean(false);
private final Scheduler protonExecutor;
private final ProtonReceiver protonReceiver;
private final MetricsCollector metricsCollector;
private final AtomicLong unsettledMessageCount;
private final Runnable replenishCreditOperation;
private final AmqpConsumer consumer;

private BatchDeliveryContext(
private BatchContext(
int batchSizeHint,
Scheduler protonExecutor,
ProtonReceiver protonReceiver,
MetricsCollector metricsCollector,
AtomicLong unsettledMessageCount,
Runnable replenishCreditOperation,
AmqpConsumer consumer) {
this.contexts = new ArrayList<>(batchSizeHint);
ConsumerUtils.CloseableConsumer consumer) {
super(batchSizeHint, consumer);
this.protonExecutor = protonExecutor;
this.protonReceiver = protonReceiver;
this.metricsCollector = metricsCollector;
this.unsettledMessageCount = unsettledMessageCount;
this.replenishCreditOperation = replenishCreditOperation;
this.consumer = consumer;
}

@Override
public void add(Consumer.Context context) {
if (this.settled.get()) {
throw new IllegalStateException("Batch is closed");
} else {
if (context instanceof DeliveryContext) {
DeliveryContext dctx = (DeliveryContext) context;
// marking the context as settled avoids operation on it and deduplicates as well
if (dctx.settled.compareAndSet(false, true)) {
this.contexts.add(dctx);
} else {
throw new IllegalStateException("Message already settled");
}
} else {
throw new IllegalArgumentException("Context type not supported: " + context);
}
}
}

@Override
public int size() {
return this.contexts.size();
}

@Override
public void accept() {
this.settle(Accepted.getInstance(), ACCEPTED, "accept");
}

@Override
public void discard() {
this.settle(REJECTED, DISCARDED, "discard");
}

@Override
public void discard(Map<String, Object> annotations) {
annotations = annotations == null ? Collections.emptyMap() : annotations;
Utils.checkMessageAnnotations(annotations);
Modified state =
new Modified(false, true, ClientConversionSupport.toSymbolKeyedMap(annotations));
this.settle(state, DISCARDED, "discard (modified)");
}

@Override
public void requeue() {
this.settle(Released.getInstance(), REQUEUED, "requeue");
}

@Override
public void requeue(Map<String, Object> annotations) {
annotations = annotations == null ? Collections.emptyMap() : annotations;
Utils.checkMessageAnnotations(annotations);
Modified state =
new Modified(false, false, ClientConversionSupport.toSymbolKeyedMap(annotations));
this.settle(state, REQUEUED, "requeue (modified)");
}

@Override
public BatchContext batch(int batchSizeHint) {
return this;
}

private void settle(
protected void doSettle(
org.apache.qpid.protonj2.types.transport.DeliveryState state,
MetricsCollector.ConsumeDisposition disposition,
String label) {
if (settled.compareAndSet(false, true)) {
int batchSize = this.contexts.size();
try {
protonExecutor.execute(replenishCreditOperation);
long[][] ranges =
SerialNumberUtils.ranges(this.contexts, ctx -> ctx.delivery.getDeliveryId());
this.protonExecutor.execute(
() -> {
for (long[] range : ranges) {
this.protonReceiver.disposition(state, range);
}
MetricsCollector.ConsumeDisposition disposition) {
int batchSize = this.size();
protonExecutor.execute(replenishCreditOperation);
long[][] ranges =
SerialNumberUtils.ranges(this.contexts(), ctx -> ctx.delivery.getDeliveryId());
this.protonExecutor.execute(
() -> {
for (long[] range : ranges) {
this.protonReceiver.disposition(state, range);
}
});
unsettledMessageCount.addAndGet(-batchSize);
IntStream.range(0, batchSize)
.forEach(
ignored -> {
metricsCollector.consumeDisposition(disposition);
});
unsettledMessageCount.addAndGet(-batchSize);
IntStream.range(0, batchSize)
.forEach(
ignored -> {
metricsCollector.consumeDisposition(disposition);
});
} catch (Exception e) {
handleContextException(this.consumer, e, label);
}
}
}
}

private static void handleContextException(
AmqpConsumer consumer, Exception ex, String operation) {
if (maybeCloseConsumerOnException(consumer, ex)) {
return;
}
if (ex instanceof ClientIllegalStateException
|| ex instanceof RejectedExecutionException
|| ex instanceof ClientIOException) {
LOGGER.debug("message {} failed: {}", operation, ex.getMessage());
} else if (ex instanceof ClientException) {
throw ExceptionUtils.convert((ClientException) ex);
}
}

private static boolean maybeCloseConsumerOnException(AmqpConsumer consumer, Exception ex) {
return ExceptionUtils.maybeCloseOnException(consumer::close, ex);
}
}
Loading