Refactor code

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

@@ -1,6 +1,7 @@
 package com.github.dbmdz.flusswerk.framework.engine;
 
 import com.github.dbmdz.flusswerk.framework.flow.Flow;
+import com.github.dbmdz.flusswerk.framework.rabbitmq.ChannelListener;
 import com.github.dbmdz.flusswerk.framework.rabbitmq.RabbitClient;
 import java.io.IOException;
 import java.util.ArrayList;
@@ -96,7 +97,7 @@ public void stop() {
     List<Task> remainingTasks = new ArrayList<>();
     taskQueue.drainTo(remainingTasks);
 
-    // NACK and requeue all messages that have not be processed yet
+    // NACK and requeue all messages that have not been processed yet
     for (var task : remainingTasks) {
       long deliveryTag = task.getMessage().getEnvelope().getDeliveryTag();
       try {

framework/src/main/java/com/github/dbmdz/flusswerk/framework/rabbitmq/ChannelCommand.java

@@ -0,0 +1,7 @@
+package com.github.dbmdz.flusswerk.framework.rabbitmq;
+
+import java.io.IOException;
+
+public interface ChannelCommand<T> {
+  T execute() throws IOException;
+}

framework/src/main/java/com/github/dbmdz/flusswerk/framework/rabbitmq/ChannelCommands.java

@@ -0,0 +1,81 @@
+package com.github.dbmdz.flusswerk.framework.rabbitmq;
+
+import com.rabbitmq.client.*;
+import java.util.Map;
+
+public class ChannelCommands {
+
+  private final Channel channel;
+
+  public ChannelCommands(Channel channel) {
+    this.channel = channel;
+  }
+
+  public ChannelCommand<Void> basicPublish(
+      String exchange, String routingKey, AMQP.BasicProperties properties, byte[] data) {
+    return () -> {
+      channel.basicPublish(exchange, routingKey, properties, data);
+      return null;
+    };
+  }
+
+  public ChannelCommand<Void> basicAck(long deliveryTag, boolean multiple) {
+    return () -> {
+      channel.basicAck(deliveryTag, multiple);
+      return null;
+    };
+  }
+
+  public ChannelCommand<Void> basicReject(long deliveryTag, boolean requeue) {
+    return () -> {
+      channel.basicReject(deliveryTag, requeue);
+      return null;
+    };
+  }
+
+  public ChannelCommand<GetResponse> basicGet(String queue, boolean autoAck) {
+    return () -> channel.basicGet(queue, autoAck);
+  }
+
+  public ChannelCommand<Void> basicConsume(String queue, boolean autoAck, Consumer consumer) {
+    return () -> {
+      channel.basicConsume(queue, autoAck, consumer);
+      return null;
+    };
+  }
+
+  public ChannelCommand<Void> exchangeDeclare(
+      String exchange, BuiltinExchangeType type, boolean durable) {
+    return () -> {
+      channel.exchangeDeclare(exchange, type, durable);
+      return null;
+    };
+  }
+
+  public ChannelCommand<Void> queueDeclare(
+      String name,
+      boolean durable,
+      boolean exclusive,
+      boolean autoDelete,
+      Map<String, Object> args) {
+    return () -> {
+      channel.queueDeclare(name, durable, exclusive, autoDelete, args);
+      return null;
+    };
+  }
+
+  public ChannelCommand<Void> queueBind(String name, String exchange, String routingKey) {
+    return () -> {
+      channel.queueBind(name, exchange, routingKey);
+      return null;
+    };
+  }
+
+  public ChannelCommand<Long> messageCount(String queue) {
+    return () -> channel.messageCount(queue);
+  }
+
+  public ChannelCommand<AMQP.Queue.PurgeOk> queuePurge(String queue) {
+    return () -> channel.queuePurge(queue);
+  }
+}

framework/src/main/java/com/github/dbmdz/flusswerk/framework/engine/ChannelListener.java → framework/src/main/java/com/github/dbmdz/flusswerk/framework/rabbitmq/ChannelListener.java

@@ -1,4 +1,4 @@
-package com.github.dbmdz.flusswerk.framework.engine;
+package com.github.dbmdz.flusswerk.framework.rabbitmq;
 
 /** A ChannelListener receives notifications about channel recovery. */
 public interface ChannelListener {

framework/src/main/java/com/github/dbmdz/flusswerk/framework/rabbitmq/Queue.java

@@ -4,7 +4,6 @@
 
 import com.github.dbmdz.flusswerk.framework.exceptions.InvalidMessageException;
 import com.github.dbmdz.flusswerk.framework.model.Message;
-import java.io.IOException;
 import java.util.Objects;
 import java.util.Optional;
 import org.slf4j.Logger;
@@ -38,7 +37,7 @@ public int purge() {
   /**
    * @return the number of messages in this queue.
    */
-  public long messageCount() throws IOException {
+  public long messageCount() {
     return rabbitClient.getMessageCount(this.name);
   }