PipesWorker.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 java.io.InputStream;
import java.io.OutputStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.Collections;
import java.util.HashMap;
import java.util.IdentityHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.stream.Collectors;
import java.util.zip.ZipEntry;
import java.util.zip.ZipOutputStream;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.tika.exception.TikaConfigException;
import org.apache.tika.exception.TikaException;
import org.apache.tika.extractor.EmbeddedDocumentExtractor;
import org.apache.tika.extractor.UnpackHandler;
import org.apache.tika.io.TikaInputStream;
import org.apache.tika.metadata.HttpHeaders;
import org.apache.tika.metadata.Metadata;
import org.apache.tika.metadata.Property;
import org.apache.tika.metadata.TikaCoreProperties;
import org.apache.tika.metadata.writelimiter.MetadataWriteLimiterFactory;
import org.apache.tika.parser.AutoDetectParser;
import org.apache.tika.parser.ParseContext;
import org.apache.tika.pipes.api.FetchEmitTuple;
import org.apache.tika.pipes.api.ParseMode;
import org.apache.tika.pipes.api.PipesResult;
import org.apache.tika.pipes.api.emitter.EmitKey;
import org.apache.tika.pipes.api.emitter.Emitter;
import org.apache.tika.pipes.api.emitter.StreamEmitter;
import org.apache.tika.pipes.core.PipesResults;
import org.apache.tika.pipes.core.emitter.EmitterManager;
import org.apache.tika.pipes.core.extractor.EmittingUnpackHandler;
import org.apache.tika.pipes.core.extractor.FrictionlessUnpackHandler;
import org.apache.tika.pipes.core.extractor.TempFileUnpackHandler;
import org.apache.tika.pipes.core.extractor.UnpackConfig;
import org.apache.tika.pipes.core.extractor.UnpackExtractor;
import org.apache.tika.pipes.core.extractor.UnpackedByteCount;
import org.apache.tika.pipes.core.extractor.frictionless.DataPackage;
import org.apache.tika.pipes.core.extractor.frictionless.FrictionlessResource;
import org.apache.tika.utils.ExceptionUtils;
import org.apache.tika.utils.StringUtils;
class PipesWorker implements Callable<PipesResult> {
private static final Logger LOG = LoggerFactory.getLogger(PipesWorker.class);
static final String METADATA_FILE_NAME = "rmeta.json";
private final FetchEmitTuple fetchEmitTuple;
private final ParseContext parseContext;
private final AutoDetectParser autoDetectParser;
private final EmitterManager emitterManager;
private final FetchHandler fetchHandler;
private final ParseHandler parseHandler;
private final EmitHandler emitHandler;
private final MetadataWriteLimiterFactory defaultMetadataWriteLimiterFactory;
private final ParseMode defaultParseMode;
public PipesWorker(FetchEmitTuple fetchEmitTuple, ParseContext parseContext, AutoDetectParser autoDetectParser,
EmitterManager emitterManager, FetchHandler fetchHandler, ParseHandler parseHandler,
EmitHandler emitHandler, MetadataWriteLimiterFactory defaultMetadataWriteLimiterFactory,
ParseMode defaultParseMode) {
this.fetchEmitTuple = fetchEmitTuple;
this.parseContext = parseContext;
this.autoDetectParser = autoDetectParser;
this.emitterManager = emitterManager;
this.fetchHandler = fetchHandler;
this.parseHandler = parseHandler;
this.emitHandler = emitHandler;
this.defaultMetadataWriteLimiterFactory = defaultMetadataWriteLimiterFactory;
this.defaultParseMode = defaultParseMode;
}
// Per-parse stage nanos, read by ConnectionHandler for the timing line after the
// worker completes. -1 means the stage never ran.
private volatile long fetchNanos = -1;
private volatile long parseNanos = -1;
// text inference, metadata filters, user metadata
private volatile long prepareNanos = -1;
private volatile long emitNanos = -1;
private volatile long wallNanos = -1;
long getFetchNanos() {
return fetchNanos;
}
long getParseNanos() {
return parseNanos;
}
long getPrepareNanos() {
return prepareNanos;
}
long getEmitNanos() {
return emitNanos;
}
long getWallNanos() {
return wallNanos;
}
@Override
public PipesResult call() throws Exception {
long wallStart = System.nanoTime();
try {
return doCall();
} finally {
wallNanos = System.nanoTime() - wallStart;
}
}
private PipesResult doCall() throws Exception {
try {
//this can be null if there is a fetch exception
ParseDataOrPipesResult parseDataResult = parseFromTuple();
if (parseDataResult.pipesResult != null) {
return parseDataResult.pipesResult;
}
MetadataListAndEmbeddedBytes parseData = parseDataResult.parseDataResult;
if (parseData == null || metadataIsEmpty(parseData.getMetadataList())) {
return PipesResults.EMPTY_OUTPUT;
}
// before prepare: a filter may strip tk:embedded-id
Map<Integer, Metadata> rowsById = rowsById(parseData.getMetadataList());
// before the unpack package: it carries the same metadata list as the emit
long prepareStart = System.nanoTime();
EmitHandler.Prepared prepared = emitHandler.prepare(fetchEmitTuple, parseData, parseContext);
prepareNanos = System.nanoTime() - prepareStart;
// the unpack output is part of the emit
long emitStart = System.nanoTime();
try {
// Check if we need to zip and emit embedded files
UnpackHandler handler = parseContext.get(UnpackHandler.class);
if (handler instanceof FrictionlessUnpackHandler) {
PipesResult frictionlessResult =
emitFrictionlessOutput((FrictionlessUnpackHandler) handler, parseData);
if (frictionlessResult != null) {
// Frictionless emit failed - return the error
return frictionlessResult;
}
} else if (handler instanceof TempFileUnpackHandler) {
PipesResult zipResult = zipAndEmitEmbeddedFiles((TempFileUnpackHandler) handler,
parseData, rowsById);
if (zipResult != null) {
// Zipping/emitting failed - return the error
return zipResult;
}
}
return emitHandler.emitParseData(fetchEmitTuple, parseData, prepared, parseContext);
} finally {
emitNanos = System.nanoTime() - emitStart;
}
} finally {
// the handler exists from setup on, so fetch and parse failures must close it too
UnpackHandler handler = parseContext.get(UnpackHandler.class);
if (handler instanceof Closeable) {
try {
((Closeable) handler).close();
} catch (IOException e) {
LOG.warn("problem closing unpack handler", e);
}
}
}
}
static boolean metadataIsEmpty(List<Metadata> metadataList) {
return metadataList == null || metadataList.isEmpty();
}
/**
* Zips all embedded files from the temp handler and emits the zip to the user's emitter.
*
* @param tempHandler the handler containing embedded files in temp directory
* @return PipesResult if there was an error, null if successful
*/
private PipesResult zipAndEmitEmbeddedFiles(TempFileUnpackHandler tempHandler,
MetadataListAndEmbeddedBytes parseData,
Map<Integer, Metadata> rowsById) {
// Skip if no embedded files
if (!tempHandler.hasEmbeddedFiles()) {
LOG.debug("No embedded files to zip");
return null;
}
UnpackConfig unpackConfig = parseContext.get(UnpackConfig.class);
String emitterName = unpackConfig.getEmitter();
Emitter emitter;
try {
emitter = emitterManager.getEmitter(emitterName);
} catch (Exception e) {
LOG.warn("Failed to get emitter for zip: {}", emitterName, e);
return new PipesResult(PipesResult.RESULT_STATUS.EMITTER_NOT_FOUND,
"Emitter not found for zipping: " + emitterName);
}
if (!(emitter instanceof StreamEmitter)) {
LOG.warn("Emitter {} is not a StreamEmitter, cannot emit zip", emitterName);
return new PipesResult(PipesResult.RESULT_STATUS.EMIT_EXCEPTION,
"Emitter must be a StreamEmitter to emit zipped embedded files. Found: " +
emitter.getClass().getName());
}
StreamEmitter streamEmitter = (StreamEmitter) emitter;
EmitKey containerEmitKey = fetchEmitTuple.getEmitKey();
// Create zip file in temp directory
Path zipFile = tempHandler.getTempDirectory().resolve("embedded-files.zip");
try {
createZipFile(zipFile, tempHandler, unpackConfig, parseData.getMetadataList(), rowsById);
} catch (IOException e) {
LOG.warn("Failed to create zip file", e);
return new PipesResult(PipesResult.RESULT_STATUS.EMIT_EXCEPTION,
"Failed to create zip file: " + ExceptionUtils.format(e, parseContext));
}
// Emit the zip file
String zipEmitKey = containerEmitKey.getEmitKey() + "-embedded.zip";
try (InputStream zipStream = Files.newInputStream(zipFile)) {
streamEmitter.emit(zipEmitKey, zipStream, new Metadata(), parseContext);
} catch (IOException e) {
LOG.warn("Failed to emit zip file", e);
return new PipesResult(PipesResult.RESULT_STATUS.EMIT_EXCEPTION,
"Failed to emit zip file: " + ExceptionUtils.format(e, parseContext));
}
LOG.debug("Successfully zipped and emitted {} embedded files to {}",
tempHandler.getEmbeddedFiles().size(), zipEmitKey);
return null;
}
/**
* Emits Frictionless Data Package output (zipped or directory mode).
*
* @param frictionlessHandler the handler containing embedded files with SHA256 hashes
* @param parseData the parsed metadata list for optional inclusion
* @return PipesResult if there was an error, null if successful
*/
private PipesResult emitFrictionlessOutput(FrictionlessUnpackHandler frictionlessHandler,
MetadataListAndEmbeddedBytes parseData) {
UnpackConfig unpackConfig = frictionlessHandler.getUnpackConfig();
String emitterName = unpackConfig.getEmitter();
// Same emitter: no separate <emitKey>.json, the package carries the metadata list
boolean replacesEmit = emitterName != null &&
emitterName.equals(fetchEmitTuple.getEmitKey().getEmitterId());
// Written even with no embedded files, same shape on every surface, unless asked not to
if (!frictionlessHandler.hasEmbeddedFiles() &&
unpackConfig.getEmptyPackage() == UnpackConfig.EMPTY_PACKAGE.SKIP) {
parseData.setPackageReplacesEmit(replacesEmit);
return null;
}
Emitter emitter;
try {
emitter = emitterManager.getEmitter(emitterName);
} catch (Exception e) {
LOG.warn("Failed to get emitter for Frictionless output: {}", emitterName, e);
return new PipesResult(PipesResult.RESULT_STATUS.EMITTER_NOT_FOUND,
"Emitter not found for Frictionless output: " + emitterName);
}
if (!(emitter instanceof StreamEmitter)) {
LOG.warn("Emitter {} is not a StreamEmitter, cannot emit Frictionless output", emitterName);
return new PipesResult(PipesResult.RESULT_STATUS.EMIT_EXCEPTION,
"Emitter must be a StreamEmitter for Frictionless output. Found: " +
emitter.getClass().getName());
}
StreamEmitter streamEmitter = (StreamEmitter) emitter;
EmitKey containerEmitKey = fetchEmitTuple.getEmitKey();
// Package name: the host's, else the fetch key's file name
String containerName = unpackConfig.getPackageName();
if (containerName == null) {
String fetchKey = fetchEmitTuple.getFetchKey().getFetchKey();
containerName = fetchKey;
int lastSlash = Math.max(fetchKey.lastIndexOf('/'), fetchKey.lastIndexOf('\\'));
if (lastSlash >= 0 && lastSlash < fetchKey.length() - 1) {
containerName = fetchKey.substring(lastSlash + 1);
}
}
DataPackage dataPackage = frictionlessHandler.buildDataPackage(containerName);
byte[] metadataJson = null;
if (unpackConfig.writesMetadata() && parseData != null &&
parseData.getMetadataList() != null) {
try {
metadataJson = writeMetadataListAsJson(parseData.getMetadataList());
} catch (IOException e) {
LOG.warn("Failed to serialize metadata for Frictionless output", e);
return new PipesResult(PipesResult.RESULT_STATUS.EMIT_EXCEPTION,
"Failed to serialize metadata for Frictionless output: " +
ExceptionUtils.format(e, parseContext));
}
dataPackage.getResources().add(0, FrictionlessResource.create(METADATA_FILE_NAME,
"application/json", metadataJson.length,
FrictionlessResource.formatHash(sha256(metadataJson))));
}
PipesResult result;
if (unpackConfig.getOutputMode() == UnpackConfig.OUTPUT_MODE.ZIPPED) {
result = emitFrictionlessZipped(frictionlessHandler, streamEmitter, containerEmitKey,
dataPackage, metadataJson);
} else {
result = emitFrictionlessDirectory(frictionlessHandler, streamEmitter,
packageRoot(containerEmitKey, unpackConfig), dataPackage, metadataJson);
}
if (result == null) {
parseData.setPackageReplacesEmit(replacesEmit);
}
return result;
}
/** Emit key prefix of a directory-mode package; see {@link UnpackConfig#getPackageRoot()}. */
private static String packageRoot(EmitKey containerEmitKey, UnpackConfig unpackConfig) {
String root = unpackConfig.getPackageRoot() != null ?
unpackConfig.getPackageRoot() : containerEmitKey.getEmitKey();
return root.isEmpty() ? "" : root + "/";
}
private static byte[] sha256(byte[] bytes) {
try {
return MessageDigest.getInstance("SHA-256").digest(bytes);
} catch (NoSuchAlgorithmException e) {
throw new IllegalStateException("SHA-256 algorithm not available", e);
}
}
/**
* Emits Frictionless Data Package as a single zip file.
*/
private PipesResult emitFrictionlessZipped(FrictionlessUnpackHandler frictionlessHandler,
StreamEmitter streamEmitter,
EmitKey containerEmitKey,
DataPackage dataPackage,
byte[] metadataJson) {
// Create zip file in temp directory
Path zipFile = frictionlessHandler.getTempDirectory().resolve("frictionless-package.zip");
try {
createFrictionlessZipFile(zipFile, frictionlessHandler, dataPackage, metadataJson);
} catch (IOException e) {
LOG.warn("Failed to create Frictionless zip file", e);
return new PipesResult(PipesResult.RESULT_STATUS.EMIT_EXCEPTION,
"Failed to create Frictionless zip file: " + ExceptionUtils.format(e, parseContext));
}
// Emit the zip file
String zipEmitKey = containerEmitKey.getEmitKey() + "-frictionless.zip";
try (InputStream zipStream = Files.newInputStream(zipFile)) {
streamEmitter.emit(zipEmitKey, zipStream, new Metadata(), parseContext);
} catch (IOException e) {
LOG.warn("Failed to emit Frictionless zip file", e);
return new PipesResult(PipesResult.RESULT_STATUS.EMIT_EXCEPTION,
"Failed to emit Frictionless zip file: " + ExceptionUtils.format(e, parseContext));
}
LOG.debug("Successfully emitted Frictionless package with {} resources to {}",
dataPackage.resourceCount(), zipEmitKey);
return null;
}
/**
* Emits Frictionless Data Package files directly to the emitter (directory mode).
*/
private PipesResult emitFrictionlessDirectory(FrictionlessUnpackHandler frictionlessHandler,
StreamEmitter streamEmitter,
String packageRoot,
DataPackage dataPackage,
byte[] metadataJson) {
try {
// Emit each embedded file under unpacked/.
// When includeOriginal=true the container itself is added as id 0 by
// ParseHandler._preParse, so it appears here as one of the embedded entries.
for (FrictionlessUnpackHandler.FrictionlessFileInfo fileInfo : frictionlessHandler.getEmbeddedFiles()) {
String fileEmitKey = packageRoot + "unpacked/" + fileInfo.fileName();
try (InputStream is = Files.newInputStream(fileInfo.filePath())) {
streamEmitter.emit(fileEmitKey, is, fileInfo.metadata(), parseContext);
}
}
// Emit datapackage.json
String dpEmitKey = packageRoot + "datapackage.json";
byte[] dpBytes = dataPackage.toJson().getBytes(java.nio.charset.StandardCharsets.UTF_8);
try (InputStream dpStream = new java.io.ByteArrayInputStream(dpBytes)) {
streamEmitter.emit(dpEmitKey, dpStream, new Metadata(), parseContext);
}
if (metadataJson != null) {
try (InputStream metadataStream = new java.io.ByteArrayInputStream(metadataJson)) {
streamEmitter.emit(packageRoot + METADATA_FILE_NAME, metadataStream,
new Metadata(), parseContext);
}
}
} catch (IOException e) {
LOG.warn("Failed to emit Frictionless directory output", e);
return new PipesResult(PipesResult.RESULT_STATUS.EMIT_EXCEPTION,
"Failed to emit Frictionless directory output: " + ExceptionUtils.format(e, parseContext));
}
LOG.debug("Successfully emitted Frictionless package with {} resources (directory mode) to {}",
dataPackage.resourceCount(), packageRoot);
return null;
}
/**
* Creates a Frictionless Data Package zip file.
*/
private void createFrictionlessZipFile(Path zipFile, FrictionlessUnpackHandler frictionlessHandler,
DataPackage dataPackage, byte[] metadataJson) throws IOException {
try (ZipOutputStream zos = new ZipOutputStream(Files.newOutputStream(zipFile))) {
// Add datapackage.json at root
ZipEntry dpEntry = new ZipEntry("datapackage.json");
zos.putNextEntry(dpEntry);
dataPackage.writeTo(zos);
zos.closeEntry();
if (metadataJson != null) {
zos.putNextEntry(new ZipEntry(METADATA_FILE_NAME));
zos.write(metadataJson);
zos.closeEntry();
}
// Add all embedded files under unpacked/.
// When includeOriginal=true the container itself is added as id 0 by
// ParseHandler._preParse, so it appears here as one of the embedded entries.
for (FrictionlessUnpackHandler.FrictionlessFileInfo fileInfo : frictionlessHandler.getEmbeddedFiles()) {
ZipEntry fileEntry = new ZipEntry("unpacked/" + fileInfo.fileName());
zos.putNextEntry(fileEntry);
Files.copy(fileInfo.filePath(), zos);
zos.closeEntry();
}
}
}
/**
* Writes a list of metadata objects as JSON array to output stream.
*/
private void writeMetadataListAsJson(OutputStream os, List<Metadata> metadataList) throws IOException {
ObjectMapper mapper = new ObjectMapper();
mapper.configure(com.fasterxml.jackson.core.JsonGenerator.Feature.AUTO_CLOSE_TARGET, false);
mapper.enable(com.fasterxml.jackson.databind.SerializationFeature.INDENT_OUTPUT);
List<java.util.Map<String, Object>> metadataMapList = new java.util.ArrayList<>();
for (Metadata metadata : metadataList) {
metadataMapList.add(metadataMap(metadata));
}
mapper.writeValue(os, metadataMapList);
}
/**
* Writes a list of metadata objects as JSON array to byte array.
*/
private byte[] writeMetadataListAsJson(List<Metadata> metadataList) throws IOException {
java.io.ByteArrayOutputStream baos = new java.io.ByteArrayOutputStream();
writeMetadataListAsJson(baos, metadataList);
return baos.toByteArray();
}
/**
* Creates a zip file containing all embedded files.
*/
private void createZipFile(Path zipFile, TempFileUnpackHandler tempHandler,
UnpackConfig unpackConfig, List<Metadata> metadataList,
Map<Integer, Metadata> rowsById) throws IOException {
// Sidecars are the list's rows, so they get what rmeta.json gets: inference, filters
Set<Metadata> kept = Collections.newSetFromMap(new IdentityHashMap<>());
kept.addAll(metadataList);
try (ZipOutputStream zos = new ZipOutputStream(Files.newOutputStream(zipFile))) {
// When includeOriginal=true the container itself is added as id 0 by
// ParseHandler._preParse, so it appears here as one of the embedded entries.
for (TempFileUnpackHandler.EmbeddedFileInfo fileInfo : tempHandler.getEmbeddedFiles()) {
// Add the embedded file
ZipEntry fileEntry = new ZipEntry(fileInfo.fileName());
zos.putNextEntry(fileEntry);
Files.copy(fileInfo.filePath(), zos);
zos.closeEntry();
// Add metadata JSON if requested; a row a filter removed gets no sidecar
Metadata row = rowsById.get(fileInfo.id());
if (unpackConfig.writesMetadata() && row != null && kept.contains(row)) {
String metadataFileName = fileInfo.fileName() + ".metadata.json";
ZipEntry metadataEntry = new ZipEntry(metadataFileName);
zos.putNextEntry(metadataEntry);
writeMetadataAsJson(zos, row);
zos.closeEntry();
}
}
}
}
/** Rows by {@code tk:embedded-id}; the container, which has none, is 0. */
private static Map<Integer, Metadata> rowsById(List<Metadata> metadataList) {
Map<Integer, Metadata> rows = new HashMap<>();
for (int i = 0; i < metadataList.size(); i++) {
Metadata row = metadataList.get(i);
Integer id = i == 0 ? Integer.valueOf(0) : row.getInt(TikaCoreProperties.EMBEDDED_ID);
if (id != null) {
rows.put(id, row);
}
}
return rows;
}
/**
* Writes metadata as JSON to the output stream.
* Note: Does not close the output stream.
*/
private void writeMetadataAsJson(OutputStream os, Metadata metadata) throws IOException {
ObjectMapper mapper = new ObjectMapper();
// Disable auto-close so we don't close the zip output stream
mapper.configure(com.fasterxml.jackson.core.JsonGenerator.Feature.AUTO_CLOSE_TARGET, false);
mapper.writeValue(os, metadataMap(metadata));
}
private static java.util.Map<String, Object> metadataMap(Metadata metadata) {
java.util.Map<String, Object> metadataMap = new java.util.LinkedHashMap<>();
for (String name : metadata.names()) {
String[] values = metadata.getValues(name);
metadataMap.put(name, values.length == 1 ? values[0] : values);
}
return metadataMap;
}
protected ParseDataOrPipesResult parseFromTuple() throws TikaException, InterruptedException {
//start a new metadata object to gather info from the fetch process
//we want to isolate and not touch the metadata sent into the fetchEmitTuple
//so that we can inject it after the filter at the very end
ParseContext localContext = null;
try {
localContext = setupParseContext();
} catch (IOException e) {
LOG.warn("fetcher initialization exception id={}", fetchEmitTuple.getId(), e);
return new ParseDataOrPipesResult(null,
new PipesResult(PipesResult.RESULT_STATUS.FETCHER_INITIALIZATION_EXCEPTION, ExceptionUtils.format(e, parseContext)));
}
// Use newMetadata() to apply any configured write limits
Metadata metadata = localContext.newMetadata();
// Carry the caller's resource name and Content-Type detection hints (see javadoc).
carryCallerHints(fetchEmitTuple.getMetadata(), metadata);
long fetchStart = System.nanoTime();
FetchHandler.TisOrResult tisOrResult = fetchHandler.fetch(fetchEmitTuple, metadata, localContext);
fetchNanos = System.nanoTime() - fetchStart;
if (tisOrResult.pipesResult() != null) {
return new ParseDataOrPipesResult(null, tisOrResult.pipesResult());
}
try (TikaInputStream tis = tisOrResult.tis()) {
long parseStart = System.nanoTime();
try {
return parseHandler.parseWithStream(fetchEmitTuple, tis, metadata, localContext);
} finally {
parseNanos = System.nanoTime() - parseStart;
}
} catch (SecurityException e) {
LOG.error("security exception id={}", fetchEmitTuple.getId(), e);
throw e;
} catch (TikaException | IOException e) {
LOG.warn("fetch exception id={}", fetchEmitTuple.getId(), e);
return new ParseDataOrPipesResult(null,
new PipesResult(PipesResult.RESULT_STATUS.UNSPECIFIED_CRASH, ExceptionUtils.format(e, parseContext)));
}
}
/**
* The user-metadata keys that are inputs to the parse: carried across the fresh-metadata
* boundary into detection here, and NOT re-injected onto the result by
* {@code EmitHandler#injectUserMetadata} -- the parse's own value for them (the detected
* type, the given name) is what the caller gets back. Every other user key is the reverse:
* invisible to the parse, injected verbatim on output.
* <p>
* {@code Content-Type} is a soft hint: {@code MimeTypes.detect} keeps it only when it equals
* or specializes the magic-detected type (e.g. {@code image/tiff} ->
* {@code image/x-raw-nikon} for a NEF supplied without a filename). The
* {@code CONTENT_TYPE_USER_OVERRIDE} and {@code CONTENT_TYPE_PARSER_OVERRIDE} keys are
* deliberately NOT in this set: they short-circuit detection unconditionally and would let a
* caller force any type.
*/
static final Set<Property> CALLER_INPUT_KEYS = Set.of(
TikaCoreProperties.RESOURCE_NAME_KEY,
HttpHeaders.CONTENT_TYPE);
/** {@link #CALLER_INPUT_KEYS} by name, for callers holding a key as a String. */
static final Set<String> CALLER_INPUT_KEY_NAMES = CALLER_INPUT_KEYS.stream()
.map(Property::getName).collect(Collectors.toUnmodifiableSet());
/**
* Carries {@link #CALLER_INPUT_KEYS} from the tuple metadata into the metadata used for
* fetch and detection.
*
* @param tupleMetadata the caller-supplied metadata (may be null)
* @param target the fresh metadata used for fetch and detection
*/
static void carryCallerHints(Metadata tupleMetadata, Metadata target) {
if (tupleMetadata == null) {
return;
}
for (Property key : CALLER_INPUT_KEYS) {
String supplied = tupleMetadata.get(key);
if (!StringUtils.isBlank(supplied)) {
target.set(key, supplied);
}
}
}
private ParseContext setupParseContext() throws TikaException, IOException {
// ContentHandlerFactory and ParseMode are retrieved from ParseContext in ParseHandler.
// They are set in ParseContext from PipesConfig loaded via TikaLoader at startup.
// If the parseContext from the FetchEmitTuple doesn't have a MetadataWriteLimiterFactory,
// use the default one loaded from config in PipesServer
MetadataWriteLimiterFactory existingFactory = parseContext.get(MetadataWriteLimiterFactory.class);
if (existingFactory == null && defaultMetadataWriteLimiterFactory != null) {
parseContext.set(MetadataWriteLimiterFactory.class, defaultMetadataWriteLimiterFactory);
}
ParseMode parseMode = parseContext.get(ParseMode.class);
if (parseMode == null) {
// Write the resolved default back so EmitHandler sees it too - it reads ParseMode
// only from the context, with no fallback to the PipesConfig default (TIKA-4735).
parseMode = defaultParseMode;
parseContext.set(ParseMode.class, parseMode);
}
UnpackConfig unpackConfig = parseContext.get(UnpackConfig.class);
// For UNPACK mode, automatically set up byte extraction
if (parseMode == ParseMode.UNPACK) {
if (unpackConfig == null) {
unpackConfig = new UnpackConfig();
parseContext.set(UnpackConfig.class, unpackConfig);
}
// Determine emitter: prefer UnpackConfig, fall back to FetchEmitTuple
String emitterName = unpackConfig.getEmitter();
if (StringUtils.isBlank(emitterName)) {
emitterName = fetchEmitTuple.getEmitKey().getEmitterId();
if (StringUtils.isBlank(emitterName)) {
throw new TikaConfigException(
"UNPACK parse mode requires an emitter. Set emitter in UnpackConfig " +
"or specify an emitterId in FetchEmitTuple.emitKey.");
}
unpackConfig.setEmitter(emitterName);
}
// Set up the extractor and its per-request byte budget together: UnpackExtractor
// is stateless, so the running byte count lives in the context, not on the
// extractor, and must be created here -- never lazily inside the extractor --
// so it's scoped to exactly this request.
parseContext.set(EmbeddedDocumentExtractor.class, UnpackExtractor.INSTANCE);
parseContext.set(UnpackedByteCount.class, new UnpackedByteCount());
// Set up the bytes handler based on output format and mode
if (unpackConfig.getOutputFormat() == UnpackConfig.OUTPUT_FORMAT.FRICTIONLESS) {
// Frictionless Data Package format - always uses FrictionlessUnpackHandler
// which computes SHA256 hashes and stores files for datapackage.json generation
parseContext.set(UnpackHandler.class,
new FrictionlessUnpackHandler(fetchEmitTuple.getEmitKey(), unpackConfig));
} else if (unpackConfig.isZipEmbeddedFiles()) {
// Regular format with zipping - use TempFileUnpackHandler
parseContext.set(UnpackHandler.class,
new TempFileUnpackHandler(fetchEmitTuple.getEmitKey(), unpackConfig));
} else {
// Regular format, direct emission - use EmittingUnpackHandler
parseContext.set(UnpackHandler.class,
new EmittingUnpackHandler(fetchEmitTuple, emitterManager, parseContext));
}
return parseContext;
}
// For non-UNPACK modes, no byte extraction setup needed
// UnpackConfig may be present from config file but is only used when ParseMode.UNPACK is set
return parseContext;
}
//parse data result or a terminal pipesresult
record ParseDataOrPipesResult(MetadataListAndEmbeddedBytes parseDataResult, PipesResult pipesResult) {
}
}