/*
 * Copyright The OpenTelemetry Authors
 * SPDX-License-Identifier: Apache-2.0
 */

package io.opentelemetry.sdk.metrics;

import io.opentelemetry.api.common.Labels;
import io.opentelemetry.api.metrics.AsynchronousInstrument;
import io.opentelemetry.sdk.metrics.aggregator.Aggregator;
import io.opentelemetry.sdk.metrics.data.MetricData;
import java.util.List;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer;
import javax.annotation.Nullable;

final class AsynchronousInstrumentAccumulator {
  private final ReentrantLock collectLock = new ReentrantLock();
  private final InstrumentProcessor instrumentProcessor;
  private final Consumer<InstrumentProcessor> metricUpdater;

  private AsynchronousInstrumentAccumulator(
      InstrumentProcessor instrumentProcessor, Consumer<InstrumentProcessor> metricUpdater) {
    this.metricUpdater = metricUpdater;
    this.instrumentProcessor = instrumentProcessor;
  }

  public List<MetricData> collectAll() {
    collectLock.lock();
    try {
      metricUpdater.accept(instrumentProcessor);
      return instrumentProcessor.completeCollectionCycle();
    } finally {
      collectLock.unlock();
    }
  }

  static AsynchronousInstrumentAccumulator doubleAsynchronousAccumulator(
      InstrumentProcessor instrumentProcessor,
      @Nullable Consumer<AsynchronousInstrument.DoubleResult> metricUpdater) {
    // TODO: Decide what to do with null updater.
    if (metricUpdater == null) {
      return new AsynchronousInstrumentAccumulator(instrumentProcessor, instrumentProcessor1 -> {});
    }
    DoubleResultSdk result = new DoubleResultSdk(instrumentProcessor);
    return new AsynchronousInstrumentAccumulator(
        instrumentProcessor, instrumentProcessor1 -> metricUpdater.accept(result));
  }

  static AsynchronousInstrumentAccumulator longAsynchronousAccumulator(
      InstrumentProcessor instrumentProcessor,
      @Nullable Consumer<AsynchronousInstrument.LongResult> metricUpdater) {
    // TODO: Decide what to do with null updater.
    if (metricUpdater == null) {
      return new AsynchronousInstrumentAccumulator(instrumentProcessor, instrumentProcessor1 -> {});
    }
    LongResultSdk result = new LongResultSdk(instrumentProcessor);
    return new AsynchronousInstrumentAccumulator(
        instrumentProcessor, instrumentProcessor1 -> metricUpdater.accept(result));
  }

  private static final class DoubleResultSdk implements AsynchronousInstrument.DoubleResult {
    private final InstrumentProcessor instrumentProcessor;

    private DoubleResultSdk(InstrumentProcessor instrumentProcessor) {
      this.instrumentProcessor = instrumentProcessor;
    }

    @Override
    public void observe(double sum, Labels labels) {
      Aggregator aggregator = instrumentProcessor.getAggregator();
      aggregator.recordDouble(sum);
      instrumentProcessor.batch(labels, aggregator, /* mappedAggregator= */ false);
    }
  }

  private static final class LongResultSdk implements AsynchronousInstrument.LongResult {
    private final InstrumentProcessor instrumentProcessor;

    private LongResultSdk(InstrumentProcessor instrumentProcessor) {
      this.instrumentProcessor = instrumentProcessor;
    }

    @Override
    public void observe(long sum, Labels labels) {
      Aggregator aggregator = instrumentProcessor.getAggregator();
      aggregator.recordLong(sum);
      instrumentProcessor.batch(labels, aggregator, /* mappedAggregator= */ false);
    }
  }
}

@@ -122,51 +122,8 @@
 s;%0A%0A
-import io.opentelemetry.api.common.Labels;%0A
 impo
@@ -623,527 +623,31 @@
 nal 
