package com.atlassian.bitbucket.mesh.test.cluster;

import com.atlassian.bitbucket.mesh.io.IoConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

class ConsoleLogProcessor {

    private static final Logger log = LoggerFactory.getLogger(ConsoleLogProcessor.class);

    private final List<ConsoleLogPump> logPumps;

    ConsoleLogProcessor() {
        logPumps = new CopyOnWriteArrayList<>();
    }

    void add(Process process, IoConsumer<ByteBuffer> consumer) {
        if (consumer != null) {
            ConsoleLogPump pump = new ConsoleLogPump(process, consumer);
            logPumps.add(pump);
            pump.start();
        }
    }

    void shutdown() {
        logPumps.forEach(ConsoleLogPump::shutdown);
        logPumps.clear();
    }

    private class ConsoleLogPump {

        private final IoConsumer<ByteBuffer> consumer;
        private final Process process;
        private final Thread thread;

        ConsoleLogPump(Process process, IoConsumer<ByteBuffer> consumer) {
            this.consumer = consumer;
            this.process = process;

            thread = new Thread(this::pump, "mesh-cluster-log-pump");
            thread.setDaemon(true);
        }

        void pump() {
            try (InputStream stdout = process.getInputStream()) {
                byte[] buffer = new byte[2048];
                int read;

                while ((read = stdout.read(buffer)) != -1) {
                    ByteBuffer byteBuffer = ByteBuffer.wrap(buffer, 0, read);
                    while (byteBuffer.hasRemaining()) {
                        consumer.accept(byteBuffer);
                    }
                }
            } catch (IOException e) {
                log.warn("Error while pumping console logs from Mesh node", e);
            } finally {
                logPumps.remove(this);
            }
        }

        void shutdown() {
            if (thread.isAlive()) {
                thread.interrupt();
            }
        }

        void start() {
            thread.start();
        }
    }
}
