/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.activemq.artemis.cli.commands.messages.perf;

import java.io.FileWriter;
import java.io.IOException;
import java.text.DecimalFormat;
import java.util.concurrent.TimeUnit;

import org.HdrHistogram.Histogram;
import org.HdrHistogram.HistogramLogWriter;
import org.apache.activemq.artemis.json.JsonObjectBuilder;
import org.apache.activemq.artemis.utils.JsonLoader;

public final class LiveStatistics {

   public enum ReportInterval {
      ns(1),
      us(TimeUnit.MICROSECONDS.toNanos(1)),
      ms(TimeUnit.MILLISECONDS.toNanos(1)),
      sec(TimeUnit.SECONDS.toNanos(1));

      public final long nanoseconds;

      ReportInterval(long value) {
         this.nanoseconds = value;
      }
   }

   private final ProducerLoadGenerator[] producers;
   private final Histogram[] waitLatencies;
   private final Histogram intervalWaitLatencies;
   private final Histogram[] sentLatencies;
   private final Histogram intervalSentLatencies;
   private final RecordingMessageListener[] listeners;
   private final Histogram[] endToEndLatencies;
   private final Histogram intervalEndToEndLatencies;
   private final RateSampler sentMsg;
   private final RateSampler blockedMsg;
   private final RateSampler completedMsg;
   private final RateSampler receivedMsg;
   private final Histogram accumulatedWaitLatencies;
   private final Histogram accumulatedSentLatencies;
   private final Histogram accumulatedEndToEndLatencies;
   private final DecimalFormat jsonLatencyFormat;
   private final PaddingDecimalFormat latencyFormat;
   private final PaddingDecimalFormat rateFormat;
   private final PaddingDecimalFormat countFormat;
   private long sampleTime;
   private final FileWriter jsonWriter;
   private long jsonSamples;
   private final HistogramLogWriter latenciesLogWriter;

   public LiveStatistics(final String jsonOutput, final String hdrOutput, final ProducerLoadGenerator[] producers, final RecordingMessageListener[] listeners) throws IOException {
      if (producers != null && producers.length > 0) {
         this.producers = producers;
         sentLatencies = new Histogram[producers.length];
         intervalSentLatencies = new Histogram(2);
         accumulatedSentLatencies = new Histogram(2);
         if (producers[0] instanceof ProducerTargetRateLoadGenerator) {
            waitLatencies = new Histogram[producers.length];
            intervalWaitLatencies = new Histogram(2);
            accumulatedWaitLatencies = new Histogram(2);
         } else {
            waitLatencies = null;
            intervalWaitLatencies = null;
            accumulatedWaitLatencies = null;
         }
         sentMsg = RateSampler.of(() -> {
            long sum = 0;
            for (ProducerLoadGenerator producer : producers) {
               sum += producer.getProducer().getMessageSent();
            }
            return sum;
         });
         blockedMsg = RateSampler.of(() -> {
            long sum = 0;
            for (ProducerLoadGenerator producer : producers) {
               sum += producer.getProducer().getNotAvailable();
            }
            return sum;
         });
         completedMsg = RateSampler.of(() -> {
            long sum = 0;
            for (ProducerLoadGenerator producer : producers) {
               sum += producer.getProducer().getMessageCompleted();
            }
            return sum;
         });
      } else {
         this.producers = null;
         sentLatencies = null;
         intervalSentLatencies = null;
         accumulatedSentLatencies = null;
         waitLatencies = null;
         intervalWaitLatencies = null;
         accumulatedWaitLatencies = null;
         sentMsg = null;
         blockedMsg = null;
         completedMsg = null;
      }
      if (listeners != null) {
         this.listeners = listeners;
         endToEndLatencies = new Histogram[listeners.length];
         intervalEndToEndLatencies = new Histogram(2);
         accumulatedEndToEndLatencies = new Histogram(2);
         receivedMsg = RateSampler.of(() -> {
            long sum = 0;
            for (RecordingMessageListener listener : listeners) {
               sum += listener.getReceivedMessages();
            }
            return sum;
         });
      } else {
         this.listeners = null;
         endToEndLatencies = null;
         intervalEndToEndLatencies = null;
         accumulatedEndToEndLatencies = null;
         receivedMsg = null;
      }
      this.sampleTime = System.currentTimeMillis();
      this.jsonLatencyFormat = new DecimalFormat("0.00");
      this.latencyFormat = new PaddingDecimalFormat("0.00", 9);
      this.rateFormat = new PaddingDecimalFormat("0", 8);
      this.countFormat = new PaddingDecimalFormat("0", 12);
      this.jsonSamples = 0;
      if (jsonOutput != null) {
         this.jsonWriter = new FileWriter(jsonOutput);
         this.jsonWriter.write("[\n");
      } else {
         this.jsonWriter = null;
      }
      if (hdrOutput != null) {
         this.latenciesLogWriter = new HistogramLogWriter(hdrOutput);
         this.latenciesLogWriter.outputLogFormatVersion();
         this.latenciesLogWriter.outputLegend();
      } else {
         this.latenciesLogWriter = null;
      }
   }

