SharedServerResources.java

/*
 * 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.tika.pipes.core.server;

import java.io.Closeable;
import java.io.IOException;

import org.xml.sax.SAXException;

import org.apache.tika.config.ExceptionReporting;
import org.apache.tika.config.loader.PresetRegistry;
import org.apache.tika.config.loader.TikaJsonConfig;
import org.apache.tika.config.loader.TikaLoader;
import org.apache.tika.detect.Detector;
import org.apache.tika.exception.TikaConfigException;
import org.apache.tika.exception.TikaException;
import org.apache.tika.metadata.filter.MetadataFilter;
import org.apache.tika.metadata.writelimiter.MetadataWriteLimiterFactory;
import org.apache.tika.parser.AutoDetectParser;
import org.apache.tika.parser.ParseContext;
import org.apache.tika.parser.RecursiveParserWrapper;
import org.apache.tika.parser.inference.EngineRegistry;
import org.apache.tika.parser.inference.InferenceDispatcher;
import org.apache.tika.pipes.core.EmitStrategy;
import org.apache.tika.pipes.core.PipesConfig;
import org.apache.tika.pipes.core.config.ConfigStore;
import org.apache.tika.pipes.core.config.ConfigStoreFactory;
import org.apache.tika.pipes.core.emitter.EmitterManager;
import org.apache.tika.pipes.core.fetcher.FetcherManager;
import org.apache.tika.plugins.ExtensionConfig;
import org.apache.tika.plugins.TikaPluginManager;
import org.apache.tika.sax.ContentHandlerFactory;

/**
 * Holds shared resources for a shared PipesServer.
 * <p>
 * These resources are thread-safe and shared among all ConnectionHandlers:
 * <ul>
 *   <li>AutoDetectParser, Detector - immutable after initialization</li>
 *   <li>FetcherManager, EmitterManager - use ConcurrentHashMap internally</li>
 *   <li>MetadataFilter, ContentHandlerFactory - stateless</li>
 * </ul>
 */
public class SharedServerResources implements Closeable {

    private final TikaLoader tikaLoader;
    private final PipesConfig pipesConfig;
    private final AutoDetectParser autoDetectParser;
    private final Detector detector;
    private final RecursiveParserWrapper rMetaParser;
    private final FetcherManager fetcherManager;
    private final EmitterManager emitterManager;
    private final MetadataFilter defaultMetadataFilter;
    private final InferenceDispatcher inferenceDispatcher;
    private final EngineRegistry engineRegistry;
    private final ContentHandlerFactory defaultContentHandlerFactory;
    private final MetadataWriteLimiterFactory defaultMetadataWriteLimiterFactory;
    private final EmitStrategy emitStrategy;
    private final ConfigStore configStore;
    private final ExceptionReporting exceptionReporting;
    private final PresetRegistry presetRegistry;

    private SharedServerResources(TikaLoader tikaLoader, PipesConfig pipesConfig,
                                  AutoDetectParser autoDetectParser, Detector detector,
                                  RecursiveParserWrapper rMetaParser, FetcherManager fetcherManager,
                                  EmitterManager emitterManager, MetadataFilter defaultMetadataFilter,
                                  InferenceDispatcher inferenceDispatcher, EngineRegistry engineRegistry,
                                  ContentHandlerFactory defaultContentHandlerFactory,
                                  MetadataWriteLimiterFactory defaultMetadataWriteLimiterFactory,
                                  EmitStrategy emitStrategy, ConfigStore configStore,
                                  ExceptionReporting exceptionReporting,
                                  PresetRegistry presetRegistry) {
        this.tikaLoader = tikaLoader;
        this.pipesConfig = pipesConfig;
        this.autoDetectParser = autoDetectParser;
        this.detector = detector;
        this.rMetaParser = rMetaParser;
        this.fetcherManager = fetcherManager;
        this.emitterManager = emitterManager;
        this.defaultMetadataFilter = defaultMetadataFilter;
        this.inferenceDispatcher = inferenceDispatcher;
        this.engineRegistry = engineRegistry;
        this.defaultContentHandlerFactory = defaultContentHandlerFactory;
        this.defaultMetadataWriteLimiterFactory = defaultMetadataWriteLimiterFactory;
        this.emitStrategy = emitStrategy;
        this.configStore = configStore;
        this.exceptionReporting = exceptionReporting;
        this.presetRegistry = presetRegistry;
    }

