/*
 * 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.nifi.processors.hadoop;

import java.io.IOException;
import java.security.PrivilegedExceptionAction;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.regex.Pattern;

import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsAction;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.nifi.annotation.behavior.InputRequirement;
import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
import org.apache.nifi.annotation.behavior.WritesAttribute;
import org.apache.nifi.annotation.behavior.WritesAttributes;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.SeeAlso;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.components.AllowableValue;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.PropertyValue;
import org.apache.nifi.components.ValidationContext;
import org.apache.nifi.components.ValidationResult;
import org.apache.nifi.expression.ExpressionLanguageScope;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.flowfile.attributes.CoreAttributes;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.ProcessorInitializationContext;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.util.StandardValidators;

import static org.apache.nifi.processors.hadoop.GetHDFSFileInfo.HDFSFileInfoRequest.Grouping.ALL;
import static org.apache.nifi.processors.hadoop.GetHDFSFileInfo.HDFSFileInfoRequest.Grouping.DIR;
import static org.apache.nifi.processors.hadoop.GetHDFSFileInfo.HDFSFileInfoRequest.Grouping.NONE;

@InputRequirement(Requirement.INPUT_ALLOWED)
@Tags({"hadoop", "HCFS", "HDFS", "get", "list", "ingest", "source", "filesystem"})
@CapabilityDescription("Retrieves a listing of files and directories from HDFS. "
        + "This processor creates a FlowFile(s) that represents the HDFS file/dir with relevant information. "
        + "Main purpose of this processor to provide functionality similar to HDFS Client, i.e. count, du, ls, test, etc. "
        + "Unlike ListHDFS, this processor is stateless, supports incoming connections and provides information on a dir level. "
)
@WritesAttributes({
        @WritesAttribute(attribute = "hdfs.objectName", description = "The name of the file/dir found on HDFS."),
        @WritesAttribute(attribute = "hdfs.path", description = "The path is set to the absolute path of the object's parent directory on HDFS. "
                + "For example, if an object is a directory 'foo', under directory '/bar' then 'hdfs.objectName' will have value 'foo', and 'hdfs.path' will be '/bar'"),
        @WritesAttribute(attribute = "hdfs.type", description = "The type of an object. Possible values: directory, file, link"),
        @WritesAttribute(attribute = "hdfs.owner", description = "The user that owns the object in HDFS"),
        @WritesAttribute(attribute = "hdfs.group", description = "The group that owns the object in HDFS"),
        @WritesAttribute(attribute = "hdfs.lastModified", description = "The timestamp of when the object in HDFS was last modified, as milliseconds since midnight Jan 1, 1970 UTC"),
        @WritesAttribute(attribute = "hdfs.length", description = ""
                + "In case of files: The number of bytes in the file in HDFS.  "
                + "In case of dirs: Retuns storage space consumed by directory. "
                + ""),
        @WritesAttribute(attribute = "hdfs.count.files", description = "In case of type='directory' will represent total count of files under this dir. "
                + "Won't be populated to other types of HDFS objects. "),
        @WritesAttribute(attribute = "hdfs.count.dirs", description = "In case of type='directory' will represent total count of directories under this dir (including itself). "
                + "Won't be populated to other types of HDFS objects. "),
        @WritesAttribute(attribute = "hdfs.replication", description = "The number of HDFS replicas for the file"),
        @WritesAttribute(attribute = "hdfs.permissions", description = "The permissions for the object in HDFS. This is formatted as 3 characters for the owner, "
                + "3 for the group, and 3 for other users. For example rw-rw-r--"),
        @WritesAttribute(attribute = "hdfs.status", description = "The status contains comma separated list of file/dir paths, which couldn't be listed/accessed. "
                + "Status won't be set if no errors occured."),
        @WritesAttribute(attribute = "hdfs.full.tree", description = "When destination is 'attribute', will be populated with full tree of HDFS directory in JSON format."
                + "WARNING: In case when scan finds thousands or millions of objects, having huge values in attribute could impact flow file repo and GC/heap usage. "
                + "Use content destination for such cases")
})
@SeeAlso({ListHDFS.class, GetHDFS.class, FetchHDFS.class, PutHDFS.class})
public class GetHDFSFileInfo extends AbstractHadoopProcessor {
    public static final String APPLICATION_JSON = "application/json";
    public static final PropertyDescriptor FULL_PATH = new PropertyDescriptor.Builder()
            .displayName("Full path")
            .name("gethdfsfileinfo-full-path")
            .description("A directory to start listing from, or a file's full path.")
            .required(true)
            .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
            .defaultValue("")
            .addValidator(StandardValidators.NON_BLANK_VALIDATOR)
            .build();

    public static final PropertyDescriptor RECURSE_SUBDIRS = new PropertyDescriptor.Builder()
            .displayName("Recurse Subdirectories")
            .name("gethdfsfileinfo-recurse-subdirs")
            .description("Indicates whether to list files from subdirectories of the HDFS directory")
            .required(true)
            .allowableValues("true", "false")
            .defaultValue("true")
            .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
            .build();

    public static final PropertyDescriptor DIR_FILTER = new PropertyDescriptor.Builder()
            .displayName("Directory Filter")
            .name("gethdfsfileinfo-dir-filter")
            .description("Regex. Only directories whose names match the given regular expression will be picked up. If not provided, any filter would be apply (performance considerations).")
            .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
            .required(false)
            .addValidator(StandardValidators.createRegexValidator(0, Integer.MAX_VALUE, true))
            .build();

    public static final PropertyDescriptor FILE_FILTER = new PropertyDescriptor.Builder()
            .displayName("File Filter")
            .name("gethdfsfileinfo-file-filter")
            .description("Regex. Only files whose names match the given regular expression will be picked up. If not provided, any filter would be apply (performance considerations).")
            .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
            .required(false)
            .addValidator(StandardValidators.createRegexValidator(0, Integer.MAX_VALUE, true))
            .build();

    public static final PropertyDescriptor FILE_EXCLUDE_FILTER = new PropertyDescriptor.Builder()
            .displayName("Exclude Files")
            .name("gethdfsfileinfo-file-exclude-filter")
            .description("Regex. Files whose names match the given regular expression will not be picked up. If not provided, any filter won't be apply (performance considerations).")
            .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
            .required(false)
            .addValidator(StandardValidators.createRegexValidator(0, Integer.MAX_VALUE, true))
            .build();

    public static final PropertyDescriptor IGNORE_DOTTED_DIRS = new PropertyDescriptor.Builder()
            .displayName("Ignore Dotted Directories")
            .name("gethdfsfileinfo-ignore-dotted-dirs")
            .description("If true, directories whose names begin with a dot (\".\") will be ignored")
            .required(true)
            .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
            .allowableValues("true", "false")
            .defaultValue("true")
            .build();

    public static final PropertyDescriptor IGNORE_DOTTED_FILES = new PropertyDescriptor.Builder()
            .displayName("Ignore Dotted Files")
            .name("gethdfsfileinfo-ignore-dotted-files")
            .description("If true, files whose names begin with a dot (\".\") will be ignored")
            .required(true)
            .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
            .allowableValues("true", "false")
            .defaultValue("true")
            .build();

    static final AllowableValue GROUP_ALL = new AllowableValue("gethdfsfileinfo-group-all", "All",
            "Group all results into a single flowfile.");

    static final AllowableValue GROUP_PARENT_DIR = new AllowableValue("gethdfsfileinfo-group-parent-dir", "Parent Directory",
            "Group HDFS objects by their parent directories only. Processor will generate flowfile for each directory (if recursive). "
                    + "If 'Recurse Subdirectories' property set to 'false', then will have the same effect as 'All'");

    static final AllowableValue GROUP_NONE = new AllowableValue("gethdfsfileinfo-group-none", "None",
            "Don't group results. Generate flowfile per each HDFS object.");

    public static final PropertyDescriptor GROUPING = new PropertyDescriptor.Builder()
            .displayName("Group Results")
            .name("gethdfsfileinfo-group")
            .description("Groups HDFS objects")
            .required(true)
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .allowableValues(GROUP_ALL, GROUP_PARENT_DIR, GROUP_NONE)
            .defaultValue(GROUP_ALL.getValue())
            .build();

    public static final PropertyDescriptor BATCH_SIZE = new PropertyDescriptor.Builder()
            .displayName("Batch Size")
            .name("gethdfsfileinfo-batch-size")
            .description("Number of records to put into an output flowfile when 'Destination' is set to 'Content'"
                    + " and 'Group Results' is set to 'None'")
            .required(false)
            .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
            .build();

    static final AllowableValue DESTINATION_ATTRIBUTES = new AllowableValue("gethdfsfileinfo-dest-attr", "Attributes",
            "Details of given HDFS object will be stored in attributes of flowfile. "
                    + "WARNING: In case when scan finds thousands or millions of objects, having huge values in attribute could impact flow file repo and GC/heap usage. "
                    + "Use content destination for such cases.");

    static final AllowableValue DESTINATION_CONTENT = new AllowableValue("gethdfsfileinfo-dest-content", "Content",
            "Details of given HDFS object will be stored in a content in JSON format");

    public static final PropertyDescriptor DESTINATION = new PropertyDescriptor.Builder()
            .displayName("Destination")
            .name("gethdfsfileinfo-destination")
            .description("Sets the destination for the resutls. When set to 'Content', attributes of flowfile won't be used for storing results. ")
            .required(true)
            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
            .allowableValues(DESTINATION_ATTRIBUTES, DESTINATION_CONTENT)
            .defaultValue(DESTINATION_CONTENT.getValue())
            .build();

    public static final Relationship REL_SUCCESS = new Relationship.Builder()
            .name("success")
            .description("All successfully generated FlowFiles are transferred to this relationship")
            .build();

    public static final Relationship REL_NOT_FOUND = new Relationship.Builder()
            .name("not found")
            .description("If no objects are found, original FlowFile are transferred to this relationship")
            .build();

    public static final Relationship REL_ORIGINAL = new Relationship.Builder()
            .name("original")
            .description("Original FlowFiles are transferred to this relationship")
            .build();

    public static final Relationship REL_FAILURE = new Relationship.Builder()
            .name("failure")
            .description("All failed attempts to access HDFS will be routed to this relationship")
            .build();

    @Override
    protected void init(final ProcessorInitializationContext context) {
        super.init(context);
    }

    @Override
    protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
        final List<PropertyDescriptor> props = new ArrayList<>(properties);
        props.add(FULL_PATH);
        props.add(RECURSE_SUBDIRS);
        props.add(DIR_FILTER);
        props.add(FILE_FILTER);
        props.add(FILE_EXCLUDE_FILTER);
        props.add(IGNORE_DOTTED_DIRS);
        props.add(IGNORE_DOTTED_FILES);
        props.add(GROUPING);
        props.add(BATCH_SIZE);
        props.add(DESTINATION);
        return props;
    }

    @Override
    public Set<Relationship> getRelationships() {
        final Set<Relationship> relationships = new HashSet<>();
        relationships.add(REL_SUCCESS);
        relationships.add(REL_NOT_FOUND);
        relationships.add(REL_ORIGINAL);
        relationships.add(REL_FAILURE);
        return relationships;
    }

    @Override
    protected Collection<ValidationResult> customValidate(ValidationContext validationContext) {
        final List<ValidationResult> validationResults = new ArrayList<>(super.customValidate(validationContext));

        String destination = validationContext.getProperty(DESTINATION).getValue();
        String grouping = validationContext.getProperty(GROUPING).getValue();
        String batchSize = validationContext.getProperty(BATCH_SIZE).getValue();

        if (
                (!DESTINATION_CONTENT.getValue().equals(destination) || !GROUP_NONE.getValue().equals(grouping))
                        && batchSize != null
        ) {
            validationResults.add(new ValidationResult.Builder()
                    .valid(false)
                    .subject(BATCH_SIZE.getDisplayName())
                    .explanation("'" + BATCH_SIZE.getDisplayName() + "' is applicable only when " +
                            "'" + DESTINATION.getDisplayName() + "'='" + DESTINATION_CONTENT.getDisplayName() + "' and " +
                            "'" + GROUPING.getDisplayName() + "'='" + GROUP_NONE.getDisplayName() + "'")
                    .build());
        }

        return validationResults;
    }

    @Override
    public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
        FlowFile ff = null;
        if (context.hasIncomingConnection()) {
            ff = session.get();

            // If we have no FlowFile, and all incoming connections are self-loops then we can continue on.
            // However, if we have no FlowFile and we have connections coming from other Processors, then
            // we know that we should run only if we have a FlowFile.
            if (ff == null && context.hasNonLoopConnection()) {
                context.yield();
                return;
            }
        }
        boolean scheduledFF = false;
        if (ff == null) {
            ff = session.create();
            scheduledFF = true;
        }
        HDFSFileInfoRequest req = buildRequestDetails(context, ff);

        try {
            final FileSystem hdfs = getFileSystem();
            UserGroupInformation ugi = getUserGroupInformation();
            ExecutionContext executionContext = new ExecutionContext();
            HDFSObjectInfoDetails res = walkHDFSTree(session, executionContext, ff, hdfs, ugi, req, null, false);
            executionContext.finish(session);
            if (res == null) {
                ff = session.putAttribute(ff, "hdfs.status", "Path not found: " + req.fullPath);
                session.transfer(ff, REL_NOT_FOUND);
                return;
            }
            if (!scheduledFF) {
                session.transfer(ff, REL_ORIGINAL);
            } else {
                session.remove(ff);
            }
        } catch (final InterruptedException e) {
            Thread.currentThread().interrupt();
            getLogger().error("Interrupted while performing listing of HDFS", e);
            ff = session.putAttribute(ff, "hdfs.status", "Failed due to: " + e);
            session.transfer(ff, REL_FAILURE);
        } catch (final Exception e) {
            getLogger().error("Failed to perform listing of HDFS due to {}", new Object[]{e});
            ff = session.putAttribute(ff, "hdfs.status", "Failed due to: " + e);
            session.transfer(ff, REL_FAILURE);
        }
    }


    /*
     * Walks thru HDFS tree. This method will return null to the main if there is no provided path existing.
     */
    protected HDFSObjectInfoDetails walkHDFSTree(final ProcessSession session, ExecutionContext executionContext,
                                                 FlowFile origFF, final FileSystem hdfs, final UserGroupInformation ugi,
                                                 final HDFSFileInfoRequest req, HDFSObjectInfoDetails parent, final boolean statsOnly
    ) throws Exception {

        final HDFSObjectInfoDetails p = parent;

        if (!ugi.doAs((PrivilegedExceptionAction<Boolean>) () -> hdfs.exists(p != null ? p.getPath() : new Path(req.getFullPath())))) {
            return null;
        }

        if (parent == null) {
            parent = new HDFSObjectInfoDetails(ugi.doAs((PrivilegedExceptionAction<FileStatus>) () -> hdfs.getFileStatus(new Path(req.getFullPath()))));
        }
        if (parent.isFile() && p == null) {
            //single file path requested and found, lets send to output:
            processHDFSObject(session, executionContext, origFF, req, parent, true);
            return parent;
        }

        final Path path = parent.getPath();

        FileStatus[] listFSt = null;
        try {
            listFSt = ugi.doAs((PrivilegedExceptionAction<FileStatus[]>) () -> hdfs.listStatus(path));
        } catch (IOException e) {
            parent.error = "Couldn't list directory: " + e;
            processHDFSObject(session, executionContext, origFF, req, parent, p == null);
            return parent; //File not found exception, or access denied - don't interrupt, just don't list
        }
        if (listFSt != null) {
            for (FileStatus f : listFSt) {
                HDFSObjectInfoDetails o = new HDFSObjectInfoDetails(f);
                HDFSObjectInfoDetails vo = validateMatchingPatterns(o, req);
                if (o.isDirectory() && !o.isSymlink() && req.isRecursive()) {
                    o = walkHDFSTree(session, executionContext, origFF, hdfs, ugi, req, o, vo == null || statsOnly);
                    parent.countDirs += o.countDirs;
                    parent.totalLen += o.totalLen;
                    parent.countFiles += o.countFiles;
                } else if (o.isDirectory() && o.isSymlink()) {
                    parent.countDirs += 1;
                } else if (o.isFile() && !o.isSymlink()) {
                    parent.countFiles += 1;
                    parent.totalLen += o.getLen();
                } else if (o.isFile() && o.isSymlink()) {
                    parent.countFiles += 1; // do not add length of the symlink, as it doesn't consume space under THIS directory, but count files, as it is still an object.
                }

                // Decide what to do with child: if requested FF per object or per dir - just emit new FF with info in 'o' object
                if (vo != null && !statsOnly) {
                    parent.addChild(vo);
                    if (vo.isFile() && !vo.isSymlink()) {
                        processHDFSObject(session, executionContext, origFF, req, vo, false);
                    }
                }
            }
            if (!statsOnly) {
                processHDFSObject(session, executionContext, origFF, req, parent, p == null);
            }
            if (req.getGrouping() != ALL) {
                parent.setChildren(null); //we need children in full tree only when single output requested.
            }
        }

        return parent;
    }

    protected HDFSObjectInfoDetails validateMatchingPatterns(final HDFSObjectInfoDetails o, HDFSFileInfoRequest req) {
        if (o == null || o.getPath() == null) {
            return null;
        }

        if (o.isFile()) {
            if (req.isIgnoreDotFiles() && o.getPath().getName().startsWith(".")) {
                return null;
            } else if (req.getFileExcludeFilter() != null && req.getFileExcludeFilter().matcher(o.getPath().getName()).matches()) {
                return null;
            } else if (req.getFileFilter() != null && req.getFileFilter().matcher(o.getPath().getName()).matches()) {
                return o;
            } else if (req.getFileFilter() == null) {
                return o;
            }
            return null;
        }

        if (o.isDirectory()) {
            if (req.isIgnoreDotDirs() && o.getPath().getName().startsWith(".")) {
                return null;
            } else if (req.getDirFilter() != null && req.getDirFilter().matcher(o.getPath().getName()).matches()) {
                return o;
            } else if (req.getDirFilter() == null) {
                return o;
            }
        }
        return null;
    }

    /*
     * Checks whether HDFS object should be sent to output.
     * If it should be sent, new flowfile will be created, its content and attributes will be populated according to other request params.
     */
    protected void processHDFSObject(
            final ProcessSession session,
            final ExecutionContext executionContext,
            FlowFile origFF,
            final HDFSFileInfoRequest req,
            final HDFSObjectInfoDetails o,
            final boolean isRoot
    ) {
        if (o.isFile() && req.getGrouping() != NONE) {
            return;
        }
        if (o.isDirectory() && o.isSymlink() && req.getGrouping() != NONE) {
            return;
        }
        if (o.isDirectory() && req.getGrouping() == ALL && !isRoot) {
            return;
        }

        FlowFile ff = getReadyFlowFile(executionContext, session, origFF);

        //if destination type is content - always add mime type
        if (req.isDestContent()) {
            ff = session.putAttribute(ff, CoreAttributes.MIME_TYPE.key(), APPLICATION_JSON);
        }

        //won't combine conditions for similar actions for better readability and maintenance.
        if (o.isFile() && isRoot && req.isDestContent()) {
            ff = addAsContent(executionContext, session, o, ff);
            // ------------------------------
        } else if (o.isFile() && isRoot && !req.isDestContent()) {
            ff = addAsAttributes(session, o, ff);
            // ------------------------------
        } else if (o.isFile() && req.isDestContent()) {
            ff = addAsContent(executionContext, session, o, ff);
            // ------------------------------
        } else if (o.isFile() && !req.isDestContent()) {
            ff = addAsAttributes(session, o, ff);
            // ------------------------------
        } else if (o.isDirectory() && o.isSymlink() && req.isDestContent()) {
            ff = addAsContent(executionContext, session, o, ff);
            // ------------------------------
        } else if (o.isDirectory() && o.isSymlink() && !req.isDestContent()) {
            ff = addAsAttributes(session, o, ff);
            // ------------------------------
        } else if (o.isDirectory() && req.getGrouping() == NONE && req.isDestContent()) {
            o.setChildren(null);
            ff = addAsContent(executionContext, session, o, ff);
            // ------------------------------
        } else if (o.isDirectory() && req.getGrouping() == NONE && !req.isDestContent()) {
            ff = addAsAttributes(session, o, ff);
            // ------------------------------
        } else if (o.isDirectory() && req.getGrouping() == DIR && req.isDestContent()) {
            ff = addAsContent(executionContext, session, o, ff);
            // ------------------------------
        } else if (o.isDirectory() && req.getGrouping() == DIR && !req.isDestContent()) {
            ff = addAsAttributes(session, o, ff);
            ff = addFullTreeToAttribute(session, o, ff);
            // ------------------------------
        } else if (o.isDirectory() && req.getGrouping() == ALL && req.isDestContent()) {
            ff = addAsContent(executionContext, session, o, ff);
            // ------------------------------
        } else if (o.isDirectory() && req.getGrouping() == ALL && !req.isDestContent()) {
            ff = addAsAttributes(session, o, ff);
            ff = addFullTreeToAttribute(session, o, ff);
        } else {
            getLogger().error("Illegal State!");
            session.remove(ff);
            return;
        }

        executionContext.flowfile = ff;
        finishProcessing(req, executionContext, session);
    }

    private FlowFile getReadyFlowFile(ExecutionContext executionContext, ProcessSession session, FlowFile origFF) {
        if (executionContext.flowfile == null) {
            executionContext.flowfile = session.create(origFF);
        }

        return executionContext.flowfile;
    }

    private void finishProcessing(HDFSFileInfoRequest req, ExecutionContext executionContext, ProcessSession session) {
        executionContext.nrOfWaitingHDFSObjects++;

        if (req.grouping == NONE && req.isDestContent() && executionContext.nrOfWaitingHDFSObjects < req.getBatchSize()) {
            return;
        }
        session.transfer(executionContext.flowfile, REL_SUCCESS);

        executionContext.reset();
    }

    private FlowFile addAsContent(ExecutionContext executionContext, ProcessSession session, HDFSObjectInfoDetails o, FlowFile ff) {
        if (executionContext.nrOfWaitingHDFSObjects > 0) {
            ff = session.append(ff, (out) -> out.write(("\n").getBytes()));
        }

        return session.append(ff, (out) -> out.write((o.toJsonString()).getBytes()));
    }

    private FlowFile addAsAttributes(ProcessSession session, HDFSObjectInfoDetails o, FlowFile ff) {
        return session.putAllAttributes(ff, o.toAttributesMap());
    }

    private FlowFile addFullTreeToAttribute(ProcessSession session, HDFSObjectInfoDetails o, FlowFile ff) {
        return session.putAttribute(ff, "hdfs.full.tree", o.toJsonString());
    }

    /*
     * Returns permissions in readable format like rwxr-xr-x (755)
     */
    protected String getPerms(final FsPermission permission) {

        final StringBuilder sb = new StringBuilder();
        for (FsAction action : new FsAction[]{permission.getUserAction(), permission.getGroupAction(), permission.getOtherAction()}) {
            if (action.implies(FsAction.READ)) {
                sb.append("r");
            } else {
                sb.append("-");
            }

            if (action.implies(FsAction.WRITE)) {
                sb.append("w");
            } else {
                sb.append("-");
            }

            if (action.implies(FsAction.EXECUTE)) {
                sb.append("x");
            } else {
                sb.append("-");
            }
        }

        return sb.toString();
    }

    /*
     * Creates internal request object and initialize the fields that won't be changed every call (onTrigger).
     * Dynamic fields will be updated per each call separately.
     */
    protected HDFSFileInfoRequest buildRequestDetails(ProcessContext context, FlowFile ff) {
        HDFSFileInfoRequest req = new HDFSFileInfoRequest();
        String fullPath = getNormalizedPath(context, FULL_PATH, ff).toString();
        req.setFullPath(fullPath);
        req.setRecursive(context.getProperty(RECURSE_SUBDIRS).asBoolean());

        PropertyValue pv;
        String v;

        if (context.getProperty(DIR_FILTER).isSet() && (pv = context.getProperty(DIR_FILTER).evaluateAttributeExpressions(ff)) != null) {
            v = pv.getValue();
            req.setDirFilter(v == null ? null : Pattern.compile(v));
        }

        if (context.getProperty(FILE_FILTER).isSet() && (pv = context.getProperty(FILE_FILTER).evaluateAttributeExpressions(ff)) != null) {
            v = pv.getValue();
            req.setFileFilter(v == null ? null : Pattern.compile(v));
        }

        if (context.getProperty(FILE_EXCLUDE_FILTER).isSet()
                && (pv = context.getProperty(FILE_EXCLUDE_FILTER).evaluateAttributeExpressions(ff)) != null) {
            v = pv.getValue();
            req.setFileExcludeFilter(v == null ? null : Pattern.compile(v));
        }

        req.setIgnoreDotFiles(context.getProperty(IGNORE_DOTTED_FILES).asBoolean());
        req.setIgnoreDotDirs(context.getProperty(IGNORE_DOTTED_DIRS).asBoolean());

        req.setGrouping(HDFSFileInfoRequest.Grouping.getEnum(context.getProperty(GROUPING).getValue()));
        req.setBatchSize(context.getProperty(BATCH_SIZE).asInteger() != null ? context.getProperty(BATCH_SIZE).asInteger() : 1);

        v = context.getProperty(DESTINATION).getValue();

        req.setDestContent(DESTINATION_CONTENT.getValue().equals(v));

        return req;
    }

    static class ExecutionContext {
        int nrOfWaitingHDFSObjects;
        FlowFile flowfile;

        void reset() {
            nrOfWaitingHDFSObjects = 0;
            flowfile = null;
        }

        void finish(ProcessSession session) {
            if (flowfile != null) {
                session.transfer(flowfile, REL_SUCCESS);
                flowfile = null;
            }
        }
    }

    /*
     * Keeps all request details in single object.
     */
    static class HDFSFileInfoRequest {
        enum Grouping {
            ALL(GROUP_ALL.getValue()),
            DIR(GROUP_PARENT_DIR.getValue()),
            NONE(GROUP_NONE.getValue());

            final private String val;

            Grouping(String val) {
                this.val = val;
            }

            public String toString() {
                return this.val;
            }

            public static Grouping getEnum(String value) {
                for (Grouping v : values()) {
                    if (v.val.equals(value)) {
                        return v;
                    }
                }
                return null;
            }
        }

        private String fullPath;
        private boolean recursive;
        private Pattern dirFilter;
        private Pattern fileFilter;
        private Pattern fileExcludeFilter;
        private boolean ignoreDotFiles;
        private boolean ignoreDotDirs;
        private boolean destContent;
        private Grouping grouping;
        private int batchSize;

        String getFullPath() {
            return fullPath;
        }

        void setFullPath(String fullPath) {
            this.fullPath = fullPath;
        }

        boolean isRecursive() {
            return this.recursive;
        }

        void setRecursive(boolean recursive) {
            this.recursive = recursive;
        }

        Pattern getDirFilter() {
            return this.dirFilter;
        }

        void setDirFilter(Pattern dirFilter) {
            this.dirFilter = dirFilter;
        }

        Pattern getFileFilter() {
            return fileFilter;
        }

        void setFileFilter(Pattern fileFilter) {
            this.fileFilter = fileFilter;
        }

        Pattern getFileExcludeFilter() {
            return fileExcludeFilter;
        }

        void setFileExcludeFilter(Pattern fileExcludeFilter) {
            this.fileExcludeFilter = fileExcludeFilter;
        }

        boolean isIgnoreDotFiles() {
            return ignoreDotFiles;
        }

        void setIgnoreDotFiles(boolean ignoreDotFiles) {
            this.ignoreDotFiles = ignoreDotFiles;
        }

        boolean isIgnoreDotDirs() {
            return this.ignoreDotDirs;
        }

        void setIgnoreDotDirs(boolean ignoreDotDirs) {
            this.ignoreDotDirs = ignoreDotDirs;
        }

        boolean isDestContent() {
            return this.destContent;
        }

        void setDestContent(boolean destContent) {
            this.destContent = destContent;
        }

        Grouping getGrouping() {
            return grouping;
        }

        void setGrouping(Grouping grouping) {
            this.grouping = grouping;
        }

        int getBatchSize() {
            return this.batchSize;
        }

        void setBatchSize(int batchSize) {
            this.batchSize = batchSize;
        }
    }

    /*
     * Keeps details of HDFS objects.
     * This class is based on FileStatus and adds additional feature/properties for count, total size of directories, and subtrees/hierarchy of recursive listings.
     */
    class HDFSObjectInfoDetails extends FileStatus {

        private long countFiles;
        private long countDirs = 1;
        private long totalLen;
        private Collection<HDFSObjectInfoDetails> children = new LinkedList<>();
        private String error;

        HDFSObjectInfoDetails(FileStatus fs) throws IOException {
            super(fs);
        }

        public long getCountFiles() {
            return countFiles;
        }

        public void setCountFiles(long countFiles) {
            this.countFiles = countFiles;
        }

        public long getCountDirs() {
            return countDirs;
        }

        public void setCountDirs(long countDirs) {
            this.countDirs = countDirs;
        }

        public long getTotalLen() {
            return totalLen;
        }

        public void setTotalLen(long totalLen) {
            this.totalLen = totalLen;
        }

        public Collection<HDFSObjectInfoDetails> getChildren() {
            return children;
        }

        public String getError() {
            return error;
        }

        public void setError(String error) {
            this.error = error;
        }

        public void setChildren(Collection<HDFSObjectInfoDetails> children) {
            this.children = children;
        }

        public void addChild(HDFSObjectInfoDetails child) {
            this.children.add(child);
        }

        public void updateTotals(boolean deepUpdate) {
            if (deepUpdate) {
                this.countDirs = 1;
                this.countFiles = 0;
                this.totalLen = 0;
            }

            for (HDFSObjectInfoDetails c : children) {
                if (c.isSymlink()) {
                    continue; //do not count symlinks. they either will be counted under their actual directories, or won't be count if actual location is not under provided root for scan.
                } else if (c.isDirectory()) {
                    if (deepUpdate) {
                        c.updateTotals(deepUpdate);
                    }
                    this.totalLen += c.totalLen;
                    this.countDirs += c.countDirs;
                    this.countFiles += c.countFiles;
                } else if (c.isFile()) {
                    this.totalLen += c.getLen();
                    this.countFiles++;
                }
            }

        }

        /*
         * Since, by definition, FF will keep only attributes for parent/single object, we don't need to recurse the children
         */
        public Map<String, String> toAttributesMap() {
            Map<String, String> map = new HashMap<>();

            map.put("hdfs.objectName", this.getPath().getName());
            map.put("hdfs.path", Path.getPathWithoutSchemeAndAuthority(this.getPath().getParent()).toString());
            map.put("hdfs.type", this.isSymlink() ? "link" : (this.isDirectory() ? "directory" : "file"));
            map.put("hdfs.owner", this.getOwner());
            map.put("hdfs.group", this.getGroup());
            map.put("hdfs.lastModified", Long.toString(this.getModificationTime()));
            map.put("hdfs.length", Long.toString(this.isDirectory() ? this.totalLen : this.getLen()));
            map.put("hdfs.replication", Long.toString(this.getReplication()));
            if (this.isDirectory()) {
                map.put("hdfs.count.files", Long.toString(this.getCountFiles()));
                map.put("hdfs.count.dirs", Long.toString(this.getCountDirs()));
            }
            map.put("hdfs.permissions", getPerms(this.getPermission()));
            if (this.error != null) {
                map.put("hdfs.status", "Error: " + this.error);
            }

            return map;
        }

        /*
         * The decision to use custom serialization (vs jackson/velocity/gson/etc) is behind the performance.
         * This object is pretty simple, with limited number of members of simple types.
         */
        public String toJsonString() {
            StringBuilder sb = new StringBuilder();
            return toJsonString(sb).toString();
        }

        private StringBuilder toJsonString(StringBuilder sb) {
            sb.append("{");

            appendProperty(sb, "objectName", this.getPath().getName()).append(",");
            appendProperty(sb, "path", Path.getPathWithoutSchemeAndAuthority(this.getPath().getParent()).toString()).append(",");
            appendProperty(sb, "type", this.isSymlink() ? "link" : (this.isDirectory() ? "directory" : "file")).append(",");
            appendProperty(sb, "owner", this.getOwner()).append(",");
            appendProperty(sb, "group", this.getGroup()).append(",");
            appendProperty(sb, "lastModified", Long.toString(this.getModificationTime())).append(",");
            appendProperty(sb, "length", Long.toString(this.isDirectory() ? this.totalLen : this.getLen())).append(",");
            appendProperty(sb, "replication", Long.toString(this.getReplication())).append(",");
            if (this.isDirectory()) {
                appendProperty(sb, "countFiles", Long.toString(this.getCountFiles())).append(",");
                appendProperty(sb, "countDirs", Long.toString(this.getCountDirs())).append(",");
            }
            appendProperty(sb, "permissions", getPerms(this.getPermission()));

            if (this.error != null) {
                sb.append(",");
                appendProperty(sb, "status", this.error);
            }

            if (this.getChildren() != null && this.getChildren().size() > 0) {
                sb.append(",\"content\":[");
                for (HDFSObjectInfoDetails c : this.getChildren()) {
                    c.toJsonString(sb).append(",");
                }
                sb.deleteCharAt(sb.length() - 1).append("]");
            }
            sb.append("}");
            return sb;
        }

        private StringBuilder appendProperty(StringBuilder sb, String name, String value) {
            return sb.append("\"").append(name).append("\":\"").append(value == null ? "" : value).append("\"");
        }
    }
}