   private boolean anyFatalError() {
      if (producers != null) {
         for (ProducerLoadGenerator producer : producers) {
            if (producer.getFatalException() != null) {
               return true;
            }
         }
      }
      if (listeners != null) {
         for (RecordingMessageListener listener : listeners) {
            if (listener.anyFatalException()) {
               return true;
            }
         }
      }
      return false;
   }

   public void sampleMetrics(final boolean warmup) {
      final long lastSampleTime = this.sampleTime;
      sampleTime = System.currentTimeMillis();
      if (receivedMsg != null) {
         receivedMsg.run();
      }
      if (completedMsg != null) {
         completedMsg.run();
      }
      if (blockedMsg != null) {
         blockedMsg.run();
      }
      if (sentMsg != null) {
         sentMsg.run();
      }
      if (endToEndLatencies != null) {
         for (int i = 0, size = listeners.length; i < size; i++) {
            endToEndLatencies[i] = listeners[i].getReceiveLatencyRecorder().getIntervalHistogram(endToEndLatencies[i]);
         }
      }
      if (sentLatencies != null) {
         for (int i = 0, size = producers.length; i < size; i++) {
            sentLatencies[i] = producers[i].getSendCompletedLatencies().getIntervalHistogram(sentLatencies[i]);
         }
      }
      if (waitLatencies != null) {
         for (int i = 0, size = producers.length; i < size; i++) {
            waitLatencies[i] = producers[i].getWaitLatencies().getIntervalHistogram(waitLatencies[i]);
         }
      }
      aggregateLatencies(warmup, lastSampleTime, sampleTime, intervalEndToEndLatencies, endToEndLatencies, accumulatedEndToEndLatencies);
      aggregateLatencies(warmup, lastSampleTime, sampleTime, intervalSentLatencies, sentLatencies, accumulatedSentLatencies);
      aggregateLatencies(warmup, lastSampleTime, sampleTime, intervalWaitLatencies, waitLatencies, accumulatedWaitLatencies);
   }

   private static void aggregateLatencies(final boolean warmup,
                                          final long lastSampleTime,
                                          final long sampleTime,
                                          final Histogram intervalSentLatencies,
                                          final Histogram[] sentLatencies,
                                          final Histogram accumulatedSentLatencies) {
      if (intervalSentLatencies != null) {
         intervalSentLatencies.reset();
         intervalSentLatencies.setStartTimeStamp(lastSampleTime);
         intervalSentLatencies.setEndTimeStamp(sampleTime);
         for (Histogram histogram : sentLatencies) {
            intervalSentLatencies.add(histogram);
         }
         if (!warmup) {
            accumulatedSentLatencies.add(intervalSentLatencies);
         }
      }
   }

