/*
 * 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 com.facebook.presto.metadata;

import com.facebook.airlift.log.Logger;
import com.google.common.base.Splitter;
import com.google.common.collect.ImmutableList;

import javax.inject.Inject;

import java.io.File;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;

import static com.facebook.presto.util.PropertiesUtil.loadProperties;
import static com.google.common.base.Preconditions.checkState;
import static com.google.common.io.Files.getNameWithoutExtension;

public class StaticFunctionNamespaceStore
{
    private static final Logger log = Logger.get(StaticFunctionNamespaceStore.class);
    private static final Splitter SPLITTER = Splitter.on(',').trimResults().omitEmptyStrings();

    private final FunctionAndTypeManager functionAndTypeManager;
    private final File configDir;
    private final AtomicBoolean functionNamespaceLoading = new AtomicBoolean();

    @Inject
    public StaticFunctionNamespaceStore(FunctionAndTypeManager functionAndTypeManager, StaticFunctionNamespaceStoreConfig config)
    {
        this.functionAndTypeManager = functionAndTypeManager;
        this.configDir = config.getFunctionNamespaceConfigurationDir();
    }

    public void loadFunctionNamespaceManagers()
            throws Exception
    {
        if (!functionNamespaceLoading.compareAndSet(false, true)) {
            return;
        }

        for (File file : listFiles(configDir)) {
            if (file.isFile() && file.getName().endsWith(".properties")) {
                loadFunctionNamespaceManager(file);
            }
        }
    }

    private void loadFunctionNamespaceManager(File file)
            throws Exception
    {
        String catalogName = getNameWithoutExtension(file.getName());
        log.info("-- Loading function namespace manager from %s --", file);
        Map<String, String> properties = new HashMap<>(loadProperties(file));

        String functionNamespaceManagerName = properties.remove("function-namespace-manager.name");
        checkState(functionNamespaceManagerName != null, "Function namespace configuration %s does not contain function-namespace-manager.name", file.getAbsoluteFile());

        functionAndTypeManager.loadFunctionNamespaceManager(functionNamespaceManagerName, catalogName, properties);
        log.info("-- Added function namespace manager [%s] --", catalogName);
    }

    private static List<File> listFiles(File dir)
    {
        if (dir != null && dir.isDirectory()) {
            File[] files = dir.listFiles();
            if (files != null) {
                return ImmutableList.copyOf(files);
            }
        }
        return ImmutableList.of();
    }
}

@@ -638,48 +638,8 @@
 er;%0A
-import com.google.common.base.Splitter;%0A
 impo
@@ -988,16 +988,76 @@
 kState;%0A
+import static com.google.common.base.Strings.isNullOrEmpty;%0A
 import s
@@ -1275,76 +1275,81 @@
 al S
-plitter SPLITTER = Splitter.on(',').trimResults().omitEmptyStrings()
+tring FUNCTION_NAMESPACE_MANAGER_NAME = %22function-namespace-manager.name%22
 ;%0A%0A 
@@ -2118,24 +2118,454 @@
 erties%22)) %7B%0A
+                String catalogName = getNameWithoutExtension(file.getName());%0A                Map%3CString, String%3E properties = loadProperties(file);%0A                checkState(!isNullOrEmpty(properties.get(FUNCTION_NAMESPACE_MANAGER_NAME)),%0A                        %22Function namespace configuration %25s does not contain %25s%22,%0A                        file.getAbsoluteFile(),%0A                        FUNCTION_NAMESPACE_MANAGER_NAME);%0A
             
@@ -2597,20 +2597,39 @@
 Manager(
-file
+catalogName, properties
 );%0A     
@@ -2659,22 +2659,21 @@
 %7D%0A%0A    p
-rivate
+ublic
  void lo
@@ -2702,124 +2702,323 @@
 ager
-(File file)%0A            throws Exception%0A    %7B%0A        String catalogName = getNameWithoutExtension(file.getName());
+s(Map%3CString, Map%3CString, String%3E%3E catalogProperties)%0A    %7B%0A        catalogProperties.entrySet().stream()%0A                .forEach(entry -%3E loadFunctionNamespaceManager(entry.getKey(), entry.getValue()));%0A    %7D%0A%0A    private void loadFunctionNamespaceManager(String catalogName, Map%3CString, String%3E properties)%0A    %7B
 %0A   
@@ -3075,54 +3075,48 @@
 er f
-rom %25s --%22, file);%0A        Map%3CString, String%3E
+or catalog %25s --%22, catalogName);%0A       
  pro
@@ -3139,21 +3139,17 @@
 shMap%3C%3E(
-loadP
+p
 ropertie
@@ -3153,18 +3153,11 @@
 ties
-(file)
 );%0A
-%0A
     
@@ -3220,41 +3220,39 @@
 ove(
-%22function-namespace-manager.name%22
+FUNCTION_NAMESPACE_MANAGER_NAME
 );%0A 
@@ -3269,16 +3269,31 @@
 ckState(
+!isNullOrEmpty(
 function
@@ -3316,128 +3316,73 @@
 Name
- != null, %22Function namespace configuration %25s does not contain function-namespace-manager.name%22, file.getAbsoluteFile()
+), %22%25s property must be present%22, FUNCTION_NAMESPACE_MANAGER_NAME
 );%0A%0A

Refactor StaticFunctionNamespaceStore to load properties map