/*
 * 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.asterix.external.input.stream.factory;

import java.io.File;
import java.io.IOException;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.apache.asterix.common.cluster.ClusterPartition;
import org.apache.asterix.common.exceptions.AsterixException;
import org.apache.asterix.external.api.AsterixInputStream;
import org.apache.asterix.external.api.IInputStreamFactory;
import org.apache.asterix.external.api.INodeResolver;
import org.apache.asterix.external.api.INodeResolverFactory;
import org.apache.asterix.external.input.stream.LocalFSInputStream;
import org.apache.asterix.external.util.ExternalDataConstants;
import org.apache.asterix.external.util.ExternalDataUtils;
import org.apache.asterix.external.util.FileSystemWatcher;
import org.apache.asterix.external.util.NodeResolverFactory;
import org.apache.asterix.om.util.AsterixAppContextInfo;
import org.apache.hyracks.algebricks.common.constraints.AlgebricksAbsolutePartitionConstraint;
import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.io.FileReference;
import org.apache.hyracks.dataflow.std.file.FileSplit;

public class LocalFSInputStreamFactory implements IInputStreamFactory {

    private static final long serialVersionUID = 1L;

    protected static final INodeResolver DEFAULT_NODE_RESOLVER = new NodeResolverFactory().createNodeResolver();
    protected static final Logger LOGGER = Logger.getLogger(LocalFSInputStreamFactory.class.getName());
    protected static INodeResolver nodeResolver;
    protected Map<String, String> configuration;
    protected FileSplit[] inputFileSplits;
    protected boolean isFeed;
    protected String expression;
    // transient fields (They don't need to be serialized and transferred)
    private transient AlgebricksAbsolutePartitionConstraint constraints;
    private transient FileSystemWatcher watcher;

    @Override
    public synchronized AsterixInputStream createInputStream(IHyracksTaskContext ctx, int partition)
            throws HyracksDataException {
        if (watcher == null) {
            String nodeName = ctx.getJobletContext().getApplicationContext().getNodeId();
            ArrayList<Path> inputResources = new ArrayList<>();
            for (int i = 0; i < inputFileSplits.length; i++) {
                if (inputFileSplits[i].getNodeName().equals(nodeName)) {
                    inputResources.add(inputFileSplits[i].getLocalFile().getFile().toPath());
                }
            }
            watcher = new FileSystemWatcher(inputResources, expression, isFeed);
        }
        try {
            return new LocalFSInputStream(watcher);
        } catch (IOException e) {
            throw new HyracksDataException(e);
        }
    }

    @Override
    public DataSourceType getDataSourceType() {
        return DataSourceType.STREAM;
    }

    @Override
    public boolean isIndexible() {
        return false;
    }

    @Override
    public void configure(Map<String, String> configuration) throws AsterixException {
        this.configuration = configuration;
        String[] splits = configuration.get(ExternalDataConstants.KEY_PATH).split(",");
        configureFileSplits(splits);
        configurePartitionConstraint();
        this.isFeed = ExternalDataUtils.isFeed(configuration) && ExternalDataUtils.keepDataSourceOpen(configuration);
        this.expression = configuration.get(ExternalDataConstants.KEY_EXPRESSION);
    }

    @Override
    public AlgebricksAbsolutePartitionConstraint getPartitionConstraint() {
        return constraints;
    }

    private void configureFileSplits(String[] splits) throws AsterixException {
        INodeResolver resolver = getNodeResolver();
        if (inputFileSplits == null) {
            inputFileSplits = new FileSplit[splits.length];
            String nodeName;
            String nodeLocalPath;
            int count = 0;
            String trimmedValue;
            for (String splitPath : splits) {
                trimmedValue = splitPath.trim();
                if (!trimmedValue.contains("://")) {
                    throw new AsterixException(
                            "Invalid path: " + splitPath + "\nUsage- path=\"Host://Absolute File Path\"");
                }
                nodeName = resolver.resolveNode(trimmedValue.split(":")[0]);
                nodeLocalPath = trimmedValue.split("://")[1];
                FileSplit fileSplit = new FileSplit(nodeName, new FileReference(new File(nodeLocalPath)));
                inputFileSplits[count++] = fileSplit;
            }
        }
    }

    private void configurePartitionConstraint() throws AsterixException {
        Map<String, ClusterPartition[]> partitions = AsterixAppContextInfo.getInstance().getMetadataProperties()
                .getNodePartitions();
        List<String> locs = new ArrayList<>();
        for (int i = 0; i < inputFileSplits.length; i++) {
            String location = inputFileSplits[i].getNodeName();
            if (!locs.contains(location)) {
                int numOfPartitions = partitions.get(location).length;
                int j = 0;
                while (j < numOfPartitions) {
                    locs.add(location);
                    j++;
                }
            }
        }
        constraints = new AlgebricksAbsolutePartitionConstraint(locs.toArray(new String[locs.size()]));
    }

    protected INodeResolver getNodeResolver() {
        if (nodeResolver == null) {
            synchronized (DEFAULT_NODE_RESOLVER) {
                if (nodeResolver == null) {
                    nodeResolver = initializeNodeResolver();
                }
            }
        }
        return nodeResolver;
    }

    private static INodeResolver initializeNodeResolver() {
        INodeResolver nodeResolver = null;
        String configuredNodeResolverFactory = System.getProperty(ExternalDataConstants.NODE_RESOLVER_FACTORY_PROPERTY);
        if (configuredNodeResolverFactory != null) {
            try {
                nodeResolver = ((INodeResolverFactory) (Class.forName(configuredNodeResolverFactory).newInstance()))
                        .createNodeResolver();

            } catch (Exception e) {
                if (LOGGER.isLoggable(Level.WARNING)) {
                    LOGGER.log(Level.WARNING, "Unable to create node resolver from the configured classname "
                            + configuredNodeResolverFactory + "\n" + e.getMessage());
                }
                nodeResolver = DEFAULT_NODE_RESOLVER;
            }
        } else {
            nodeResolver = DEFAULT_NODE_RESOLVER;
        }
        return nodeResolver;
    }
}

@@ -981,19 +981,40 @@
 va.util.
-Lis
+Map;%0Aimport java.util.Se
 t;%0Aimpor
@@ -1017,35 +1017,39 @@
 mport java.util.
-Map
+TreeSet
 ;%0Aimport java.ut
@@ -1104,67 +1104,8 @@
 r;%0A%0A
-import org.apache.asterix.common.cluster.ClusterPartition;%0A
 impo
@@ -1710,65 +1710,8 @@
 ry;%0A
-import org.apache.asterix.om.util.AsterixAppContextInfo;%0A
 impo
@@ -5552,162 +5552,10 @@
     
-Map%3CString, ClusterPartition%5B%5D%3E partitions = AsterixAppContextInfo.getInstance().getMetadataProperties()%0A                .getNodePartitions();%0A        Lis
+Se
 t%3CSt
@@ -5567,32 +5567,30 @@
  locs = new 
-ArrayLis
+TreeSe
 t%3C%3E();%0A     
@@ -5723,280 +5723,27 @@
     
-if (!locs.contains(location)) %7B%0A                int numOfPartitions = partitions.get(location).length;%0A                int j = 0;%0A                while (j %3C numOfPartitions) %7B%0A                    locs.add(location);%0A                    j++;%0A                %7D%0A            %7D
+locs.add(location);
 %0A   

Fix Number of LocalFS Readers