/*
 * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
 * or more contributor license agreements. Licensed under the Elastic License;
 * you may not use this file except in compliance with the Elastic License.
 */
package org.elasticsearch.xpack.ml.job.process.autodetect;

import org.elasticsearch.xpack.ml.job.config.DetectionRule;
import org.elasticsearch.xpack.ml.job.config.ModelPlotConfig;
import org.elasticsearch.xpack.ml.job.process.autodetect.output.FlushAcknowledgement;
import org.elasticsearch.xpack.ml.job.process.autodetect.params.DataLoadParams;
import org.elasticsearch.xpack.ml.job.process.autodetect.params.InterimResultsParams;
import org.elasticsearch.xpack.ml.job.results.AutodetectResult;

import java.io.IOException;
import java.time.ZonedDateTime;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

/**
 * A placeholder class simulating the actions of the native Autodetect process.
 * Most methods consume data without performing any action however, after a call to
 * {@link #flushJob(InterimResultsParams)} a {@link org.elasticsearch.xpack.ml.job.process.autodetect.output.FlushAcknowledgement}
 * message is expected on the {@link #readAutodetectResults()} ()} stream. This class writes the flush
 * acknowledgement immediately.
 */
public class BlackHoleAutodetectProcess implements AutodetectProcess {

    private static final String FLUSH_ID = "flush-1";
    private static final AutodetectResult EMPTY = new AutodetectResult(null, null, null, null, null, null, null, null, null);

    private final ZonedDateTime startTime;

    private final BlockingQueue<AutodetectResult> results = new ArrayBlockingQueue<>(128);

    public BlackHoleAutodetectProcess() {
        startTime = ZonedDateTime.now();
    }

    @Override
    public void writeRecord(String[] record) throws IOException {
    }

    @Override
    public void writeResetBucketsControlMessage(DataLoadParams params) throws IOException {
    }

    @Override
    public void writeUpdateModelPlotMessage(ModelPlotConfig modelPlotConfig) throws IOException {
    }

    @Override
    public void writeUpdateDetectorRulesMessage(int detectorIndex, List<DetectionRule> rules) throws IOException {
    }

    /**
     * Accept the request do nothing with it but write the flush acknowledgement to {@link #readAutodetectResults()}
     * @param params Should interim results be generated
     * @return {@link #FLUSH_ID}
     */
    @Override
    public String flushJob(InterimResultsParams params) throws IOException {
        FlushAcknowledgement flushAcknowledgement = new FlushAcknowledgement(FLUSH_ID);
        AutodetectResult result = new AutodetectResult(null, null, null, null, null, null, null, null, flushAcknowledgement);
        results.add(result);
        return FLUSH_ID;
    }

    @Override
    public void flushStream() throws IOException {
    }

    @Override
    public void close() throws IOException {
        results.add(EMPTY);
    }

    @Override
    public Iterator<AutodetectResult> readAutodetectResults() {
        // Create a custom iterator here, because ArrayBlockingQueue iterator and stream are not blocking when empty:
        return new Iterator<AutodetectResult>() {

            AutodetectResult result;

            @Override
            public boolean hasNext() {
                try {
                    result = results.take();
                    return result != EMPTY;
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    return false;
                }
            }

            @Override
            public AutodetectResult next() {
                return result;
            }
        };
    }

    @Override
    public ZonedDateTime getProcessStartTime() {
        return startTime;
    }

    @Override
    public boolean isProcessAlive() {
        return true;
    }

    @Override
    public String readError() {
        return "";
    }
}

@@ -872,21 +872,16 @@
 current.
-Array
 Blocking
@@ -911,32 +911,38 @@
 .concurrent.
+Linked
 Blocking
 Queue;%0A%0A/**%0A
@@ -929,21 +929,59 @@
 Blocking
-Queue
+Deque;%0Aimport java.util.concurrent.TimeUnit
 ;%0A%0A/**%0A 
@@ -1542,134 +1542,8 @@
 -1%22;
-%0A    private static final AutodetectResult EMPTY = new AutodetectResult(null, null, null, null, null, null, null, null, null);
 %0A%0A  
@@ -1583,17 +1583,16 @@
 rtTime;%0A
-%0A
     priv
@@ -1647,29 +1647,30 @@
 s = new 
-Array
+Linked
 Blocking
 Queue%3C%3E(
@@ -1665,20 +1665,59 @@
 king
-Que
+Deq
 ue%3C%3E(
-128)
+);%0A    private volatile boolean open = true
 ;%0A%0A 
@@ -2992,26 +2992,20 @@
     
-results.add(EMPTY)
+open = false
 ;%0A  
@@ -3141,21 +3141,22 @@
 use 
-Array
+Linked
 Blocking
 Queu
@@ -3151,19 +3151,19 @@
 Blocking
-Que
+Deq
 ue itera
@@ -3402,30 +3402,95 @@
     
-result = results.take(
+while (open) %7B%0A                        result = results.poll(100, TimeUnit.MILLISECONDS
 );%0A 
@@ -3504,31 +3504,32 @@
             
-return 
+    if (
 result != EM
@@ -3530,13 +3530,130 @@
  != 
-EMPTY
+null) %7B%0A                            break;%0A                        %7D%0A                    %7D%0A                    return open
 ;%0A  

Use volatile boolean to stop blackhole process