Refactor RabbitMQ access API and package visibility

framework/src/main/java/com/github/dbmdz/flusswerk/framework/config/FlusswerkConfiguration.java

@@ -1,6 +1,8 @@
 package com.github.dbmdz.flusswerk.framework.config;
 
-import com.github.dbmdz.flusswerk.framework.config.properties.*;
+import com.github.dbmdz.flusswerk.framework.config.properties.AppProperties;
+import com.github.dbmdz.flusswerk.framework.config.properties.ProcessingProperties;
+import com.github.dbmdz.flusswerk.framework.config.properties.RoutingProperties;
 import com.github.dbmdz.flusswerk.framework.engine.Engine;
 import com.github.dbmdz.flusswerk.framework.engine.FlusswerkConsumer;
 import com.github.dbmdz.flusswerk.framework.engine.Task;
@@ -12,15 +14,11 @@
 import com.github.dbmdz.flusswerk.framework.monitoring.FlowMetrics;
 import com.github.dbmdz.flusswerk.framework.monitoring.FlusswerkMetrics;
 import com.github.dbmdz.flusswerk.framework.monitoring.MeterFactory;
-import com.github.dbmdz.flusswerk.framework.rabbitmq.MessageBroker;
-import com.github.dbmdz.flusswerk.framework.rabbitmq.RabbitClient;
 import com.github.dbmdz.flusswerk.framework.rabbitmq.RabbitConnection;
 import com.github.dbmdz.flusswerk.framework.rabbitmq.RabbitMQ;
 import com.github.dbmdz.flusswerk.framework.reporting.DefaultProcessReport;
 import com.github.dbmdz.flusswerk.framework.reporting.ProcessReport;
 import com.github.dbmdz.flusswerk.framework.reporting.Tracing;
-import io.micrometer.core.instrument.MeterRegistry;
-import java.io.IOException;
 import java.util.*;
 import java.util.concurrent.PriorityBlockingQueue;
 import java.util.concurrent.Semaphore;
@@ -31,21 +29,15 @@
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
 
