Merge pull request #526 from dbmdz/reporting

Remove additional process reports

CHANGES.md

@@ -11,6 +11,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 - Outgoing routes may now include a list of topics instead of a single topic. A `Route` can be used to send a message to several topics at once.
+- The `DefaultFlusswerkReport` logs tracing information.
+
+### Removed
+- `FlusswerkProcessReport`, `StructuredProcessReport` and `SilentProcessReport` have been removed. Users who would like to customize their process report can subclass [DefaultProcessReport](framework/src/main/java/com/github/dbmdz/flusswerk/framework/reporting/DefaultProcessReport.java)
+or implement the [ProcessReport](framework/src/main/java/com/github/dbmdz/flusswerk/framework/reporting/ProcessReport.java) interface.
 
 ## [6.0.1] - 2023-12-06
 

framework/src/main/java/com/github/dbmdz/flusswerk/framework/reporting/DefaultProcessReport.java

@@ -5,46 +5,55 @@
 
 import com.github.dbmdz.flusswerk.framework.exceptions.RetryProcessingException;
 import com.github.dbmdz.flusswerk.framework.exceptions.StopProcessingException;
+import com.github.dbmdz.flusswerk.framework.model.Envelope;
 import com.github.dbmdz.flusswerk.framework.model.Message;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 public class DefaultProcessReport implements ProcessReport {
 
-  private static final Logger LOGGER = LoggerFactory.getLogger(DefaultProcessReport.class);
-
+  private final FlusswerkLogger logger;
   private final String name;
-  private final Tracing tracing;
 
   public DefaultProcessReport(String name, Tracing tracing) {
     this.name = requireNonNull(name);
-    this.tracing = requireNonNull(tracing);
+    FlusswerkLoggerFactory loggerFactory = new FlusswerkLoggerFactory(requireNonNull(tracing));
+    this.logger = loggerFactory.getFlusswerkLogger(DefaultProcessReport.class);
+  }
+
+  public DefaultProcessReport(String name, FlusswerkLogger logger) {
+    this.name = requireNonNull(name);
+    this.logger = requireNonNull(logger);
   }
 
   @Override
   public void reportSuccess(Message message) {
-    LOGGER.info("{} successful", name, keyValue("tracing", tracing.tracingPath()));
+    getLogger().info("{} successful", name);
   }
 
   @Override
   public void reportFail(Message message, StopProcessingException e) {
+    Envelope envelope = message.getEnvelope();
     getLogger()
         .error(
             "{} failed terminally: {}",
             name,
             e.getMessage(),
-            keyValue("amqp_message", message.toString()),
+            keyValue("will_retry", false),
+            keyValue("incoming_queue", envelope.getSource()),
+            keyValue("retries", envelope.getRetries()),
             e);
   }
 
   @Override
   public void reportFailAfterMaxRetries(Message message, Exception e) {
+    Envelope envelope = message.getEnvelope();
     getLogger()
         .error(
             "{} failed after maximum number of retries: {}",
             name,
             e.getMessage(),
-            keyValue("amqp_message", message.toString()),
+            keyValue("will_retry", false),
+            keyValue("incoming_queue", envelope.getSource()),
+            keyValue("retries", envelope.getRetries()),
             e);
   }
 
@@ -58,36 +67,42 @@ public void reportReject(Message message, Exception e) {
 
   @Override
   public void reportRetry(Message message, RuntimeException e) {
+    Envelope envelope = message.getEnvelope();
     getLogger()
         .warn(
             "{} rejected for retry: {}",
             name,
             e.getMessage(),
-            keyValue("amqp_message", message.toString()),
+            keyValue("will_retry", true),
+            keyValue("incoming_queue", envelope.getSource()),
+            keyValue("retries", envelope.getRetries()),
             e);
   }
 
   @Override
   public void reportComplexRetry(Message message, RetryProcessingException e) {
     int newMessagesToRetry = e.getMessagesToRetry().size();
     int messagesSent = e.getMessagesToSend().size();
+    Envelope envelope = message.getEnvelope();
     getLogger()
         .warn(
             "{} rejected for retry with ({} new, {} sent) and : {}",
             name,
             newMessagesToRetry,
             messagesSent,
             e.getMessage(),
-            keyValue("amqp_message", message.toString()),
+            keyValue("will_retry", true),
+            keyValue("incoming_queue", envelope.getSource()),
+            keyValue("retries", envelope.getRetries()),
             e);
   }
 
   @Override
   public void reportSkip(Message message, Exception skip) {
-    LOGGER.info("Skipped: {}", skip.getMessage());
+    getLogger().info("Skipped: {}", skip.getMessage());
   }
 
-  protected Logger getLogger() {
-    return LOGGER;
+  protected FlusswerkLogger getLogger() {
+    return logger;
   }
 }