   public void outAtInterval(final boolean warmup, final StringBuilder out, final ReportInterval interval, final boolean includeLatencies) throws IOException {
      // space after "true" is to ensure length to be the same as "false": don't remove it!
      out.append("\n--- warmup ").append(warmup ? "true " : "false");
      appendRateOf(out, "\n--- sent:     ", sentMsg, rateFormat, interval, "msg");
      appendRateOf(out, "\n--- blocked:  ", blockedMsg, rateFormat, interval, "msg");
      appendRateOf(out, "\n--- completed:", completedMsg, rateFormat, interval, "msg");
      appendRateOf(out, "\n--- received: ", receivedMsg, rateFormat, interval, "msg");
      if (includeLatencies) {
         outPercentiles(out, "\n--- send delay time:", intervalWaitLatencies, latencyFormat);
         outPercentiles(out, "\n--- send ack time:  ", intervalSentLatencies, latencyFormat);
         outPercentiles(out, "\n--- transfer time:  ", intervalEndToEndLatencies, latencyFormat);
      }
      appendJsonIntervalSampleOnFile(warmup, interval);
      appendTaggedHdrHistograms(warmup);
   }

   private void appendJsonIntervalSampleOnFile(final boolean warmup, final ReportInterval interval) throws IOException {
      if (jsonWriter == null) {
         return;
      }
      final JsonObjectBuilder jsonBuilder = JsonLoader.createObjectBuilder();
      jsonBuilder.add("sampleType", "interval");
      jsonBuilder.add("warmup", warmup);
      jsonBuilder.add("time", sampleTime);
      addRate(jsonBuilder,"sent", sentMsg, interval);
      addRate(jsonBuilder, "delayed", blockedMsg, interval);
      addRate(jsonBuilder, "completed", completedMsg, interval);
      addRate(jsonBuilder, "received", receivedMsg, interval);
      addPercentiles(jsonBuilder, "delaySendTime", intervalWaitLatencies);
      addPercentiles(jsonBuilder, "sendTime", intervalSentLatencies);
      addPercentiles(jsonBuilder, "transferTime", intervalEndToEndLatencies);
      final String jsonSample = jsonBuilder.build().toString();
      if (jsonSamples > 0) {
         jsonWriter.write(",\n");
      }
      jsonSamples++;
      jsonWriter.write(jsonSample);
      jsonWriter.flush();
   }

   private void appendJsonSummarySampleOnFile(final boolean failedBenchmark) throws IOException {
      if (jsonWriter == null) {
         return;
      }
      final JsonObjectBuilder jsonBuilder = JsonLoader.createObjectBuilder();
      jsonBuilder.add("sampleType", "summary");
      jsonBuilder.add("time", sampleTime);
      jsonBuilder.add("result", failedBenchmark ? "fail" : "success");
      if (sentMsg != null) {
         jsonBuilder.add("totalSent", sentMsg.getLastSample());
      }
      if (blockedMsg != null) {
         jsonBuilder.add("totalBlocked", blockedMsg.getLastSample());
      }
      if (completedMsg != null) {
         jsonBuilder.add("totalCompleted", completedMsg.getLastSample());
      }
      if (receivedMsg != null) {
         jsonBuilder.add("totalReceived", receivedMsg.getLastSample());
      }
      addPercentiles(jsonBuilder, "totalDelaySendTime", accumulatedWaitLatencies);
      addPercentiles(jsonBuilder, "totalSendTime", accumulatedSentLatencies);
      addPercentiles(jsonBuilder, "totalTransferTime", accumulatedEndToEndLatencies);
      final String jsonSample = jsonBuilder.build().toString();
      if (jsonSamples > 0) {
         jsonWriter.write(",\n");
      }
      jsonSamples++;
      jsonWriter.write(jsonSample);
      jsonWriter.flush();
   }

   private void appendTaggedHdrHistograms(final boolean warmup) {
      if (latenciesLogWriter == null) {
         return;
      }
      if (intervalWaitLatencies != null) {
         intervalWaitLatencies.setTag(warmup ? "warmup delay send" : "delay send");
         latenciesLogWriter.outputIntervalHistogram(intervalWaitLatencies);
      }
      if (intervalSentLatencies != null) {
         intervalSentLatencies.setTag(warmup ? "warmup send" : "send");
         latenciesLogWriter.outputIntervalHistogram(intervalSentLatencies);
      }
      if (intervalEndToEndLatencies != null) {
         intervalEndToEndLatencies.setTag(warmup ? "warmup transfer" : "transfer");
         latenciesLogWriter.outputIntervalHistogram(intervalEndToEndLatencies);
      }
   }