-/** Spring configuration to provide beans for{@link MessageBroker} and {@link Engine}. */
 @Configuration
 @Import(FlusswerkPropertiesConfiguration.class)
 public class FlusswerkConfiguration {
 
-  @Bean
-  public Tracing tracing() {
-    return new Tracing();
-  }
-
   @Bean
   public Flow flow(Optional<FlowSpec> flowSpec) {
     // No FlowSpec → no Flow. We will have to handle this case when creating the
     // Engine bean as the sole consumer of the Flow bean.
-    return flowSpec.map(spec -> new Flow(spec)).orElse(null);
+    return flowSpec.map(Flow::new).orElse(null);
   }
 
   @Bean
@@ -67,45 +59,12 @@ public Engine engine(
     return new Engine(rabbitConnection.getChannel(), flusswerkConsumers, workers);
   }
 
-  @Bean
-  public MeterFactory meterFactory(
-      MonitoringProperties monitoringProperties, MeterRegistry meterRegistry) {
-    return new MeterFactory(monitoringProperties.prefix(), meterRegistry);
-  }
-
   @Bean
   public FlusswerkObjectMapper flusswerkObjectMapper(
       ObjectProvider<IncomingMessageType> incomingMessageType) {
     return new FlusswerkObjectMapper(incomingMessageType.getIfAvailable(IncomingMessageType::new));
   }
 
-  @Bean
-  public RabbitConnection rabbitConnection(
-      AppProperties appProperties, RabbitMQProperties rabbitMQProperties) throws IOException {
-    return new RabbitConnection(rabbitMQProperties, appProperties.name());
-  }
-
-  @Bean
-  public RabbitClient rabbitClient(
-      FlusswerkObjectMapper flusswerkObjectMapper, RabbitConnection rabbitConnection) {
-    return new RabbitClient(flusswerkObjectMapper, rabbitConnection);
-  }
-
-  @Bean
-  public RabbitMQ rabbitMQ(
-      RoutingProperties routingProperties,
-      RabbitClient rabbitClient,
-      MessageBroker messageBroker,
-      Tracing tracing) {
-    return new RabbitMQ(routingProperties, rabbitClient, messageBroker, tracing);
-  }
-
-  @Bean
-  public MessageBroker messageBroker(RoutingProperties routingProperties, RabbitClient rabbitClient)
-      throws IOException {
-    return new MessageBroker(routingProperties, rabbitClient);
-  }
-
   @Bean
   public PriorityBlockingQueue<Task> taskQueue() {
     return new PriorityBlockingQueue<>();
@@ -115,10 +74,10 @@ public PriorityBlockingQueue<Task> taskQueue() {
   public List<Worker> workers(
       AppProperties appProperties,
       Optional<Flow> flow,
-      MessageBroker messageBroker,
       ProcessingProperties processingProperties,
       Optional<ProcessReport> processReport,
       PriorityBlockingQueue<Task> taskQueue,
+      RabbitMQ rabbitMQ,
       Tracing tracing,
       FlusswerkMetrics metrics) {
     return flow.map(
@@ -129,22 +88,16 @@ public List<Worker> workers(
                             new Worker(
                                 theFlow,
                                 metrics,
-                                messageBroker,
                                 processReport.orElseGet(
                                     () -> new DefaultProcessReport(appProperties.name())),
                                 taskQueue,
+                                rabbitMQ,
                                 tracing))
                     .collect(
                         Collectors.toList())) // Return workers for each thread to process the Flow
         .orElse(Collections.emptyList()); // No Flow, nothing to do
   }
 
-  @Bean
-  public FlusswerkMetrics metrics(
-      ProcessingProperties processingProperties, MeterRegistry meterRegistry) {
-    return new FlusswerkMetrics(processingProperties, meterRegistry);
-  }
-
   @Bean
   public List<FlusswerkConsumer> flusswerkConsumers(
       FlusswerkObjectMapper flusswerkObjectMapper,

framework/src/main/java/com/github/dbmdz/flusswerk/framework/engine/Engine.java

@@ -1,23 +1,15 @@
 package com.github.dbmdz.flusswerk.framework.engine;
 
 import com.github.dbmdz.flusswerk.framework.flow.Flow;
-import com.github.dbmdz.flusswerk.framework.rabbitmq.MessageBroker;
 import com.rabbitmq.client.Channel;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.PriorityBlockingQueue;
-import java.util.concurrent.Semaphore;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-/**
- * Run flows {@link Flow} for every message from the {@link MessageBroker} - usually several in
- * parallel.
- */
+/** Run flows {@link Flow} for every message from RabbitMQ - usually several in parallel. */
 public class Engine {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(Engine.class);

framework/src/main/java/com/github/dbmdz/flusswerk/framework/engine/Worker.java

@@ -7,7 +7,7 @@
 import com.github.dbmdz.flusswerk.framework.model.Envelope;
 import com.github.dbmdz.flusswerk.framework.model.Message;
 import com.github.dbmdz.flusswerk.framework.monitoring.FlusswerkMetrics;
-import com.github.dbmdz.flusswerk.framework.rabbitmq.MessageBroker;
+import com.github.dbmdz.flusswerk.framework.rabbitmq.RabbitMQ;
 import com.github.dbmdz.flusswerk.framework.reporting.ProcessReport;
 import com.github.dbmdz.flusswerk.framework.reporting.Tracing;
 import java.io.IOException;
@@ -24,21 +24,22 @@ public class Worker implements Runnable {
 
   private final Flow flow;
   private final FlusswerkMetrics metrics;
-  private final MessageBroker messageBroker;
   private final ProcessReport processReport;
   private final PriorityBlockingQueue<Task> queue;
+
+  private final RabbitMQ rabbitMQ;
   private boolean running;
   private final Tracing tracing;
 
   public Worker(
       Flow flow,
       FlusswerkMetrics metrics,
-      MessageBroker messageBroker,
       ProcessReport processReport,
       PriorityBlockingQueue<Task> queue,
+      RabbitMQ rabbitMQ,
       Tracing tracing) {
     this.flow = flow;
-    this.messageBroker = messageBroker;
+    this.rabbitMQ = rabbitMQ;
     this.metrics = metrics;
     this.processReport = processReport;
     this.queue = queue;
@@ -109,9 +110,9 @@ public void process(Message message) {
     try {
       if (!messagesToSend.isEmpty()) {
         tracing.ensureFor(messagesToSend);
-        messageBroker.send(messagesToSend);
+        rabbitMQ.route("default").send(messagesToSend);
       }
-      messageBroker.ack(message);
+      rabbitMQ.ack(message);
       if (skip != null) {
         processReport.reportSkip(message, skip);
       } else {
@@ -126,8 +127,8 @@ public void process(Message message) {
 
   private void retryOrFail(Message receivedMessage, RuntimeException e) {
     try {
-      messageBroker.ack(receivedMessage);
-      boolean isRejected = messageBroker.reject(receivedMessage);
+      rabbitMQ.ack(receivedMessage);
+      boolean isRejected = rabbitMQ.retry(receivedMessage);
       if (isRejected) {
         processReport.reportRetry(receivedMessage, e);
       } else {
@@ -141,17 +142,17 @@ private void retryOrFail(Message receivedMessage, RuntimeException e) {
 
   private void complexRetry(Message receivedMessage, RetryProcessingException e) {
     try {
-      messageBroker.ack(receivedMessage);
+      rabbitMQ.ack(receivedMessage);
       for (Message retryMessage : e.getMessagesToRetry()) {
         Envelope envelope = retryMessage.getEnvelope();
         envelope.setRetries(receivedMessage.getEnvelope().getRetries());
         envelope.setSource(receivedMessage.getEnvelope().getSource());
         tracing.ensureFor(retryMessage);
-        messageBroker.reject(retryMessage);
+        rabbitMQ.retry(retryMessage);
       }
       // Send the messages that should be sent anyway
       tracing.ensureFor(e.getMessagesToSend());
-      messageBroker.send(e.getMessagesToSend());
+      rabbitMQ.route("default").send(e.getMessagesToSend());
 
       processReport.reportRetry(receivedMessage, e);
     } catch (IOException fatalException) {
@@ -163,7 +164,7 @@ private void complexRetry(Message receivedMessage, RetryProcessingException e) {
   private void fail(Message message, StopProcessingException e) {
     try {
       processReport.reportFail(message, e);
-      messageBroker.fail(message);
+      rabbitMQ.stop(message);
     } catch (IOException fatalException) {
       var body = message.getEnvelope().getBody();
       LOGGER.error("Could not fail message" + body, fatalException);

framework/src/main/java/com/github/dbmdz/flusswerk/framework/monitoring/FlusswerkMetrics.java

@@ -10,7 +10,9 @@
 import java.util.Locale;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicInteger;
+import org.springframework.stereotype.Component;
 
+@Component
 public class FlusswerkMetrics implements FlowMetrics {
 
   private final AtomicInteger activeWorkers = new AtomicInteger(0);

framework/src/main/java/com/github/dbmdz/flusswerk/framework/monitoring/MeterFactory.java

@@ -2,16 +2,25 @@
 
 import static java.util.Arrays.asList;
 
+import com.github.dbmdz.flusswerk.framework.config.properties.MonitoringProperties;
 import io.micrometer.core.instrument.Counter;
 import io.micrometer.core.instrument.MeterRegistry;
 import java.util.ArrayList;
 import java.util.List;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
 
 /** Convenience factory to simplify the creation of {@link Counter} meters. */
+@Component
 public class MeterFactory {
   private final String basename;
   private final MeterRegistry registry;
 
+  @Autowired
+  public MeterFactory(MonitoringProperties monitoringProperties, MeterRegistry registry) {
+    this(monitoringProperties.prefix(), registry);
+  }
+
   /**
    * @param basename The prefix for the created metrics ("e.g. flusswerk for flusswerk.items.total")
    * @param registry the Micrometer registry to create the counters