/*
 * Copyright 2016 Crown Copyright
 *
 * 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 stroom.pipeline.server.writer;

import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;
import stroom.pipeline.server.errorhandler.ErrorReceiverProxy;
import stroom.pipeline.server.errorhandler.ProcessException;
import stroom.pipeline.server.factory.ConfigurableElement;
import stroom.pipeline.server.factory.PipelineProperty;
import stroom.pipeline.shared.ElementIcons;
import stroom.pipeline.shared.data.PipelineElementType;
import stroom.pipeline.shared.data.PipelineElementType.Category;
import stroom.util.io.ByteCountOutputStream;
import stroom.util.io.FileUtil;
import stroom.util.spring.StroomScope;

import javax.inject.Inject;
import java.io.BufferedOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.zip.GZIPOutputStream;

/**
 * Joins text instances into a single text instance.
 */
@Component
@Scope(StroomScope.PROTOTYPE)
@ConfigurableElement(
        type = "FileAppender",
        category = Category.DESTINATION,
        roles = {PipelineElementType.ROLE_TARGET,
                PipelineElementType.ROLE_DESTINATION,
                PipelineElementType.VISABILITY_STEPPING},
        icon = ElementIcons.FILE)
public class FileAppender extends AbstractAppender {
    private static final String LOCK_EXTENSION = ".lock";

    private final PathCreator pathCreator;
    private ByteCountOutputStream byteCountOutputStream;
    private String[] outputPaths;
    private boolean useCompression;

    @Inject
    public FileAppender(final ErrorReceiverProxy errorReceiverProxy,
                        final PathCreator pathCreator) {
        super(errorReceiverProxy);
        this.pathCreator = pathCreator;
    }

    @Override
    protected OutputStream createOutputStream() throws IOException {
        try {
            if (outputPaths == null || outputPaths.length == 0) {
                throw new IOException("No output paths have been set");
            }

            // Get a path to use.
            String path;
            if (outputPaths.length == 1) {
                path = outputPaths[0];
            } else {
                // Choose one of the output paths at random.
                path = outputPaths[(int) Math.round(Math.random() * (outputPaths.length - 1))];
            }

            // Replace some of the path elements with system variables.
            path = pathCreator.replaceAll(path);

            // Make sure we can create this path.
            final Path file = Paths.get(path);
            final Path dir = file.getParent();
            if (!Files.isDirectory(dir)) {
                try {
                    Files.createDirectories(dir);
                } catch (final IOException e) {
                    throw new ProcessException("Unable to create output dirs: " + FileUtil.getCanonicalPath(dir));
                }
            }

            final Path lockFile = Paths.get(path + LOCK_EXTENSION);
            final Path outFile = file;

            // Make sure we can create both output files without overwriting
            // another file.
            if (Files.exists(lockFile)) {
                throw new ProcessException("Output file \"" + FileUtil.getCanonicalPath(lockFile) + "\" already exists");
            }
            if (Files.exists(outFile)) {
                throw new ProcessException("Output file \"" + FileUtil.getCanonicalPath(outFile) + "\" already exists");
            }

            // Get a writer for the new lock file.
            if (useCompression) {
                byteCountOutputStream = new ByteCountOutputStream(new GZIPOutputStream(new BufferedOutputStream(Files.newOutputStream(lockFile))));
            }
            else {
                byteCountOutputStream = new ByteCountOutputStream(new BufferedOutputStream(Files.newOutputStream(lockFile)));
            }

            return new LockedOutputStream(byteCountOutputStream, lockFile, outFile);

        } catch (final IOException e) {
            throw e;
        } catch (final Exception e) {
            throw new IOException(e.getMessage(), e);
        }
    }

    @Override
    long getCurrentOutputSize() {
        if (byteCountOutputStream == null) {
            return 0;
        }
        return byteCountOutputStream.getCount();
    }

    /**
     * @param outputPaths the outputPaths to set
     */
    @PipelineProperty(description = "One or more destination paths for output files separated with commas. Replacement variables can be used in path strings such as ${feed}.")
    public void setOutputPaths(final String outputPaths) {
        this.outputPaths = outputPaths.split(",");
    }

    @PipelineProperty(description = "Apply GZIP compression to output files", defaultValue = "false")
    public void setUseCompression(final boolean useCompression) { this.useCompression = useCompression; }

    @SuppressWarnings("unused")
    @PipelineProperty(description = "When the current output file exceeds this size it will be closed and a new one created.")
    public void setRollSize(final String size) {
        super.setRollSize(size);
    }

    @PipelineProperty(description = "Choose if you want to split aggregated streams into separate output files.", defaultValue = "false")
    public void setSplitAggregatedStreams(final boolean splitAggregatedStreams) {
        super.setSplitAggregatedStreams(splitAggregatedStreams);
    }

    @PipelineProperty(description = "Choose if you want to split individual records into separate output files.", defaultValue = "false")
    public void setSplitRecords(final boolean splitRecords) {
        super.setSplitRecords(splitRecords);
    }
}

@@ -637,110 +637,8 @@
 r;%0A%0A
-import org.springframework.context.annotation.Scope;%0Aimport org.springframework.stereotype.Component;%0A
 impo
@@ -1137,26 +1137,218 @@
 til.
-spring.StroomScope
+io.GZipByteCountOutputStream;%0Aimport stroom.util.io.GZipOutputStream;%0Aimport stroom.util.spring.StroomScope;%0A%0Aimport org.springframework.context.annotation.Scope;%0Aimport org.springframework.stereotype.Component
 ;%0A%0Ai
@@ -1523,16 +1523,16 @@
 e.Path;%0A
+
 import j
@@ -1554,47 +1554,8 @@
 ths;
-%0Aimport java.util.zip.GZIPOutputStream;
 %0A%0A/*
@@ -4298,32 +4298,36 @@
 putStream = new 
+GZip
 ByteCountOutputS
@@ -4342,35 +4342,10 @@
 w GZ
-IPOutputStream(new Buffered
+ip
 Outp
@@ -4386,17 +4386,16 @@
 kFile)))
-)
 ;%0A      

Switch to custom GZIP class and avoid unnecessary use of `BufferedOutputStream`