Add GoogleFetcher

tika-core/src/main/java/org/apache/tika/pipes/PipesConfig.java

@@ -46,6 +46,12 @@ public static PipesConfig load(Path tikaConfig) throws IOException, TikaConfigEx
         return pipesConfig;
     }
 
+    public static PipesConfig load(InputStream tikaConfigInputStream) throws IOException, TikaConfigException {
+        PipesConfig pipesConfig = new PipesConfig();
+        pipesConfig.configure("pipes", tikaConfigInputStream);
+        return pipesConfig;
+    }
+
     private PipesConfig() {
 
     }

tika-core/src/main/java/org/apache/tika/pipes/PipesConfigBase.java

@@ -54,7 +54,10 @@ public class PipesConfigBase extends ConfigBase {
     private int numClients = DEFAULT_NUM_CLIENTS;
 
     private int maxFilesProcessedPerProcess = DEFAULT_MAX_FILES_PROCESSED_PER_PROCESS;
-
+    public static final int DEFAULT_STALE_FETCHER_TIMEOUT_SECONDS = 600;
+    private int staleFetcherTimeoutSeconds = DEFAULT_STALE_FETCHER_TIMEOUT_SECONDS;
+    public static final int DEFAULT_STALE_FETCHER_DELAY_SECONDS = 60;
+    private int staleFetcherDelaySeconds = DEFAULT_STALE_FETCHER_DELAY_SECONDS;
     private List<String> forkedJvmArgs = new ArrayList<>();
     private Path tikaConfig;
     private String javaPath = "java";
@@ -171,4 +174,20 @@ public long getSleepOnStartupTimeoutMillis() {
     public void setSleepOnStartupTimeoutMillis(long sleepOnStartupTimeoutMillis) {
         this.sleepOnStartupTimeoutMillis = sleepOnStartupTimeoutMillis;
     }
+
+    public int getStaleFetcherTimeoutSeconds() {
+        return staleFetcherTimeoutSeconds;
+    }
+
+    public void setStaleFetcherTimeoutSeconds(int staleFetcherTimeoutSeconds) {
+        this.staleFetcherTimeoutSeconds = staleFetcherTimeoutSeconds;
+    }
+
+    public int getStaleFetcherDelaySeconds() {
+        return staleFetcherDelaySeconds;
+    }
+
+    public void setStaleFetcherDelaySeconds(int staleFetcherDelaySeconds) {
+        this.staleFetcherDelaySeconds = staleFetcherDelaySeconds;
+    }
 }

tika-core/src/main/java/org/apache/tika/pipes/PipesServer.java

@@ -69,10 +69,11 @@
 import org.apache.tika.pipes.emitter.EmitterManager;
 import org.apache.tika.pipes.emitter.StreamEmitter;
 import org.apache.tika.pipes.emitter.TikaEmitterException;
-import org.apache.tika.pipes.extractor.EmbeddedDocumentBytesConfig;
 import org.apache.tika.pipes.extractor.EmittingEmbeddedDocumentBytesHandler;
+import org.apache.tika.pipes.fetcher.FetchKey;
 import org.apache.tika.pipes.fetcher.Fetcher;
 import org.apache.tika.pipes.fetcher.FetcherManager;
+import org.apache.tika.pipes.fetcher.RangeFetcher;
 import org.apache.tika.sax.BasicContentHandlerFactory;
 import org.apache.tika.sax.ContentHandlerFactory;
 import org.apache.tika.sax.RecursiveParserWrapperHandler;
@@ -280,7 +281,7 @@ private String getContainerStacktrace(FetchEmitTuple t, List<Metadata> metadataL
 
     private void emit(String taskId, EmitKey emitKey,
                       boolean isExtractEmbeddedBytes, MetadataListAndEmbeddedBytes parseData,
-                      String parseExceptionStack, ParseContext parseContext) {
+                      String parseExceptionStack) {
         Emitter emitter = null;
 
         try {
@@ -296,7 +297,7 @@ private void emit(String taskId, EmitKey emitKey,
                     parseData.toBePackagedForStreamEmitter()) {
                 emitContentsAndBytes(emitter, emitKey, parseData);
             } else {
-                emitter.emit(emitKey.getEmitKey(), parseData.getMetadataList(), parseContext);
+                emitter.emit(emitKey.getEmitKey(), parseData.getMetadataList());
             }
         } catch (IOException | TikaEmitterException e) {
             LOG.warn("emit exception", e);
@@ -377,7 +378,7 @@ private void actuallyParse(FetchEmitTuple t) {
                 LOG.trace("timer -- to parse: {} ms", System.currentTimeMillis() - start);
             }
 
-            if (metadataIsEmpty(parseData.getMetadataList())) {
+            if (parseData == null || metadataIsEmpty(parseData.getMetadataList())) {
                 write(STATUS.EMPTY_OUTPUT);
                 return;
             }
@@ -400,26 +401,23 @@ private void emitParseData(FetchEmitTuple t, MetadataListAndEmbeddedBytes parseD
         String stack = getContainerStacktrace(t, parseData.getMetadataList());
         //we need to apply this after we pull out the stacktrace
         filterMetadata(parseData.getMetadataList());
-        ParseContext parseContext = t.getParseContext();
-        FetchEmitTuple.ON_PARSE_EXCEPTION onParseException = t.getOnParseException();
-        EmbeddedDocumentBytesConfig embeddedDocumentBytesConfig = parseContext.get(EmbeddedDocumentBytesConfig.class);
         if (StringUtils.isBlank(stack) ||
-                onParseException == FetchEmitTuple.ON_PARSE_EXCEPTION.EMIT) {
+                t.getOnParseException() == FetchEmitTuple.ON_PARSE_EXCEPTION.EMIT) {
             injectUserMetadata(t.getMetadata(), parseData.getMetadataList());
             EmitKey emitKey = t.getEmitKey();
             if (StringUtils.isBlank(emitKey.getEmitKey())) {
                 emitKey = new EmitKey(emitKey.getEmitterName(), t.getFetchKey().getFetchKey());
                 t.setEmitKey(emitKey);
             }
             EmitData emitData = new EmitData(t.getEmitKey(), parseData.getMetadataList(), stack);
-            if (embeddedDocumentBytesConfig.isExtractEmbeddedDocumentBytes() &&
+            if (t.getEmbeddedDocumentBytesConfig().isExtractEmbeddedDocumentBytes() &&
                     parseData.toBePackagedForStreamEmitter()) {
-                emit(t.getId(), emitKey, embeddedDocumentBytesConfig.isExtractEmbeddedDocumentBytes(),
-                        parseData, stack, parseContext);
+                emit(t.getId(), emitKey, t.getEmbeddedDocumentBytesConfig().isExtractEmbeddedDocumentBytes(),
+                        parseData, stack);
             } else if (maxForEmitBatchBytes >= 0 &&
                     emitData.getEstimatedSizeBytes() >= maxForEmitBatchBytes) {
-                emit(t.getId(), emitKey, embeddedDocumentBytesConfig.isExtractEmbeddedDocumentBytes(),
-                        parseData, stack, parseContext);
+                emit(t.getId(), emitKey, t.getEmbeddedDocumentBytesConfig().isExtractEmbeddedDocumentBytes(),
+                        parseData, stack);
             } else {
                 //send back to the client
                 write(emitData);
@@ -458,18 +456,35 @@ private Fetcher getFetcher(FetchEmitTuple t) {
     }
 
     protected MetadataListAndEmbeddedBytes parseFromTuple(FetchEmitTuple t, Fetcher fetcher) {
-
-        Metadata metadata = new Metadata();
-        try (InputStream stream = fetcher.fetch(t.getFetchKey().getFetchKey(), metadata, t.getParseContext())) {
-            return parseWithStream(t, stream, metadata);
-        } catch (SecurityException e) {
-            LOG.error("security exception " + t.getId(), e);
-            throw e;
-        } catch (TikaException | IOException e) {
-            LOG.warn("fetch exception " + t.getId(), e);
-            write(STATUS.FETCH_EXCEPTION, ExceptionUtils.getStackTrace(e));
+        FetchKey fetchKey = t.getFetchKey();
+        if (fetchKey.hasRange()) {
+            if (!(fetcher instanceof RangeFetcher)) {
+                throw new IllegalArgumentException(
+                        "fetch key has a range, but the fetcher is not a range fetcher");
+            }
+            Metadata metadata = t.getMetadata() == null ? new Metadata() : t.getMetadata();
+            try (InputStream stream = ((RangeFetcher) fetcher).fetch(fetchKey.getFetchKey(),
+                    fetchKey.getRangeStart(), fetchKey.getRangeEnd(), metadata)) {
+                return parseWithStream(t, stream, metadata);
+            } catch (SecurityException e) {
+                LOG.error("security exception " + t.getId(), e);
+                throw e;
+            } catch (TikaException | IOException e) {
+                LOG.warn("fetch exception " + t.getId(), e);
+                write(STATUS.FETCH_EXCEPTION, ExceptionUtils.getStackTrace(e));
+            }
+        } else {
+            Metadata metadata = t.getMetadata() == null ? new Metadata() : t.getMetadata();
+            try (InputStream stream = fetcher.fetch(t.getFetchKey().getFetchKey(), metadata)) {
+                return parseWithStream(t, stream, metadata);
+            } catch (SecurityException e) {
+                LOG.error("security exception " + t.getId(), e);
+                throw e;
+            } catch (TikaException | IOException e) {
+                LOG.warn("fetch exception " + t.getId(), e);
+                write(STATUS.FETCH_EXCEPTION, ExceptionUtils.getStackTrace(e));
+            }
         }
-
         return null;
     }
 
@@ -513,11 +528,10 @@ private void handleOOM(String taskId, OutOfMemoryError oom) {
     private MetadataListAndEmbeddedBytes parseWithStream(FetchEmitTuple fetchEmitTuple,
                                                          InputStream stream, Metadata metadata)
             throws TikaConfigException {
-
+        HandlerConfig handlerConfig = fetchEmitTuple.getHandlerConfig();
         List<Metadata> metadataList;
         //this adds the EmbeddedDocumentByteStore to the parsecontext
-        ParseContext parseContext = setupParseContext(fetchEmitTuple);
-        HandlerConfig handlerConfig = parseContext.get(HandlerConfig.class);
+        ParseContext parseContext = createParseContext(fetchEmitTuple);
         if (handlerConfig.getParseMode() == HandlerConfig.PARSE_MODE.RMETA) {
             metadataList =
                     parseRecursive(fetchEmitTuple, handlerConfig, stream, metadata, parseContext);
@@ -530,16 +544,10 @@ private MetadataListAndEmbeddedBytes parseWithStream(FetchEmitTuple fetchEmitTup
                 parseContext.get(EmbeddedDocumentBytesHandler.class));
     }
 
-    private ParseContext setupParseContext(FetchEmitTuple fetchEmitTuple)
+    private ParseContext createParseContext(FetchEmitTuple fetchEmitTuple)
             throws TikaConfigException {
-        ParseContext parseContext = fetchEmitTuple.getParseContext();
-        if (parseContext.get(HandlerConfig.class) == null) {
-            parseContext.set(HandlerConfig.class, HandlerConfig.DEFAULT_HANDLER_CONFIG);
-        }
-        EmbeddedDocumentBytesConfig embeddedDocumentBytesConfig = parseContext.get(EmbeddedDocumentBytesConfig.class);
-        if (embeddedDocumentBytesConfig == null) {
-            //make sure there's one here -- or do we make this default in fetchemit tuple?
-            parseContext.set(EmbeddedDocumentBytesConfig.class, EmbeddedDocumentBytesConfig.SKIP);
+        ParseContext parseContext = new ParseContext();
+        if (! fetchEmitTuple.getEmbeddedDocumentBytesConfig().isExtractEmbeddedDocumentBytes()) {
             return parseContext;
         }
         EmbeddedDocumentExtractorFactory factory = ((AutoDetectParser)autoDetectParser)
@@ -553,17 +561,18 @@ private ParseContext setupParseContext(FetchEmitTuple fetchEmitTuple)
                         "instance of EmbeddedDocumentByteStoreExtractorFactory if you want" +
                         "to extract embedded bytes! I see this embedded doc factory: " +
                         factory.getClass() + "and a request: " +
-                        embeddedDocumentBytesConfig);
+                        fetchEmitTuple.getEmbeddedDocumentBytesConfig());
             }
         }
         //TODO: especially clean this up.
-        if (!StringUtils.isBlank(embeddedDocumentBytesConfig.getEmitter())) {
+        if (!StringUtils.isBlank(fetchEmitTuple.getEmbeddedDocumentBytesConfig().getEmitter())) {
             parseContext.set(EmbeddedDocumentBytesHandler.class,
-                    new EmittingEmbeddedDocumentBytesHandler(fetchEmitTuple, emitterManager));
+                    new EmittingEmbeddedDocumentBytesHandler(fetchEmitTuple.getEmitKey(),
+                            fetchEmitTuple.getEmbeddedDocumentBytesConfig(), emitterManager));
         } else {
             parseContext.set(EmbeddedDocumentBytesHandler.class,
                     new BasicEmbeddedDocumentBytesHandler(
-                    embeddedDocumentBytesConfig));
+                    fetchEmitTuple.getEmbeddedDocumentBytesConfig()));
         }
         return parseContext;
     }
@@ -684,10 +693,11 @@ private void _preParse(FetchEmitTuple t, TikaInputStream tis, Metadata metadata,
         } catch (IOException e) {
             LOG.warn("problem detecting: " + t.getId(), e);
         }
-        EmbeddedDocumentBytesConfig embeddedDocumentBytesConfig = parseContext.get(EmbeddedDocumentBytesConfig.class);
-        if (embeddedDocumentBytesConfig != null &&
-                embeddedDocumentBytesConfig.isIncludeOriginal()) {
-            EmbeddedDocumentBytesHandler embeddedDocumentByteStore = parseContext.get(EmbeddedDocumentBytesHandler.class);
+
+        if (t.getEmbeddedDocumentBytesConfig() != null &&
+                t.getEmbeddedDocumentBytesConfig().isIncludeOriginal()) {
+            EmbeddedDocumentBytesHandler embeddedDocumentByteStore =
+                    parseContext.get(EmbeddedDocumentBytesHandler.class);
             try (InputStream is = Files.newInputStream(tis.getPath())) {
                 embeddedDocumentByteStore.add(0, metadata, is);
             } catch (IOException e) {

tika-core/src/main/java/org/apache/tika/pipes/fetcher/FetcherManager.java

@@ -25,6 +25,9 @@
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import org.apache.tika.config.ConfigBase;
 import org.apache.tika.exception.TikaConfigException;
 import org.apache.tika.exception.TikaException;
@@ -35,7 +38,7 @@
  * This forbids multiple fetchers supporting the same name.
  */
 public class FetcherManager extends ConfigBase {
-
+    private static final Logger LOG = LoggerFactory.getLogger(FetcherManager.class);
     public static FetcherManager load(Path p) throws IOException, TikaConfigException {
         try (InputStream is =
                      Files.newInputStream(p)) {
@@ -48,12 +51,12 @@ public static FetcherManager load(Path p) throws IOException, TikaConfigExceptio
     public FetcherManager(List<Fetcher> fetchers) throws TikaConfigException {
         for (Fetcher fetcher : fetchers) {
             String name = fetcher.getName();
-            if (name == null || name.trim().length() == 0) {
-                throw new TikaConfigException("fetcher name must not be blank");
+            if (name == null || name.trim().isEmpty()) {
+                throw new TikaConfigException("Fetcher name must not be blank");
             }
             if (fetcherMap.containsKey(fetcher.getName())) {
-                throw new TikaConfigException(
-                        "Multiple fetchers cannot support the same prefix: " + fetcher.getName());
+                LOG.warn("Duplicate fetcher saved in the tika-config xml: {}. Ignoring.", fetcher.getName());
+                continue;
             }
             fetcherMap.put(fetcher.getName(), fetcher);
         }

tika-core/src/main/java/org/apache/tika/pipes/fetcher/config/AbstractConfig.java

@@ -0,0 +1,21 @@
+/*
+ * 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.fetcher.config;
+
+public abstract class AbstractConfig {
+    // Nothing to do here yet.
+}

tika-core/src/main/java/org/apache/tika/pipes/fetcher/fs/FileSystemFetcher.java

@@ -43,8 +43,16 @@
 import org.apache.tika.metadata.TikaCoreProperties;
 import org.apache.tika.parser.ParseContext;
 import org.apache.tika.pipes.fetcher.AbstractFetcher;
+import org.apache.tika.pipes.fetcher.fs.config.FileSystemFetcherConfig;
 
 public class FileSystemFetcher extends AbstractFetcher implements Initializable {
+    public FileSystemFetcher() {
+    }
+
+    public FileSystemFetcher(FileSystemFetcherConfig fileSystemFetcherConfig) {
+        setBasePath(fileSystemFetcherConfig.getBasePath());
+        setExtractFileSystemMetadata(fileSystemFetcherConfig.isExtractFileSystemMetadata());
+    }
 
     private static final Logger LOG = LoggerFactory.getLogger(FileSystemFetcher.class);
 

tika-core/src/main/java/org/apache/tika/pipes/fetcher/fs/config/FileSystemFetcherConfig.java

@@ -0,0 +1,42 @@
+/*
+ * 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.fetcher.fs.config;
+
+import org.apache.tika.pipes.fetcher.config.AbstractConfig;
+
+public class FileSystemFetcherConfig extends AbstractConfig {
+    private String basePath;
+    private boolean extractFileSystemMetadata;
+
+    public String getBasePath() {
+        return basePath;
+    }
+
+    public FileSystemFetcherConfig setBasePath(String basePath) {
+        this.basePath = basePath;
+        return this;
+    }
+
+    public boolean isExtractFileSystemMetadata() {
+        return extractFileSystemMetadata;
+    }
+
+    public FileSystemFetcherConfig setExtractFileSystemMetadata(boolean extractFileSystemMetadata) {
+        this.extractFileSystemMetadata = extractFileSystemMetadata;
+        return this;
+    }
+}