   private static JsonObjectBuilder addRate(final JsonObjectBuilder obj,
                                            final String metric,
                                            final RateSampler rate,
                                            final ReportInterval interval) {
      if (rate == null) {
         return obj;
      }
      return obj.add(metric, rate.reportRate(interval.nanoseconds));
   }

   private JsonObjectBuilder addPercentiles(final JsonObjectBuilder obj,
                                                   final String metric,
                                                   final Histogram distribution) {
      if (distribution == null) {
         return obj;
      }
      return obj
         .add(metric, JsonLoader.createObjectBuilder()
         .add("mean", jsonLatencyFormat.format(distribution.getMean()))
         .add("50", distribution.getValueAtPercentile(50.0d))
         .add("90", distribution.getValueAtPercentile(90.0d))
         .add("99", distribution.getValueAtPercentile(99.0d))
         .add("99.9", distribution.getValueAtPercentile(99.9d))
         .add("99.99", distribution.getValueAtPercentile(99.99d))
         .add("max", distribution.getMaxValue())
         .add("count", distribution.getTotalCount()));
   }

   public void outSummary(final StringBuilder out) throws IOException {
      out.append("\n--- SUMMARY");
      final boolean failedBenchmark = anyFatalError();
      out.append("\n--- result:         ").append(failedBenchmark ? "        fail" : "     success");
      if (sentMsg != null) {
         out.append("\n--- total sent:     ").append(countFormat.format(sentMsg.getLastSample()));
      }
      if (blockedMsg != null) {
         out.append("\n--- total blocked:  ").append(countFormat.format(blockedMsg.getLastSample()));
      }
      if (completedMsg != null) {
         out.append("\n--- total completed:").append(countFormat.format(completedMsg.getLastSample()));
      }
      if (receivedMsg != null) {
         out.append("\n--- total received: ").append(countFormat.format(receivedMsg.getLastSample()));
      }
      outPercentiles(out, "\n--- aggregated delay send time:", accumulatedWaitLatencies, latencyFormat);
      outPercentiles(out, "\n--- aggregated send time:      ", accumulatedSentLatencies, latencyFormat);
      outPercentiles(out, "\n--- aggregated transfer time:  ", accumulatedEndToEndLatencies, latencyFormat);
      appendJsonSummarySampleOnFile(failedBenchmark);
   }

   public void close() throws IOException {
      if (jsonWriter != null) {
         jsonWriter.write("\n]");
         jsonWriter.close();
      }
      if (latenciesLogWriter != null) {
         latenciesLogWriter.close();
      }
   }

   private static void outPercentiles(final StringBuilder out,
                                      final String metric,
                                      final Histogram histogram,
                                      final DecimalFormat latencyFormat) {
      if (histogram == null) {
         return;
      }
      out.append(' ').append(metric);
      out.append(' ').append("mean: ").append(latencyFormat.format(histogram.getMean())).append(" us");
      out.append(" - ").append("50.00%: ").append(latencyFormat.format(histogram.getValueAtPercentile(50.0d))).append(" us");
      out.append(" - ").append("90.00%: ").append(latencyFormat.format(histogram.getValueAtPercentile(90.0d))).append(" us");
      out.append(" - ").append("99.00%: ").append(latencyFormat.format(histogram.getValueAtPercentile(99.0d))).append(" us");
      out.append(" - ").append("99.90%: ").append(latencyFormat.format(histogram.getValueAtPercentile(99.9d))).append(" us");
      out.append(" - ").append("99.99%: ").append(latencyFormat.format(histogram.getValueAtPercentile(99.99d))).append(" us");
      out.append(" - ").append("max:    ").append(latencyFormat.format(histogram.getMaxValue())).append(" us");
   }

   private static StringBuilder appendRateOf(final StringBuilder out,
                                             final String metric,
                                             final RateSampler sampler,
                                             final DecimalFormat rateFormat,
                                             final ReportInterval outInterval,
                                             final String unit) {
      if (sampler == null) {
         return out;
      }
      return out.append(' ').append(metric)
         .append(' ').append(rateFormat.format(sampler.reportRate(outInterval.nanoseconds)))
         .append(' ').append(unit).append('/').append(outInterval.name());
   }
}