-Consumer%3CInstrumentProcessor%3E metricUpdater;%0A%0A  private AsynchronousInstrumentAccumulator(%0A      InstrumentProcessor instrumentProcessor, Consumer%3CInstrumentProcessor%3E metricUpdater) %7B%0A    this.metricUpdater = metricUpdater;%0A    this.instrumentProcessor = instrumentProcessor;%0A  %7D%0A%0A  public List%3CMetricData%3E collectAll() %7B%0A    collectLock.lock();%0A    try %7B%0A      metricUpdater.accept(instrumentProcessor);%0A      return instrumentProcessor.completeCollectionCycle();%0A    %7D finally %7B%0A      collectLock.unlock();%0A    %7D%0A  %7D
+Runnable metricUpdater;
 %0A%0A  
@@ -1003,47 +1003,52 @@
 or, 
-instrumentProcessor1 -%3E %7B%7D);%0A    %7D%0A    
+() -%3E %7B%7D);%0A    %7D%0A    AsynchronousInstrument.
 Doub
@@ -1051,27 +1051,24 @@
 DoubleResult
-Sdk
  result = ne
@@ -1068,50 +1068,250 @@
 lt =
- new DoubleResultSdk(instrumentProcessor);
+%0A        (value, labels) -%3E %7B%0A          Aggregator aggregator = instrumentProcessor.getAggregator();%0A          aggregator.recordDouble(value);%0A          instrumentProcessor.batch(labels, aggregator, /* mappedAggregator= */ false);%0A        %7D;%0A
 %0A   
@@ -1386,36 +1386,18 @@
 cessor, 
-instrumentProcessor1
+()
  -%3E metr
@@ -1783,602 +1783,106 @@
 or, 
-instrumentProcessor1 -%3E %7B%7D);%0A    %7D%0A    LongResultSdk result = new LongResultSdk(instrumentProcessor);%0A    return new AsynchronousInstrumentAccumulator(%0A        instrumentProcessor, instrumentProcessor1 -%3E metricUpdater.accept(result));%0A  %7D%0A%0A  private static final class DoubleResultSdk implements AsynchronousInstrument.DoubleResult %7B%0A    private final InstrumentProcessor instrumentProcessor;%0A%0A    private DoubleResultSdk(InstrumentProcessor instrumentProcessor) %7B%0A      this.instrumentProcessor = instrumentProcessor;
+() -%3E %7B%7D);%0A    %7D%0A%0A    AsynchronousInstrument.LongResult result =
 %0A    
-%7D%0A%0A
     
-@Override%0A    public void observe(double sum, Labels labels) %7B%0A
+(value, labels) -%3E %7B%0A    
     
@@ -1942,32 +1942,36 @@
 egator();%0A      
+    
 aggregator.recor
@@ -1975,21 +1975,25 @@
 cord
-Double(sum);%0A
+Long(value);%0A    
     
@@ -2080,236 +2080,295 @@
     
-%7D%0A  %7D
+    %7D;
 %0A%0A  
-private static final class LongResultSdk implements AsynchronousInstrument.LongResult %7B%0A    private final InstrumentProcessor instrumentProcessor;%0A%0A    private LongResultSdk(InstrumentProcessor instrumentProcessor) %7B%0A  
+  return new AsynchronousInstrumentAccumulator(%0A        instrumentProcessor, () -%3E metricUpdater.accept(result));%0A  %7D%0A%0A  private AsynchronousInstrumentAccumulator(%0A      InstrumentProcessor instrumentProcessor, Runnable metricUpdater) %7B%0A    this.metricUpdater = metricUpdater;%0A
     
@@ -2419,178 +2419,112 @@
 or;%0A
-  
   %7D%0A%0A  
-  @Override%0A    public void observe(long sum, Labels labels) %7B%0A      Aggregator aggregator = instrumentProcessor.getAggregator();%0A      aggregator.recordLong(sum
+public List%3CMetricData%3E collectAll() %7B%0A    collectLock.lock();%0A    try %7B%0A      metricUpdater.run(
 );%0A 
@@ -2524,24 +2524,31 @@
 un();%0A      
+return 
 instrumentPr
@@ -2559,63 +2559,76 @@
 sor.
-batch(labels, aggregator, /* mappedAggregator= */ false
+completeCollectionCycle();%0A    %7D finally %7B%0A      collectLock.unlock(
 );%0A 

Simplify logic in AsynchronousInstrumentAccumulator (#2349)