    /**
     * Loads shared server resources from configuration.
     *
     * @param tikaLoader the tika loader
     * @param pipesConfig the pipes configuration
     * @return initialized shared resources
     * @throws TikaException if initialization fails
     * @throws IOException if I/O error occurs
     * @throws SAXException if XML parsing error occurs
     */
    public static SharedServerResources load(TikaLoader tikaLoader, PipesConfig pipesConfig)
            throws TikaException, IOException, SAXException {

        TikaJsonConfig tikaJsonConfig = tikaLoader.getConfig();
        TikaPluginManager tikaPluginManager = TikaPluginManager.load(tikaJsonConfig);

        // Create ConfigStore if specified
        ConfigStore configStore = createConfigStore(pipesConfig, tikaPluginManager);

        // Load managers
        FetcherManager fetcherManager = FetcherManager.load(tikaPluginManager, tikaJsonConfig, true, configStore);
        EmitterManager emitterManager = EmitterManager.load(tikaPluginManager, tikaJsonConfig, true, configStore);

        // Load parser and detector
        AutoDetectParser autoDetectParser = (AutoDetectParser) tikaLoader.loadAutoDetectParser();
        Detector detector = autoDetectParser.getDetector();
        RecursiveParserWrapper rMetaParser = new RecursiveParserWrapper(autoDetectParser);

        // Load filters and factories
        MetadataFilter metadataFilter = tikaLoader.loadMetadataFilters();
        EngineRegistry engineRegistry = tikaLoader.get(EngineRegistry.class);
        InferenceDispatcher inferenceDispatcher = tikaLoader.get(InferenceDispatcher.class);
        ContentHandlerFactory contentHandlerFactory = tikaLoader.loadContentHandlerFactory();
        ParseContext configContext = tikaLoader.loadParseContext();
        MetadataWriteLimiterFactory metadataWriteLimiterFactory =
                configContext.get(MetadataWriteLimiterFactory.class);

        EmitStrategy emitStrategy = pipesConfig.getEmitStrategy().getType();

        // fails startup on an unresolvable preset, mirroring the front-end's own load
        PresetRegistry presetRegistry =
                PresetRegistry.load(tikaJsonConfig, tikaLoader.getClassLoader());

        return new SharedServerResources(tikaLoader, pipesConfig, autoDetectParser, detector,
                rMetaParser, fetcherManager, emitterManager, metadataFilter, inferenceDispatcher,
                engineRegistry, contentHandlerFactory,
                metadataWriteLimiterFactory, emitStrategy, configStore,
                ExceptionReporting.get(configContext), presetRegistry);
    }

    private static ConfigStore createConfigStore(PipesConfig pipesConfig, TikaPluginManager tikaPluginManager)
            throws TikaException {
        String configStoreType = pipesConfig.getConfigStoreType();
        String configStoreParams = pipesConfig.getConfigStoreParams();

        if (configStoreType == null || "memory".equals(configStoreType)) {
            return null;
        }

        ExtensionConfig storeConfig = new ExtensionConfig(
                configStoreType, configStoreType, configStoreParams);

        return ConfigStoreFactory.createConfigStore(
                tikaPluginManager,
                configStoreType,
                storeConfig);
    }

    /**
     * Creates a merged ParseContext with defaults from tika-config overlaid with request values.
     *
     * @param requestContext the ParseContext from FetchEmitTuple
     * @param presetName name of the preset to overlay at config-tier trust, or null
     * @return a new, resolved ParseContext with defaults + preset + clamped request overrides
     */
    public ParseContext createMergedParseContext(ParseContext requestContext, String presetName)
            throws TikaConfigException {
        ParseContext mergedContext = tikaLoader.loadParseContext();
        // EmbeddedDocumentExtractor is deliberately left unset here -- see PipesServer's
        // createMergedParseContext for why defaulting it would silently disable embedded
        // content extraction for every non-UNPACK parse mode.
        PipesServer.mergePreset(presetRegistry, presetName, mergedContext);
        PipesServer.overlayRequest(requestContext, mergedContext, pipesConfig,
                defaultContentHandlerFactory);
        PipesServer.seedCacheMemoryBudget(mergedContext);
        return mergedContext;
    }

    public TikaLoader getTikaLoader() {
        return tikaLoader;
    }

    public PipesConfig getPipesConfig() {
        return pipesConfig;
    }

    public AutoDetectParser getAutoDetectParser() {
        return autoDetectParser;
    }

    public Detector getDetector() {
        return detector;
    }

    public RecursiveParserWrapper getRMetaParser() {
        return rMetaParser;
    }

    public FetcherManager getFetcherManager() {
        return fetcherManager;
    }

    public EmitterManager getEmitterManager() {
        return emitterManager;
    }

    public MetadataFilter getDefaultMetadataFilter() {
        return defaultMetadataFilter;
    }

    /** Closes what the config loaded and holds open: the filter chain and the engines. */
    @Override
    public void close() throws IOException {
        try {
            defaultMetadataFilter.close();
        } finally {
            if (engineRegistry != null) {
                engineRegistry.close();
            }
        }
    }

    /** Null when no {@code "inference"} bindings are configured. */
    public InferenceDispatcher getInferenceDispatcher() {
        return inferenceDispatcher;
    }

    public ContentHandlerFactory getDefaultContentHandlerFactory() {
        return defaultContentHandlerFactory;
    }

    public MetadataWriteLimiterFactory getDefaultMetadataWriteLimiterFactory() {
        return defaultMetadataWriteLimiterFactory;
    }

    public EmitStrategy getEmitStrategy() {
        return emitStrategy;
    }

    public ExceptionReporting getExceptionReporting() {
        return exceptionReporting;
    }

    public ConfigStore getConfigStore() {
        return configStore;
    }
}