<commit_before>/**
 * Copyright (C) 2006-2018 Talend Inc. - www.talend.com
 * <p>
 * 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
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * 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.talend.sdk.component.runtime.manager;

import static java.util.Arrays.asList;
import static java.util.Collections.emptyList;
import static java.util.Collections.emptyMap;
import static java.util.Collections.emptySet;
import static java.util.Optional.of;
import static java.util.Optional.ofNullable;
import static java.util.function.Function.identity;
import static java.util.stream.Collectors.joining;
import static java.util.stream.Collectors.toList;
import static java.util.stream.Collectors.toMap;
import static java.util.stream.Collectors.toSet;
import static org.apache.xbean.finder.archive.FileArchive.decode;
import static org.talend.sdk.component.runtime.base.lang.exception.InvocationExceptionWrapper.toRuntimeException;
import static org.talend.sdk.component.runtime.manager.reflect.Constructors.findConstructor;

import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectStreamException;
import java.io.Serializable;
import java.lang.annotation.Annotation;
import java.lang.management.ManagementFactory;
import java.lang.reflect.AnnotatedElement;
import java.lang.reflect.Constructor;
import java.lang.reflect.Executable;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.lang.reflect.Modifier;
import java.lang.reflect.Proxy;
import java.net.MalformedURLException;
import java.net.URL;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Properties;
import java.util.ServiceLoader;
import java.util.Set;
import java.util.Spliterator;
import java.util.Spliterators;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.jar.Attributes;
import java.util.jar.JarFile;
import java.util.jar.JarInputStream;
import java.util.jar.Manifest;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.json.JsonBuilderFactory;
import javax.json.JsonReaderFactory;
import javax.json.JsonWriterFactory;
import javax.json.bind.Jsonb;
import javax.json.bind.JsonbConfig;
import javax.json.bind.spi.JsonbProvider;
import javax.json.spi.JsonProvider;
import javax.json.stream.JsonGeneratorFactory;
import javax.json.stream.JsonParserFactory;
import javax.xml.parsers.DocumentBuilder;
import javax.xml.parsers.DocumentBuilderFactory;
import javax.xml.xpath.XPath;
import javax.xml.xpath.XPathFactory;

import org.apache.xbean.finder.AnnotationFinder;
import org.apache.xbean.finder.ClassFinder;
import org.apache.xbean.finder.archive.Archive;
import org.apache.xbean.finder.archive.FileArchive;
import org.apache.xbean.finder.archive.FilteredArchive;
import org.apache.xbean.finder.archive.JarArchive;
import org.apache.xbean.finder.filter.ExcludeIncludeFilter;
import org.apache.xbean.finder.filter.Filter;
import org.apache.xbean.finder.filter.Filters;
import org.slf4j.event.Level;
import org.talend.sdk.component.api.component.Components;
import org.talend.sdk.component.api.component.Icon;
import org.talend.sdk.component.api.component.MigrationHandler;
import org.talend.sdk.component.api.component.Version;
import org.talend.sdk.component.api.input.Emitter;
import org.talend.sdk.component.api.input.PartitionMapper;
import org.talend.sdk.component.api.internationalization.Internationalized;
import org.talend.sdk.component.api.processor.Processor;
import org.talend.sdk.component.api.service.ActionType;
import org.talend.sdk.component.api.service.Service;
import org.talend.sdk.component.api.service.cache.LocalCache;
import org.talend.sdk.component.api.service.configuration.LocalConfiguration;
import org.talend.sdk.component.api.service.dependency.Resolver;
import org.talend.sdk.component.api.service.http.HttpClient;
import org.talend.sdk.component.api.service.http.HttpClientFactory;
import org.talend.sdk.component.api.service.http.Request;
import org.talend.sdk.component.classloader.ConfigurableClassLoader;
import org.talend.sdk.component.container.Container;
import org.talend.sdk.component.container.ContainerListener;
import org.talend.sdk.component.container.ContainerManager;
import org.talend.sdk.component.dependencies.maven.MvnDependencyListLocalRepositoryResolver;
import org.talend.sdk.component.jmx.JmxManager;
import org.talend.sdk.component.runtime.base.Delegated;
import org.talend.sdk.component.runtime.input.LocalPartitionMapper;
import org.talend.sdk.component.runtime.input.Mapper;
import org.talend.sdk.component.runtime.input.PartitionMapperImpl;
import org.talend.sdk.component.runtime.internationalization.InternationalizationServiceFactory;
import org.talend.sdk.component.runtime.manager.asm.ProxyGenerator;
import org.talend.sdk.component.runtime.manager.extension.ComponentContextImpl;
import org.talend.sdk.component.runtime.manager.interceptor.InterceptorHandlerFacade;
import org.talend.sdk.component.runtime.manager.json.PreComputedJsonpProvider;
import org.talend.sdk.component.runtime.manager.proxy.JavaProxyEnricherFactory;
import org.talend.sdk.component.runtime.manager.reflect.ParameterModelService;
import org.talend.sdk.component.runtime.manager.reflect.ReflectionService;
import org.talend.sdk.component.runtime.manager.service.HttpClientFactoryImpl;
import org.talend.sdk.component.runtime.manager.service.LocalCacheService;
import org.talend.sdk.component.runtime.manager.service.LocalConfigurationService;
import org.talend.sdk.component.runtime.manager.service.ResolverImpl;
import org.talend.sdk.component.runtime.manager.spi.ContainerListenerExtension;
import org.talend.sdk.component.runtime.manager.xbean.FilterFactory;
import org.talend.sdk.component.runtime.manager.xbean.KnownClassesFilter;
import org.talend.sdk.component.runtime.manager.xbean.KnownJarsFilter;
import org.talend.sdk.component.runtime.manager.xbean.NestedJarArchive;
import org.talend.sdk.component.runtime.output.ProcessorImpl;
import org.talend.sdk.component.runtime.serialization.LightContainer;
import org.talend.sdk.component.runtime.visitor.ModelListener;
import org.talend.sdk.component.runtime.visitor.ModelVisitor;
import org.talend.sdk.component.spi.component.ComponentExtension;
import org.w3c.dom.Document;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.Getter;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class ComponentManager implements AutoCloseable {

    protected static final AtomicReference<ComponentManager> CONTEXTUAL_INSTANCE = new AtomicReference<>();

    private static final MigrationHandler NO_MIGRATION = (incomingVersion, incomingData) -> incomingData;

    private static final Components DEFAULT_COMPONENT = new Components() {

        @Override
        public Class<? extends Annotation> annotationType() {
            return Components.class;
        }

        @Override
        public String family() {
            return "";
        }

        @Override
        public String[] categories() {
            return new String[] { "Misc" };
        }
    };

    @Getter
    protected final ContainerManager container;

    // tcomp (org.talend + javax.annotation + jsonp) + logging (slf4j) are/can be provided
    // service
    // + tcomp "runtime" indeed (invisible from the components but required for the
    // runtime
    private final Filter classesFilter = Filters.prefixes(Stream
            .concat(Stream.of("org.talend.sdk.component.api.", "org.talend.sdk.component.spi.", "javax.annotation.",
                    "javax.json.", "org.talend.sdk.component.classloader.", "org.talend.sdk.component.runtime.",
                    "org.slf4j.", "org.apache.johnzon."), additionalContainerClasses())
            .toArray(String[]::new));

    private final Filter beamClassesFilter = FilterFactory.and(classesFilter,
            Filters.prefixes("org.apache.beam.runners.", "org.apache.beam.sdk.",
                    "org.talend.sdk.component.runtime.beam.", "org.codehaus.jackson.",
                    "com.fasterxml.jackson.annotation.", "com.fasterxml.jackson.core.",
                    "com.fasterxml.jackson.databind.", "com.thoughtwors.paranamer.", "org.apache.commons.compress.",
                    "org.tukaani.xz.", "org.objenesis.", "org.joda.time.", "org.xerial.snappy.", "avro.shaded.",
                    "org.apache.avro."));

    private final Filter excludeClassesFilter =
            Filters.prefixes("org.apache.beam.sdk.io.", "org.apache.beam.sdk.extensions.");

    private final ParameterModelService parameterModelService = new ParameterModelService();

    private final InternationalizationServiceFactory internationalizationServiceFactory =
            new InternationalizationServiceFactory();

    @Getter
    private final JsonProvider jsonpProvider;

    @Getter
    private final JsonGeneratorFactory jsonpGeneratorFactory;

    @Getter
    private final JsonReaderFactory jsonpReaderFactory;

    @Getter
    private final JsonBuilderFactory jsonpBuilderFactory;

    @Getter
    private final JsonParserFactory jsonpParserFactory;

    @Getter
    private final JsonWriterFactory jsonpWriterFactory;

    @Getter
    private final JsonbProvider jsonbProvider;

    private final ProxyGenerator proxyGenerator = new ProxyGenerator();

    private final JavaProxyEnricherFactory javaProxyEnricherFactory = new JavaProxyEnricherFactory();

    // kind of extracted to ensure we can switch it later if needed
    // (like using a Spring/CDI context and something else than xbean)
    private final ReflectionService reflections = new ReflectionService(parameterModelService);

    private final Collection<ComponentExtension> extensions;

    private final Level logInfoLevelMapping;

    /**
     * @param m2 the maven repository location if on the file system.
     * @param dependenciesResource the resource path containing dependencies.
     * @param jmxNamePattern a pattern to register the plugins (containers) in JMX, null
     * otherwise.
     */
    public ComponentManager(final File m2, final String dependenciesResource, final String jmxNamePattern) {
        final ClassLoader tccl = Thread.currentThread().getContextClassLoader();

        jsonpProvider = JsonProvider.provider();
        jsonbProvider = JsonbProvider.provider();
        // these factories have memory caches so ensure we reuse them properly
        jsonpGeneratorFactory = jsonpProvider.createGeneratorFactory(emptyMap());
        jsonpReaderFactory = jsonpProvider.createReaderFactory(emptyMap());
        jsonpBuilderFactory = jsonpProvider.createBuilderFactory(emptyMap());
        jsonpParserFactory = jsonpProvider.createParserFactory(emptyMap());
        jsonpWriterFactory = jsonpProvider.createWriterFactory(emptyMap());

        logInfoLevelMapping = findLogInfoLevel();

        final ContainerManager.ClassLoaderConfiguration defaultClassLoaderConfiguration =
                ContainerManager.ClassLoaderConfiguration
                        .builder()
                        .parent(tccl)
                        .parentClassesFilter(name -> isContainerClass(beamClassesFilter, name)) // beam is desired
                        .classesFilter(name -> !isContainerClass(classesFilter, name))
                        .supportsResourceDependencies(true)
                        .create();
        final ContainerManager.ClassLoaderConfiguration beamClassLoaderConfiguration =
                ContainerManager.ClassLoaderConfiguration
                        .builder()
                        .parent(tccl)
                        .parentClassesFilter(name -> isContainerClass(beamClassesFilter, name))
                        .classesFilter(name -> !isContainerClass(beamClassesFilter, name))
                        .supportsResourceDependencies(true)
                        .create();
        this.container = new ContainerManager(ContainerManager.DependenciesResolutionConfiguration
                .builder()
                .resolver(new MvnDependencyListLocalRepositoryResolver(dependenciesResource))
                .rootRepositoryLocation(m2)
                .create(), defaultClassLoaderConfiguration, container -> {
                    // if a beam component then ensure to use beam specific filtering
                    // since it becomes part of the container
                    if (container.getDependencies() != null && Stream.of(container.getDependencies()).anyMatch(
                            a -> a.getGroup().startsWith("org.apache.beam")
                                    || a.getArtifact().startsWith("beam-sdks-java-"))) {
                        container.set(ContainerManager.ClassLoaderConfiguration.class, beamClassLoaderConfiguration);
                    }
                }, logInfoLevelMapping);
        this.container.registerListener(new Updater());
        ofNullable(jmxNamePattern).map(String::trim).filter(n -> !n.isEmpty()).ifPresent(p -> this.container
                .registerListener(new JmxManager(container, p, ManagementFactory.getPlatformMBeanServer())));
        toStream(loadServiceProviders(ContainerListenerExtension.class, tccl)).forEach(container::registerListener);
        this.extensions = toStream(loadServiceProviders(ComponentExtension.class, tccl)).collect(toList());
    }

    private Level findLogInfoLevel() {
        if (Boolean.getBoolean("talend.component.manager.log.info")) {
            return Level.INFO;
        }
        try {
            ComponentManager.class.getClassLoader().loadClass("routines.TalendString");
            return Level.DEBUG;
        } catch (final NoClassDefFoundError | ClassNotFoundException e) {
            return Level.INFO;
        }
    }

    /**
     * Creates a default manager with default maven local repository,
     * TALEND-INF/dependencies.txt file to find the dependencies of the plugins and
     * a default JMX pattern for plugins. It also adds the caller as a plugin.
     *
     * @return the contextual manager instance.
     */
    public static ComponentManager instance() {
        ComponentManager manager = CONTEXTUAL_INSTANCE.get();
        if (manager == null) {
            synchronized (CONTEXTUAL_INSTANCE) {
                if (CONTEXTUAL_INSTANCE.get() == null) {
                    final Thread shutdownHook =
                            new Thread(ComponentManager.class.getName() + "-" + ComponentManager.class.hashCode()) {

                                @Override
                                public void run() {
                                    CONTEXTUAL_INSTANCE.get().close();
                                }
                            };

                    manager = new ComponentManager(findM2(), "TALEND-INF/dependencies.txt",
                            "org.talend.sdk.component:type=component,value=%s") {

                        private final AtomicBoolean closed = new AtomicBoolean(false);

                        {

                            info("Creating the contextual ComponentManager instance " + getIdentifiers());
                            if (!Boolean.getBoolean("component.manager.callers.skip")) {
                                addCallerAsPlugin();
                            }

                            // common for studio until job generation is updated to build a tcomp friendly
                            // bundle
                            // alternatively we could capture based on TALEND-INF/dependencies.txt jars
                            if (!Boolean.getBoolean("component.manager.classpath.skip")) {
                                final String componentClasspath = findClasspath().replace(File.pathSeparatorChar, ':');
                                if (!componentClasspath.isEmpty()) {
                                    final String[] jars = componentClasspath.split(":");
                                    if (jars.length > 1) {
                                        Stream
                                                .of(jars)
                                                .map(FileArchive::decode)
                                                .map(File::new)
                                                .filter(File::exists)
                                                .filter(f -> !f.isDirectory() && f.getName().endsWith(".jar"))
                                                .filter(f -> KnownJarsFilter.INSTANCE.test(f.getName()))
                                                .filter(f -> !hasPlugin(container.buildAutoIdFromName(f.getName())))
                                                .forEach(jar -> addPlugin(jar.getAbsolutePath()));
                                    }
                                }
                            }

                            container.getDefinedNestedPlugin().stream().filter(p -> !hasPlugin(p)).forEach(
                                    this::addPlugin);
                            info("Components: " + availablePlugins());
                        }

                        @Override
                        public void close() {
                            if (!closed.compareAndSet(false, true)) {
                                return;
                            }
                            try {
                                synchronized (CONTEXTUAL_INSTANCE) {
                                    if (CONTEXTUAL_INSTANCE.compareAndSet(this, null)) {
                                        try {
                                            Runtime.getRuntime().removeShutdownHook(shutdownHook);
                                        } catch (final IllegalStateException ise) {
                                            // already shutting down
                                        }
                                    }
                                }
                            } finally {
                                super.close();
                                info("Released the contextual ComponentManager instance " + getIdentifiers());
                            }
                        }

                        Object readResolve() throws ObjectStreamException {
                            return new SerializationReplacer();
                        }
                    };

                    Runtime.getRuntime().addShutdownHook(shutdownHook);
                    manager.info("Created the contextual ComponentManager instance " + getIdentifiers());
                    if (!CONTEXTUAL_INSTANCE.compareAndSet(null, manager)) { // unlikely it fails in a synch block
                        manager = CONTEXTUAL_INSTANCE.get();
                    }
                }
            }
        }

        return manager;
    }

    protected void info(final String msg) {
        switch (logInfoLevelMapping) {
        case DEBUG:
            log.debug(msg);
            break;
        case INFO:
        default:
            log.info(msg);
        }
    }

    private static String findClasspath() { // alternative is to use getResources("") and parse urls
        return ofNullable(System.getProperty("component.manager.classpath")).orElseGet(() -> Stream
                .of(System.getProperty("java.class.path"), findManifestClassPath())
                .filter(Objects::nonNull)
                .map(s -> s.replace(File.separatorChar, '/'))
                .collect(joining(File.pathSeparator)));
    }

    // studio specific (launching from the interface)
    private static String findManifestClassPath() {
        return ofNullable(System.getProperty("java.class.path"))
                .flatMap(cp -> Stream
                        .of(cp.split(File.pathSeparator))
                        .map(File::new)
                        .filter(f -> f.getName().equals("classpath.jar"))
                        .findFirst())
                .map(f -> {
                    // respect this format sun.misc.URLClassPath.JarLoader.parseClassPath()
                    try (final JarFile file = new JarFile(f)) {
                        return ofNullable(file.getManifest())
                                .map(Manifest::getMainAttributes)
                                .map(a -> a.getValue(Attributes.Name.CLASS_PATH))
                                .map(value -> value.replace(' ', ':'))
                                .orElse(null);
                    } catch (final IOException e) {
                        log.warn(e.getMessage());
                        log.debug(e.getMessage(), e);
                        return null;
                    }
                })
                .map(String::trim)
                .orElse(null);
    }

    private static Stream<String> additionalContainerClasses() {<commit_msg>using windows separator since linux one is used in windows paths<commit_after>/**
 * Copyright (C) 2006-2018 Talend Inc. - www.talend.com
 * <p>
 * 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
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * 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.talend.sdk.component.runtime.manager;

import static java.util.Arrays.asList;
import static java.util.Collections.emptyList;
import static java.util.Collections.emptyMap;
import static java.util.Collections.emptySet;
import static java.util.Optional.of;
import static java.util.Optional.ofNullable;
import static java.util.function.Function.identity;
import static java.util.stream.Collectors.joining;
import static java.util.stream.Collectors.toList;
import static java.util.stream.Collectors.toMap;
import static java.util.stream.Collectors.toSet;
import static org.apache.xbean.finder.archive.FileArchive.decode;
import static org.talend.sdk.component.runtime.base.lang.exception.InvocationExceptionWrapper.toRuntimeException;
import static org.talend.sdk.component.runtime.manager.reflect.Constructors.findConstructor;

import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectStreamException;
import java.io.Serializable;
import java.lang.annotation.Annotation;
import java.lang.management.ManagementFactory;
import java.lang.reflect.AnnotatedElement;
import java.lang.reflect.Constructor;
import java.lang.reflect.Executable;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.lang.reflect.Modifier;
import java.lang.reflect.Proxy;
import java.net.MalformedURLException;
import java.net.URL;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Properties;
import java.util.ServiceLoader;
import java.util.Set;
import java.util.Spliterator;
import java.util.Spliterators;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.jar.Attributes;
import java.util.jar.JarFile;
import java.util.jar.JarInputStream;
import java.util.jar.Manifest;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.json.JsonBuilderFactory;
import javax.json.JsonReaderFactory;
import javax.json.JsonWriterFactory;
import javax.json.bind.Jsonb;
import javax.json.bind.JsonbConfig;
import javax.json.bind.spi.JsonbProvider;
import javax.json.spi.JsonProvider;
import javax.json.stream.JsonGeneratorFactory;
import javax.json.stream.JsonParserFactory;
import javax.xml.parsers.DocumentBuilder;
import javax.xml.parsers.DocumentBuilderFactory;
import javax.xml.xpath.XPath;
import javax.xml.xpath.XPathFactory;

import org.apache.xbean.finder.AnnotationFinder;
import org.apache.xbean.finder.ClassFinder;
import org.apache.xbean.finder.archive.Archive;
import org.apache.xbean.finder.archive.FileArchive;
import org.apache.xbean.finder.archive.FilteredArchive;
import org.apache.xbean.finder.archive.JarArchive;
import org.apache.xbean.finder.filter.ExcludeIncludeFilter;
import org.apache.xbean.finder.filter.Filter;
import org.apache.xbean.finder.filter.Filters;
import org.slf4j.event.Level;
import org.talend.sdk.component.api.component.Components;
import org.talend.sdk.component.api.component.Icon;
import org.talend.sdk.component.api.component.MigrationHandler;
import org.talend.sdk.component.api.component.Version;
import org.talend.sdk.component.api.input.Emitter;
import org.talend.sdk.component.api.input.PartitionMapper;
import org.talend.sdk.component.api.internationalization.Internationalized;
import org.talend.sdk.component.api.processor.Processor;
import org.talend.sdk.component.api.service.ActionType;
import org.talend.sdk.component.api.service.Service;
import org.talend.sdk.component.api.service.cache.LocalCache;
import org.talend.sdk.component.api.service.configuration.LocalConfiguration;
import org.talend.sdk.component.api.service.dependency.Resolver;
import org.talend.sdk.component.api.service.http.HttpClient;
import org.talend.sdk.component.api.service.http.HttpClientFactory;
import org.talend.sdk.component.api.service.http.Request;
import org.talend.sdk.component.classloader.ConfigurableClassLoader;
import org.talend.sdk.component.container.Container;
import org.talend.sdk.component.container.ContainerListener;
import org.talend.sdk.component.container.ContainerManager;
import org.talend.sdk.component.dependencies.maven.MvnDependencyListLocalRepositoryResolver;
import org.talend.sdk.component.jmx.JmxManager;
import org.talend.sdk.component.runtime.base.Delegated;
import org.talend.sdk.component.runtime.input.LocalPartitionMapper;
import org.talend.sdk.component.runtime.input.Mapper;
import org.talend.sdk.component.runtime.input.PartitionMapperImpl;
import org.talend.sdk.component.runtime.internationalization.InternationalizationServiceFactory;
import org.talend.sdk.component.runtime.manager.asm.ProxyGenerator;
import org.talend.sdk.component.runtime.manager.extension.ComponentContextImpl;
import org.talend.sdk.component.runtime.manager.interceptor.InterceptorHandlerFacade;
import org.talend.sdk.component.runtime.manager.json.PreComputedJsonpProvider;
import org.talend.sdk.component.runtime.manager.proxy.JavaProxyEnricherFactory;
import org.talend.sdk.component.runtime.manager.reflect.ParameterModelService;
import org.talend.sdk.component.runtime.manager.reflect.ReflectionService;
import org.talend.sdk.component.runtime.manager.service.HttpClientFactoryImpl;
import org.talend.sdk.component.runtime.manager.service.LocalCacheService;
import org.talend.sdk.component.runtime.manager.service.LocalConfigurationService;
import org.talend.sdk.component.runtime.manager.service.ResolverImpl;
import org.talend.sdk.component.runtime.manager.spi.ContainerListenerExtension;
import org.talend.sdk.component.runtime.manager.xbean.FilterFactory;
import org.talend.sdk.component.runtime.manager.xbean.KnownClassesFilter;
import org.talend.sdk.component.runtime.manager.xbean.KnownJarsFilter;
import org.talend.sdk.component.runtime.manager.xbean.NestedJarArchive;
import org.talend.sdk.component.runtime.output.ProcessorImpl;
import org.talend.sdk.component.runtime.serialization.LightContainer;
import org.talend.sdk.component.runtime.visitor.ModelListener;
import org.talend.sdk.component.runtime.visitor.ModelVisitor;
import org.talend.sdk.component.spi.component.ComponentExtension;
import org.w3c.dom.Document;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.Getter;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class ComponentManager implements AutoCloseable {

    protected static final AtomicReference<ComponentManager> CONTEXTUAL_INSTANCE = new AtomicReference<>();

    private static final MigrationHandler NO_MIGRATION = (incomingVersion, incomingData) -> incomingData;

    private static final Components DEFAULT_COMPONENT = new Components() {

        @Override
        public Class<? extends Annotation> annotationType() {
            return Components.class;
        }

        @Override
        public String family() {
            return "";
        }

        @Override
        public String[] categories() {
            return new String[] { "Misc" };
        }
    };

    @Getter
    protected final ContainerManager container;

    // tcomp (org.talend + javax.annotation + jsonp) + logging (slf4j) are/can be provided
    // service
    // + tcomp "runtime" indeed (invisible from the components but required for the
    // runtime
    private final Filter classesFilter = Filters.prefixes(Stream
            .concat(Stream.of("org.talend.sdk.component.api.", "org.talend.sdk.component.spi.", "javax.annotation.",
                    "javax.json.", "org.talend.sdk.component.classloader.", "org.talend.sdk.component.runtime.",
                    "org.slf4j.", "org.apache.johnzon."), additionalContainerClasses())
            .toArray(String[]::new));

    private final Filter beamClassesFilter = FilterFactory.and(classesFilter,
            Filters.prefixes("org.apache.beam.runners.", "org.apache.beam.sdk.",
                    "org.talend.sdk.component.runtime.beam.", "org.codehaus.jackson.",
                    "com.fasterxml.jackson.annotation.", "com.fasterxml.jackson.core.",
                    "com.fasterxml.jackson.databind.", "com.thoughtwors.paranamer.", "org.apache.commons.compress.",
                    "org.tukaani.xz.", "org.objenesis.", "org.joda.time.", "org.xerial.snappy.", "avro.shaded.",
                    "org.apache.avro."));

    private final Filter excludeClassesFilter =
            Filters.prefixes("org.apache.beam.sdk.io.", "org.apache.beam.sdk.extensions.");

    private final ParameterModelService parameterModelService = new ParameterModelService();

    private final InternationalizationServiceFactory internationalizationServiceFactory =
            new InternationalizationServiceFactory();

    @Getter
    private final JsonProvider jsonpProvider;

    @Getter
    private final JsonGeneratorFactory jsonpGeneratorFactory;

    @Getter
    private final JsonReaderFactory jsonpReaderFactory;

    @Getter
    private final JsonBuilderFactory jsonpBuilderFactory;

    @Getter
    private final JsonParserFactory jsonpParserFactory;

    @Getter
    private final JsonWriterFactory jsonpWriterFactory;

    @Getter
    private final JsonbProvider jsonbProvider;

    private final ProxyGenerator proxyGenerator = new ProxyGenerator();

    private final JavaProxyEnricherFactory javaProxyEnricherFactory = new JavaProxyEnricherFactory();

    // kind of extracted to ensure we can switch it later if needed
    // (like using a Spring/CDI context and something else than xbean)
    private final ReflectionService reflections = new ReflectionService(parameterModelService);

    private final Collection<ComponentExtension> extensions;

    private final Level logInfoLevelMapping;

    /**
     * @param m2 the maven repository location if on the file system.
     * @param dependenciesResource the resource path containing dependencies.
     * @param jmxNamePattern a pattern to register the plugins (containers) in JMX, null
     * otherwise.
     */
    public ComponentManager(final File m2, final String dependenciesResource, final String jmxNamePattern) {
        final ClassLoader tccl = Thread.currentThread().getContextClassLoader();

        jsonpProvider = JsonProvider.provider();
        jsonbProvider = JsonbProvider.provider();
        // these factories have memory caches so ensure we reuse them properly
        jsonpGeneratorFactory = jsonpProvider.createGeneratorFactory(emptyMap());
        jsonpReaderFactory = jsonpProvider.createReaderFactory(emptyMap());
        jsonpBuilderFactory = jsonpProvider.createBuilderFactory(emptyMap());
        jsonpParserFactory = jsonpProvider.createParserFactory(emptyMap());
        jsonpWriterFactory = jsonpProvider.createWriterFactory(emptyMap());

        logInfoLevelMapping = findLogInfoLevel();

        final ContainerManager.ClassLoaderConfiguration defaultClassLoaderConfiguration =
                ContainerManager.ClassLoaderConfiguration
                        .builder()
                        .parent(tccl)
                        .parentClassesFilter(name -> isContainerClass(beamClassesFilter, name)) // beam is desired
                        .classesFilter(name -> !isContainerClass(classesFilter, name))
                        .supportsResourceDependencies(true)
                        .create();
        final ContainerManager.ClassLoaderConfiguration beamClassLoaderConfiguration =
                ContainerManager.ClassLoaderConfiguration
                        .builder()
                        .parent(tccl)
                        .parentClassesFilter(name -> isContainerClass(beamClassesFilter, name))
                        .classesFilter(name -> !isContainerClass(beamClassesFilter, name))
                        .supportsResourceDependencies(true)
                        .create();
        this.container = new ContainerManager(ContainerManager.DependenciesResolutionConfiguration
                .builder()
                .resolver(new MvnDependencyListLocalRepositoryResolver(dependenciesResource))
                .rootRepositoryLocation(m2)
                .create(), defaultClassLoaderConfiguration, container -> {
                    // if a beam component then ensure to use beam specific filtering
                    // since it becomes part of the container
                    if (container.getDependencies() != null && Stream.of(container.getDependencies()).anyMatch(
                            a -> a.getGroup().startsWith("org.apache.beam")
                                    || a.getArtifact().startsWith("beam-sdks-java-"))) {
                        container.set(ContainerManager.ClassLoaderConfiguration.class, beamClassLoaderConfiguration);
                    }
                }, logInfoLevelMapping);
        this.container.registerListener(new Updater());
        ofNullable(jmxNamePattern).map(String::trim).filter(n -> !n.isEmpty()).ifPresent(p -> this.container
                .registerListener(new JmxManager(container, p, ManagementFactory.getPlatformMBeanServer())));
        toStream(loadServiceProviders(ContainerListenerExtension.class, tccl)).forEach(container::registerListener);
        this.extensions = toStream(loadServiceProviders(ComponentExtension.class, tccl)).collect(toList());
    }

    private Level findLogInfoLevel() {
        if (Boolean.getBoolean("talend.component.manager.log.info")) {
            return Level.INFO;
        }
        try {
            ComponentManager.class.getClassLoader().loadClass("routines.TalendString");
            return Level.DEBUG;
        } catch (final NoClassDefFoundError | ClassNotFoundException e) {
            return Level.INFO;
        }
    }

    /**
     * Creates a default manager with default maven local repository,
     * TALEND-INF/dependencies.txt file to find the dependencies of the plugins and
     * a default JMX pattern for plugins. It also adds the caller as a plugin.
     *
     * @return the contextual manager instance.
     */
    public static ComponentManager instance() {
        ComponentManager manager = CONTEXTUAL_INSTANCE.get();
        if (manager == null) {
            synchronized (CONTEXTUAL_INSTANCE) {
                if (CONTEXTUAL_INSTANCE.get() == null) {
                    final Thread shutdownHook =
                            new Thread(ComponentManager.class.getName() + "-" + ComponentManager.class.hashCode()) {

                                @Override
                                public void run() {
                                    CONTEXTUAL_INSTANCE.get().close();
                                }
                            };

                    manager = new ComponentManager(findM2(), "TALEND-INF/dependencies.txt",
                            "org.talend.sdk.component:type=component,value=%s") {

                        private final AtomicBoolean closed = new AtomicBoolean(false);

                        {

                            info("Creating the contextual ComponentManager instance " + getIdentifiers());
                            if (!Boolean.getBoolean("component.manager.callers.skip")) {
                                addCallerAsPlugin();
                            }

                            // common for studio until job generation is updated to build a tcomp friendly
                            // bundle
                            // alternatively we could capture based on TALEND-INF/dependencies.txt jars
                            if (!Boolean.getBoolean("component.manager.classpath.skip")) {
                                final String componentClasspath = findClasspath().replace(File.pathSeparatorChar, ';');
                                if (!componentClasspath.isEmpty()) {
                                    final String[] jars = componentClasspath.split(";");
                                    if (jars.length > 1) {
                                        Stream
                                                .of(jars)
                                                .map(FileArchive::decode)
                                                .map(File::new)
                                                .filter(File::exists)
                                                .filter(f -> !f.isDirectory() && f.getName().endsWith(".jar"))
                                                .filter(f -> KnownJarsFilter.INSTANCE.test(f.getName()))
                                                .filter(f -> !hasPlugin(container.buildAutoIdFromName(f.getName())))
                                                .forEach(jar -> addPlugin(jar.getAbsolutePath()));
                                    }
                                }
                            }

                            container.getDefinedNestedPlugin().stream().filter(p -> !hasPlugin(p)).forEach(
                                    this::addPlugin);
                            info("Components: " + availablePlugins());
                        }

                        @Override
                        public void close() {
                            if (!closed.compareAndSet(false, true)) {
                                return;
                            }
                            try {
                                synchronized (CONTEXTUAL_INSTANCE) {
                                    if (CONTEXTUAL_INSTANCE.compareAndSet(this, null)) {
                                        try {
                                            Runtime.getRuntime().removeShutdownHook(shutdownHook);
                                        } catch (final IllegalStateException ise) {
                                            // already shutting down
                                        }
                                    }
                                }
                            } finally {
                                super.close();
                                info("Released the contextual ComponentManager instance " + getIdentifiers());
                            }
                        }

                        Object readResolve() throws ObjectStreamException {
                            return new SerializationReplacer();
                        }
                    };

                    Runtime.getRuntime().addShutdownHook(shutdownHook);
                    manager.info("Created the contextual ComponentManager instance " + getIdentifiers());
                    if (!CONTEXTUAL_INSTANCE.compareAndSet(null, manager)) { // unlikely it fails in a synch block
                        manager = CONTEXTUAL_INSTANCE.get();
                    }
                }
            }
        }

        return manager;
    }

    protected void info(final String msg) {
        switch (logInfoLevelMapping) {
        case DEBUG:
            log.debug(msg);
            break;
        case INFO:
        default:
            log.info(msg);
        }
    }

    private static String findClasspath() { // alternative is to use getResources("") and parse urls
        return ofNullable(System.getProperty("component.manager.classpath")).orElseGet(() -> Stream
                .of(System.getProperty("java.class.path"), findManifestClassPath())
                .filter(Objects::nonNull)
                .map(s -> s.replace(File.separatorChar, '/'))
                .collect(joining(File.pathSeparator)));
    }

    // studio specific (launching from the interface)
    private static String findManifestClassPath() {
        return ofNullable(System.getProperty("java.class.path"))
                .flatMap(cp -> Stream
                        .of(cp.split(File.pathSeparator))
                        .map(File::new)
                        .filter(f -> f.getName().equals("classpath.jar"))
                        .findFirst())
                .map(f -> {
                    // respect this format sun.misc.URLClassPath.JarLoader.parseClassPath()
                    try (final JarFile file = new JarFile(f)) {
                        return ofNullable(file.getManifest())
                                .map(Manifest::getMainAttributes)
                                .map(a -> a.getValue(Attributes.Name.CLASS_PATH))
                                .map(value -> value.replace(' ', ';'))
                                .orElse(null);
                    } catch (final IOException e) {
                        log.warn(e.getMessage());
                        log.debug(e.getMessage(), e);
                        return null;
                    }
                })
                .map(String::trim)
                .orElse(null);
    }

    private static Stream<String> additionalContainerClasses() {
{'fraction_non_alphanumeric': 0.09448564234065655, 'fraction_numerical': 0.0011088779541201747, 'mean_word_length': 2.3148020522245196, 'pattern_counts': {'":': 1, '<': 19, '<?xml version=': 0, '>': 63, 'https://': 0, 'lorem ipsum': 0, 'www.': 4, 'xml': 14}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 68, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5457226', 'n_tokens_mistral': 10733, 'n_tokens_neox': 10374, 'n_words': 2096}