Making callbacks aware of the signal type

Author: LikeTheSalad

disk-buffering/src/main/java/io/opentelemetry/contrib/disk/buffering/SignalType.java

@@ -6,7 +6,6 @@
 package io.opentelemetry.contrib.disk.buffering.exporters;
 
 import com.google.errorprone.annotations.CanIgnoreReturnValue;
-import io.opentelemetry.contrib.disk.buffering.SignalType;
 import io.opentelemetry.contrib.disk.buffering.exporters.callback.ExporterCallback;
 import io.opentelemetry.contrib.disk.buffering.internal.exporters.SignalStorageExporter;
 import io.opentelemetry.contrib.disk.buffering.storage.SignalStorage;
@@ -19,11 +18,11 @@
 /** Exporter that stores logs into disk. */
 public final class LogRecordToDiskExporter implements LogRecordExporter {
   private final SignalStorageExporter<LogRecordData> storageExporter;
-  private final ExporterCallback callback;
-  private static final SignalType TYPE = SignalType.LOG;
+  private final ExporterCallback<LogRecordData> callback;
 
   private LogRecordToDiskExporter(
-      SignalStorageExporter<LogRecordData> storageExporter, ExporterCallback callback) {
+      SignalStorageExporter<LogRecordData> storageExporter,
+      ExporterCallback<LogRecordData> callback) {
     this.storageExporter = storageExporter;
     this.callback = callback;
   }
@@ -44,17 +43,17 @@ public CompletableResultCode flush() {
 
   @Override
   public CompletableResultCode shutdown() {
-    callback.onShutdown(TYPE);
+    callback.onShutdown();
     return CompletableResultCode.ofSuccess();
   }
 
   public static final class Builder {
     private final SignalStorage.LogRecord storage;
-    private ExporterCallback callback = ExporterCallback.noop();
+    private ExporterCallback<LogRecordData> callback = ExporterCallback.noop();
     private Duration writeTimeout = Duration.ofSeconds(10);
 
     @CanIgnoreReturnValue
-    public Builder setExporterCallback(ExporterCallback value) {
+    public Builder setExporterCallback(ExporterCallback<LogRecordData> value) {
       callback = value;
       return this;
     }
@@ -67,7 +66,7 @@ public Builder setWriteTimeout(Duration value) {
 
     public LogRecordToDiskExporter build() {
       SignalStorageExporter<LogRecordData> storageExporter =
-          new SignalStorageExporter<>(storage, callback, writeTimeout, TYPE);
+          new SignalStorageExporter<>(storage, callback, writeTimeout);
       return new LogRecordToDiskExporter(storageExporter, callback);
     }
 

disk-buffering/src/main/java/io/opentelemetry/contrib/disk/buffering/exporters/LogRecordToDiskExporter.java

@@ -6,7 +6,6 @@
 package io.opentelemetry.contrib.disk.buffering.exporters;
 
 import com.google.errorprone.annotations.CanIgnoreReturnValue;
-import io.opentelemetry.contrib.disk.buffering.SignalType;
 import io.opentelemetry.contrib.disk.buffering.exporters.callback.ExporterCallback;
 import io.opentelemetry.contrib.disk.buffering.internal.exporters.SignalStorageExporter;
 import io.opentelemetry.contrib.disk.buffering.storage.SignalStorage;
@@ -23,13 +22,12 @@
 public final class MetricToDiskExporter implements MetricExporter {
   private final SignalStorageExporter<MetricData> storageExporter;
   private final AggregationTemporalitySelector aggregationTemporalitySelector;
-  private final ExporterCallback callback;
-  private static final SignalType TYPE = SignalType.METRIC;
+  private final ExporterCallback<MetricData> callback;
 
   private MetricToDiskExporter(
       SignalStorageExporter<MetricData> storageExporter,
       AggregationTemporalitySelector aggregationTemporalitySelector,
-      ExporterCallback callback) {
+      ExporterCallback<MetricData> callback) {
     this.storageExporter = storageExporter;
     this.aggregationTemporalitySelector = aggregationTemporalitySelector;
     this.callback = callback;
@@ -51,7 +49,7 @@ public CompletableResultCode flush() {
 
   @Override
   public CompletableResultCode shutdown() {
-    callback.onShutdown(TYPE);
+    callback.onShutdown();
     return CompletableResultCode.ofSuccess();
   }
 
@@ -64,11 +62,11 @@ public static final class Builder {
     private final SignalStorage.Metric storage;
     private AggregationTemporalitySelector aggregationTemporalitySelector =
         AggregationTemporalitySelector.alwaysCumulative();
-    private ExporterCallback callback = ExporterCallback.noop();
+    private ExporterCallback<MetricData> callback = ExporterCallback.noop();
     private Duration writeTimeout = Duration.ofSeconds(10);
 
     @CanIgnoreReturnValue
-    public Builder setExporterCallback(ExporterCallback value) {
+    public Builder setExporterCallback(ExporterCallback<MetricData> value) {
       callback = value;
       return this;
     }
@@ -87,7 +85,7 @@ public Builder setAggregationTemporalitySelector(AggregationTemporalitySelector
 
     public MetricToDiskExporter build() {
       SignalStorageExporter<MetricData> storageExporter =
-          new SignalStorageExporter<>(storage, callback, writeTimeout, TYPE);
+          new SignalStorageExporter<>(storage, callback, writeTimeout);
       return new MetricToDiskExporter(storageExporter, aggregationTemporalitySelector, callback);
     }
 

disk-buffering/src/main/java/io/opentelemetry/contrib/disk/buffering/exporters/MetricToDiskExporter.java

@@ -6,7 +6,6 @@
 package io.opentelemetry.contrib.disk.buffering.exporters;
 
 import com.google.errorprone.annotations.CanIgnoreReturnValue;
-import io.opentelemetry.contrib.disk.buffering.SignalType;
 import io.opentelemetry.contrib.disk.buffering.exporters.callback.ExporterCallback;
 import io.opentelemetry.contrib.disk.buffering.internal.exporters.SignalStorageExporter;
 import io.opentelemetry.contrib.disk.buffering.storage.SignalStorage;
@@ -19,11 +18,10 @@
 /** Exporter that stores spans into disk. */
 public final class SpanToDiskExporter implements SpanExporter {
   private final SignalStorageExporter<SpanData> storageExporter;
-  private final ExporterCallback callback;
-  private static final SignalType TYPE = SignalType.SPAN;
+  private final ExporterCallback<SpanData> callback;
 
   private SpanToDiskExporter(
-      SignalStorageExporter<SpanData> storageExporter, ExporterCallback callback) {
+      SignalStorageExporter<SpanData> storageExporter, ExporterCallback<SpanData> callback) {
     this.storageExporter = storageExporter;
     this.callback = callback;
   }
@@ -44,17 +42,17 @@ public CompletableResultCode flush() {
 
   @Override
   public CompletableResultCode shutdown() {
-    callback.onShutdown(TYPE);
+    callback.onShutdown();
     return CompletableResultCode.ofSuccess();
   }
 
   public static final class Builder {
     private final SignalStorage.Span storage;
-    private ExporterCallback callback = ExporterCallback.noop();
+    private ExporterCallback<SpanData> callback = ExporterCallback.noop();
     private Duration writeTimeout = Duration.ofSeconds(10);
 
     @CanIgnoreReturnValue
-    public Builder setExporterCallback(ExporterCallback value) {
+    public Builder setExporterCallback(ExporterCallback<SpanData> value) {
       callback = value;
       return this;
     }
@@ -67,7 +65,7 @@ public Builder setWriteTimeout(Duration value) {
 
     public SpanToDiskExporter build() {
       SignalStorageExporter<SpanData> storageExporter =
-          new SignalStorageExporter<>(storage, callback, writeTimeout, TYPE);
+          new SignalStorageExporter<>(storage, callback, writeTimeout);
       return new SpanToDiskExporter(storageExporter, callback);
     }
 

disk-buffering/src/main/java/io/opentelemetry/contrib/disk/buffering/exporters/SpanToDiskExporter.java

@@ -5,34 +5,30 @@
 
 package io.opentelemetry.contrib.disk.buffering.exporters.callback;
 
-import io.opentelemetry.contrib.disk.buffering.SignalType;
+import java.util.Collection;
 import javax.annotation.Nullable;
 
 /** Notifies about exporter and storage-related operations from within a signal to disk exporter. */
-public interface ExporterCallback {
+public interface ExporterCallback<T> {
   /**
    * Called when an export to disk operation succeeded.
    *
-   * @param type The type of signal associated to the exporter.
+   * @param items The items successfully stored in disk.
    */
-  void onExportSuccess(SignalType type);
+  void onExportSuccess(Collection<T> items);
 
   /**
    * Called when an export to disk operation failed.
    *
-   * @param type The type of signal associated to the exporter.
+   * @param items The items that couldn't get stored in disk.
    * @param error Optional - provides more information of why the operation failed.
    */
-  void onExportError(SignalType type, @Nullable Throwable error);
+  void onExportError(Collection<T> items, @Nullable Throwable error);
 
-  /**
-   * Called when the exporter is closed.
-   *
-   * @param type The type of signal associated to the exporter.
-   */
-  void onShutdown(SignalType type);
+  /** Called when the exporter is closed. */
+  void onShutdown();
 
-  static ExporterCallback noop() {
-    return NoopExporterCallback.INSTANCE;
+  static <T> ExporterCallback<T> noop() {
+    return new NoopExporterCallback<>();
   }
 }

disk-buffering/src/main/java/io/opentelemetry/contrib/disk/buffering/exporters/callback/ExporterCallback.java

@@ -5,20 +5,19 @@
 
 package io.opentelemetry.contrib.disk.buffering.exporters.callback;
 
-import io.opentelemetry.contrib.disk.buffering.SignalType;
+import java.util.Collection;
 import javax.annotation.Nullable;
 
-final class NoopExporterCallback implements ExporterCallback {
-  static final NoopExporterCallback INSTANCE = new NoopExporterCallback();
+final class NoopExporterCallback<T> implements ExporterCallback<T> {
 
-  private NoopExporterCallback() {}
+  NoopExporterCallback() {}
 
   @Override
-  public void onExportSuccess(SignalType type) {}
+  public void onExportSuccess(Collection<T> items) {}
 
   @Override
-  public void onExportError(SignalType type, @Nullable Throwable error) {}
+  public void onExportError(Collection<T> items, @Nullable Throwable error) {}
 
   @Override
-  public void onShutdown(SignalType type) {}
+  public void onShutdown() {}
 }

disk-buffering/src/main/java/io/opentelemetry/contrib/disk/buffering/exporters/callback/NoopExporterCallback.java

@@ -5,7 +5,6 @@
 
 package io.opentelemetry.contrib.disk.buffering.internal.exporters;
 
-import io.opentelemetry.contrib.disk.buffering.SignalType;
 import io.opentelemetry.contrib.disk.buffering.exporters.callback.ExporterCallback;
 import io.opentelemetry.contrib.disk.buffering.storage.SignalStorage;
 import io.opentelemetry.contrib.disk.buffering.storage.result.WriteResult;
@@ -20,35 +19,33 @@
 /** Internal utility for common export to disk operations across all exporters. */
 public final class SignalStorageExporter<T> {
   private final SignalStorage<T> storage;
-  private final ExporterCallback callback;
+  private final ExporterCallback<T> callback;
   private final Duration writeTimeout;
-  private final SignalType type;
 
   public SignalStorageExporter(
-      SignalStorage<T> storage, ExporterCallback callback, Duration writeTimeout, SignalType type) {
+      SignalStorage<T> storage, ExporterCallback<T> callback, Duration writeTimeout) {
     this.storage = storage;
     this.callback = callback;
     this.writeTimeout = writeTimeout;
-    this.type = type;
   }
 
   public CompletableResultCode exportToStorage(Collection<T> items) {
     CompletableFuture<WriteResult> future = storage.write(items);
     try {
       WriteResult operation = future.get(writeTimeout.toMillis(), TimeUnit.MILLISECONDS);
       if (operation.isSuccessful()) {
-        callback.onExportSuccess(type);
+        callback.onExportSuccess(items);
         return CompletableResultCode.ofSuccess();
       }
 
       Throwable error = operation.getError();
-      callback.onExportError(type, error);
+      callback.onExportError(items, error);
       if (error != null) {
         return CompletableResultCode.ofExceptionalFailure(error);
       }
       return CompletableResultCode.ofFailure();
     } catch (ExecutionException | InterruptedException | TimeoutException e) {
-      callback.onExportError(type, e);
+      callback.onExportError(items, e);
       return CompletableResultCode.ofExceptionalFailure(e);
     }
   }