/*
 * Copyright 2018 Organizations participating in ISAAC, ISAAC's KOMET, and SOLOR development include the
         US Veterans Health Administration, OSHERA, and the Health Services Platform Consortium..
 *
 * Licensed 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 sh.isaac.model.datastream;

import java.io.DataInputStream;
import java.io.FileInputStream;
import java.nio.file.Path;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import sh.isaac.api.Get;
import sh.isaac.api.externalizable.ByteArrayDataBuffer;
import sh.isaac.api.externalizable.IsaacExternalizable;
import sh.isaac.api.task.TimedTaskWithProgressTracker;

/**
 *
 * @author kec
 */
public class BinaryDatastreamReader
        extends TimedTaskWithProgressTracker<Integer> {

    private final BiConsumer<? super IsaacExternalizable, byte[]> action;
    private final Path path;
    private final int permits;
    private final Semaphore processingSemaphore;
    private final long bytesToProcess;
    private final AtomicReference<Throwable> exception = new AtomicReference<>();
    
    /**
     * Read an IBDF file, parse each object, and call the provided consumer with each parsed object.
     * @param action where to send the parsed objects from the file
     * @param path - the file to read
     * @param processInOrder - if true, this runs single threaded, so the ibdf file will be parsed in order.  If false, 
     * the parsing runs in parallel threads, and may happen out-of-order.
     */
    public BinaryDatastreamReader(BiConsumer<? super IsaacExternalizable, byte[]> action, Path path, boolean processInOrder) {
        this.action = action;
        this.path = path;
        this.bytesToProcess = path.toFile().length();
        permits = processInOrder ? 1 : Runtime.getRuntime().availableProcessors() * 2;
        processingSemaphore = new Semaphore(permits);
        addToTotalWork(this.bytesToProcess);
        updateTitle("Importing from " + path.toFile().getName());
        Get.activeTasks().add(this);
    }
    
    /**
     * Calls {@link #BinaryDatastreamReader(BiConsumer, Path, boolean)} with processInOrder set to false.
     * @param action where to send the parsed objects from the file
     * @param path - the file to read
     */
    public BinaryDatastreamReader(BiConsumer<? super IsaacExternalizable, byte[]> action, Path path) {
        this(action, path, false);
    }

    public BinaryDatastreamReader(Consumer<? super IsaacExternalizable> action, Path path) {
        this((externalizable, data) -> {action.accept(externalizable);}, path);
    }

    @Override
    protected Integer call() throws Exception {
        try (DataInputStream input = new DataInputStream(new FileInputStream(path.toFile()))) {
            long bytesProcessed = 0;
            int objectCount = 0;
            while (this.bytesToProcess > bytesProcessed) {
                throwIfException();
                objectCount++;
                final int recordSizeInBytes = input.readInt();
                bytesProcessed += 4;
                final byte[] objectData = new byte[recordSizeInBytes];
                input.readFully(objectData);
                bytesProcessed += recordSizeInBytes;
 
                ByteArrayDataBuffer byteArrayDataBuffer = new ByteArrayDataBuffer(objectData);
                byteArrayDataBuffer.setExternalData(true);
                IsaacExternalizableUnparsed unparsedObject
                        = new IsaacExternalizableUnparsed(byteArrayDataBuffer);
                completedUnitsOfWork(recordSizeInBytes);

                this.processingSemaphore.acquireUninterruptibly();
                
                Processor processor = new Processor(unparsedObject);
                Get.executor().execute(processor);
            }
            this.processingSemaphore.acquireUninterruptibly(permits);
            throwIfException();
            return objectCount;
        } finally {
            Get.activeTasks().remove(this);
        }
    }

    private void throwIfException() throws Exception {
        if (exception.get() != null) {
            throw new Exception("Exceptions during import", exception.get());
        }
    }

    private class Processor implements Runnable {

        final IsaacExternalizableUnparsed unparsedObject;

        public Processor(IsaacExternalizableUnparsed unparsedObject) {
            this.unparsedObject = unparsedObject;
        }

        @Override
        public void run() {
            try {
                IsaacExternalizable isaacObject = unparsedObject.parse();
                action.accept(isaacObject, unparsedObject.getBytes());
            } catch (Throwable t) {
                exception.set(t);
            } finally {
                BinaryDatastreamReader.this.processingSemaphore.release();
            }
        }

    }

}

@@ -917,25 +917,160 @@
 current.
-Semaphore
+LinkedBlockingQueue;%0Aimport java.util.concurrent.Semaphore;%0Aimport java.util.concurrent.ThreadPoolExecutor;%0Aimport java.util.concurrent.TimeUnit
 ;%0Aimport
@@ -1380,16 +1380,111 @@
 Tracker;
+%0Aimport sh.isaac.api.util.NamedThreadFactory;%0Aimport sh.isaac.api.util.ThreadPoolExecutorFixed;
 %0A%0A/**%0A *
@@ -1896,24 +1896,72 @@
 erence%3C%3E();%0A
+    private ThreadPoolExecutor parsingExecutor;%0A
     %0A    /**
@@ -2629,29 +2629,8 @@
 ts =
- processInOrder ? 1 :
  Run
@@ -2720,32 +2720,269 @@
 phore(permits);%0A
+        parsingExecutor = processInOrder ? %0A              new ThreadPoolExecutorFixed(1, 1, 5, TimeUnit.MINUTES, new LinkedBlockingQueue%3C%3E(), new NamedThreadFactory(%22BinaryDataStreamOrderedRead%22, false)) %0A              : Get.executor();%0A
         addToTot
@@ -4822,30 +4822,31 @@
         
-Get.e
+parsingE
 xecutor
-()
 .execute
@@ -5018,32 +5018,303 @@
     %7D finally %7B%0A
+            if (parsingExecutor != Get.executor()) %7B%0A                //they asked for in-order parsing, we need to shut down our own executor%0A                parsingExecutor.shutdown();%0A                parsingExecutor.awaitTermination(1, TimeUnit.MINUTES);%0A            %7D%0A
             Get.

use a local thread pool when in-order processing was requested, so we don't block out the IO thread pool waiting for acquires