You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tika.apache.org by ta...@apache.org on 2021/01/26 19:46:56 UTC
[tika] branch TIKA-3226 updated: TIKA-3226 -- WIP don't merge
This is an automated email from the ASF dual-hosted git repository.
tallison pushed a commit to branch TIKA-3226
in repository https://gitbox.apache.org/repos/asf/tika.git
The following commit(s) were added to refs/heads/TIKA-3226 by this push:
new 1dd15d9 TIKA-3226 -- WIP don't merge
1dd15d9 is described below
commit 1dd15d93c45ee95761bd28306da1ba07ba751503
Author: tballison <ta...@apache.org>
AuthorDate: Tue Jan 26 14:46:40 2021 -0500
TIKA-3226 -- WIP don't merge
---
pom.xml | 3 +-
.../main/java/org/apache/tika/config/Param.java | 8 +-
.../java/org/apache/tika/config/TikaConfig.java | 527 ++++++++++++---------
.../main/java/org/apache/tika/emitter/Emitter.java | 20 -
.../apache/tika/emitter/TikaEmitterException.java | 9 -
.../apache/tika/fetcher/DefaultFetchIterator.java | 57 ---
.../org/apache/tika/fetcher/FetchIterator.java | 141 ------
.../apache/tika/fetcher/FetchPrefixKeyPair.java | 53 ---
.../org/apache/tika/fetcher/FileSystemFetcher.java | 79 ---
.../emitter/AbstractEmitter.java} | 20 +-
.../emitter/Emitter.java} | 21 +-
.../emitter/EmitterManager.java} | 40 +-
.../emitter/EmptyEmitter.java} | 31 +-
.../emitter/TikaEmitterException.java} | 10 +-
.../fetcher/AbstractFetcher.java} | 31 +-
.../fetcher/EmptyFetcher.java} | 21 +-
.../org/apache/tika/pipes/fetcher/FetchId.java | 65 +++
.../fetcher/FetchIdMetadataPair.java} | 40 +-
.../apache/tika/{ => pipes}/fetcher/Fetcher.java | 12 +-
.../fetcher/FetcherManager.java} | 50 +-
.../fetcher/FetcherStringException.java | 2 +-
.../tika/pipes/fetcher/FileSystemFetcher.java | 110 +++++
.../tika/{ => pipes}/fetcher/SimpleUrlFetcher.java | 18 +-
.../fetchiterator/EmptyFetchIterator.java} | 22 +-
.../tika/pipes/fetchiterator/FetchIterator.java | 135 ++++++
.../fetchiterator}/FileSystemFetchIterator.java | 66 +--
.../org/apache/tika/config/TikaConfigTest.java | 67 +++
.../apache/tika/pipes/emitter/MockEmitter.java} | 16 +-
.../tika/pipes/fetcher/FileSystemFetcherTest.java | 54 +++
.../FileSystemFetchIteratorTest.java | 40 +-
.../org/apache/tika/config/emitters-config.xml | 18 +-
.../tika/config/emitters-duplicate-config.xml | 18 +-
.../apache/tika/config/fetch-iterator-config.xml | 14 +-
.../tika/config/fetch-iterator-multiple-config.xml | 20 +-
.../org/apache/tika/config/fetchers-config.xml | 14 +-
.../tika/config/fetchers-duplicate-config.xml | 14 +-
.../tika/config/fetchers-nobasepath-config.xml | 13 +-
.../apache/tika/config/fetchers-noname-config.xml | 13 +-
tika-fetchers/tika-fetcher-jdbc/pom.xml | 122 -----
.../org/apache/tika/fetcher/jdbc/JDBCFetcher.java | 100 ----
tika-pipes/pom.xml | 41 ++
.../tika-emitters}/pom.xml | 6 +-
.../tika-emitters}/tika-emitter-fs/pom.xml | 0
.../apache/tika/emitter/fs/FileSystemEmitter.java | 26 +-
.../tika-emitters}/tika-emitter-solr/pom.xml | 0
.../org/apache/tika/emitter/solr/SolrEmitter.java | 33 +-
.../org/apache/tika/emitter/solr/TestBasic.java | 25 +-
.../src/test/resources/log4j.properties | 0
.../test/resources/tika-config-simple-emitter.xml | 0
.../tika-fetch-iterators}/pom.xml | 21 +-
.../tika-fetch-iterator-csv}/pom.xml | 23 +-
.../pipes/fetchiterator/csv/CSVFetchIterator.java | 121 +++++
.../src/test/java/TestCSVFetchIterator.java | 129 +++++
.../src/test/resources/test-simple.csv | 6 +
.../tika-fetch-iterator-jdbc}/pom.xml | 24 +-
.../fetchiterator/jdbc/JDBCFetchIterator.java | 166 +++++++
.../fetchiterator/jdbc/TestJDBCFetchIterator.java | 178 +++++++
.../src/test/resources/log4j.properties | 4 +-
.../tika-fetch-iterator-s3}/pom.xml | 53 ++-
.../pipes/fetchiterator/s3/S3FetchIterator.java | 109 +++++
.../fetchiterator/s3/TestS3FetchIterator.java | 98 ++++
.../src/test/resources/log4j.properties | 4 +-
.../tika-fetchers}/pom.xml | 5 +-
.../tika-fetchers}/tika-fetcher-s3/pom.xml | 6 +-
.../apache/tika/pipes}/fetcher/s3/S3Fetcher.java | 73 +--
.../tika/pipes}/fetcher/s3/TestS3Fetcher.java | 7 +-
.../src/test/resources/tika-config-s3.xml | 1 +
.../tika-httpclient-commons/pom.xml | 3 +-
.../org/apache/tika/client/HttpClientUtil.java | 0
.../apache/tika/client/TikaClientException.java | 0
.../tika-pipes-integration-tests}/pom.xml | 33 +-
.../apache/tika/pipes/PipeIntegrationTests.java | 162 +++++++
.../src/test/resources/log4j.properties | 4 +-
.../src/test/resources/tika-config-s3ToFs.xml | 22 +
.../apache/tika/server/classic/FetcherTest.java | 5 +-
.../resources/config/tika-config-url-fetcher.xml | 6 +-
.../org/apache/tika/server/client/TikaClient.java | 13 +-
.../apache/tika/server/client/TikaClientCLI.java | 78 +--
.../resources/tika-config-simple-solr-emitter.xml | 4 +-
.../tika/server/core/FetcherStreamFactory.java | 24 +-
.../org/apache/tika/server/core/TikaServerCli.java | 15 +-
.../tika/server/core/resource/EmitterResource.java | 83 ++--
.../apache/tika/server/core/TikaEmitterTest.java | 21 +-
.../core/TikaServerEmitterIntegrationTest.java | 13 +-
.../resources/test-documents/mock/hello_world.xml | 1 +
85 files changed, 2424 insertions(+), 1336 deletions(-)
diff --git a/pom.xml b/pom.xml
index a31f4fe..37389a1 100644
--- a/pom.xml
+++ b/pom.xml
@@ -37,8 +37,7 @@
<modules>
<module>tika-parent</module>
<module>tika-core</module>
- <module>tika-emitters</module>
- <module>tika-fetchers</module>
+ <module>tika-pipes</module>
<module>tika-parsers</module>
<module>tika-bundles</module>
<module>tika-xmp</module>
diff --git a/tika-core/src/main/java/org/apache/tika/config/Param.java b/tika-core/src/main/java/org/apache/tika/config/Param.java
index 3462252..74123ac 100644
--- a/tika-core/src/main/java/org/apache/tika/config/Param.java
+++ b/tika-core/src/main/java/org/apache/tika/config/Param.java
@@ -218,16 +218,16 @@ public class Param<T> implements Serializable {
if (valueAttr != null && (value == null || value.getTextContent() == null)) {
value = valueAttr;
}
- if (typeAttr == null) {
- throw new TikaConfigException("Must specify a \"type\" in: " + node.getLocalName());
- }
-
+
Param<T> ret = new Param<T>();
ret.name = nameAttr.getTextContent();
if (typeAttr != null) {
ret.setTypeString(typeAttr.getTextContent());
} else {
ret.type = (Class<T>)wellKnownMap.get(ret.name);
+ if (ret.type == null) {
+ throw new TikaConfigException("Must specify a \"type\" in: " + node.getLocalName());
+ }
}
if (List.class.isAssignableFrom(ret.type)) {
diff --git a/tika-core/src/main/java/org/apache/tika/config/TikaConfig.java b/tika-core/src/main/java/org/apache/tika/config/TikaConfig.java
index df4c43c..f852c57 100644
--- a/tika-core/src/main/java/org/apache/tika/config/TikaConfig.java
+++ b/tika-core/src/main/java/org/apache/tika/config/TikaConfig.java
@@ -46,14 +46,14 @@ import org.apache.tika.detect.DefaultDetector;
import org.apache.tika.detect.DefaultEncodingDetector;
import org.apache.tika.detect.Detector;
import org.apache.tika.detect.EncodingDetector;
-import org.apache.tika.emitter.DefaultEmitter;
-import org.apache.tika.emitter.Emitter;
+import org.apache.tika.pipes.emitter.Emitter;
import org.apache.tika.exception.TikaConfigException;
import org.apache.tika.exception.TikaException;
-import org.apache.tika.fetcher.DefaultFetchIterator;
-import org.apache.tika.fetcher.DefaultFetcher;
-import org.apache.tika.fetcher.FetchIterator;
-import org.apache.tika.fetcher.Fetcher;
+import org.apache.tika.pipes.emitter.EmitterManager;
+import org.apache.tika.pipes.fetcher.FetcherManager;
+import org.apache.tika.pipes.fetchiterator.EmptyFetchIterator;
+import org.apache.tika.pipes.fetchiterator.FetchIterator;
+import org.apache.tika.pipes.fetcher.Fetcher;
import org.apache.tika.language.translate.DefaultTranslator;
import org.apache.tika.language.translate.Translator;
import org.apache.tika.metadata.filter.CompositeMetadataFilter;
@@ -73,8 +73,6 @@ import org.apache.tika.parser.ParserDecorator;
import org.apache.tika.parser.multiple.AbstractMultipleParser;
import org.apache.tika.utils.AnnotationUtils;
import org.apache.tika.utils.XMLReaderUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import org.w3c.dom.Document;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
@@ -120,16 +118,16 @@ public class TikaConfig {
return new DefaultMetadataFilter(loader);
}
- private static Fetcher getDefaultFetcher(ServiceLoader loader) {
- return new DefaultFetcher(Collections.EMPTY_LIST);
+ private static FetcherManager getDefaultFetcherManager(ServiceLoader loader) {
+ return new FetcherManager(Collections.EMPTY_LIST);
}
- private static Emitter getDefaultEmitter(ServiceLoader loader) {
- return new DefaultEmitter(Collections.EMPTY_LIST);
+ private static EmitterManager getDefaultEmitterManager(ServiceLoader loader) {
+ return new EmitterManager(Collections.EMPTY_LIST);
}
private static FetchIterator getDefaultFetchIterator(ServiceLoader loader) {
- return new DefaultFetchIterator(Collections.EMPTY_LIST);
+ return new EmptyFetchIterator();
}
//use this to look for unneeded instantiations of TikaConfig
@@ -144,8 +142,8 @@ public class TikaConfig {
private final ExecutorService executorService;
private final EncodingDetector encodingDetector;
private final MetadataFilter metadataFilter;
- private final Fetcher fetcher;
- private final Emitter emitter;
+ private final FetcherManager fetcherManager;
+ private final EmitterManager emitterManager;
private final FetchIterator fetchIterator;
public TikaConfig(String file)
@@ -157,6 +155,7 @@ public class TikaConfig {
throws TikaException, IOException, SAXException {
this(XMLReaderUtils.buildDOM(path));
}
+
public TikaConfig(Path path, ServiceLoader loader)
throws TikaException, IOException, SAXException {
this(XMLReaderUtils.buildDOM(path), loader);
@@ -176,10 +175,12 @@ public class TikaConfig {
throws TikaException, IOException, SAXException {
this(url, ServiceLoader.getContextClassLoader());
}
+
public TikaConfig(URL url, ClassLoader loader)
throws TikaException, IOException, SAXException {
this(XMLReaderUtils.buildDOM(url.toString()).getDocumentElement(), loader);
}
+
public TikaConfig(URL url, ServiceLoader loader)
throws TikaException, IOException, SAXException {
this(XMLReaderUtils.buildDOM(url.toString()).getDocumentElement(), loader);
@@ -193,6 +194,7 @@ public class TikaConfig {
public TikaConfig(Document document) throws TikaException, IOException {
this(document.getDocumentElement());
}
+
public TikaConfig(Document document, ServiceLoader loader) throws TikaException, IOException {
this(document.getDocumentElement(), loader);
}
@@ -226,8 +228,8 @@ public class TikaConfig {
this.translator = translatorLoader.loadOverall(element, mimeTypes, loader);
this.executorService = executorLoader.loadOverall(element, mimeTypes, loader);
this.metadataFilter = metadataFilterXmlLoader.loadOverall(element, mimeTypes, loader);
- this.fetcher = fetcherXmlLoader.loadOverall(element, mimeTypes, loader);
- this.emitter = emitterXmlLoader.loadOverall(element, mimeTypes, loader);
+ this.fetcherManager = fetcherXmlLoader.loadOverall(element, mimeTypes, loader);
+ this.emitterManager = emitterXmlLoader.loadOverall(element, mimeTypes, loader);
this.fetchIterator = fetchIteratorXmlLoader.loadOverall(element, mimeTypes, loader);
this.serviceLoader = loader;
TIMES_INSTANTIATED.incrementAndGet();
@@ -239,11 +241,11 @@ public class TikaConfig {
* {@link ServiceRegistry service provider mechanism} in the given
* class loader.
*
- * @since Apache Tika 0.8
* @param loader the class loader through which parser implementations
* are loaded, or <code>null</code> for no parsers
* @throws MimeTypeException if the built-in media type rules are broken
- * @throws IOException if the built-in media type rules can not be read
+ * @throws IOException if the built-in media type rules can not be read
+ * @since Apache Tika 0.8
*/
public TikaConfig(ClassLoader loader)
throws MimeTypeException, IOException {
@@ -255,8 +257,8 @@ public class TikaConfig {
this.translator = getDefaultTranslator(serviceLoader);
this.executorService = getDefaultExecutorService();
this.metadataFilter = getDefaultMetadataFilter(serviceLoader);
- this.fetcher = getDefaultFetcher(serviceLoader);
- this.emitter = getDefaultEmitter(serviceLoader);
+ this.fetcherManager = getDefaultFetcherManager(serviceLoader);
+ this.emitterManager = getDefaultEmitterManager(serviceLoader);
this.fetchIterator = getDefaultFetchIterator(serviceLoader);
TIMES_INSTANTIATED.incrementAndGet();
}
@@ -275,7 +277,7 @@ public class TikaConfig {
* the {@link ServiceRegistry service provider mechanism} in the context
* class loader of the current thread.</p>
*
- * @throws IOException if the configuration can not be read
+ * @throws IOException if the configuration can not be read
* @throws TikaException if problem with MimeTypes or parsing XML config
*/
public TikaConfig() throws TikaException, IOException {
@@ -294,8 +296,8 @@ public class TikaConfig {
this.translator = getDefaultTranslator(serviceLoader);
this.executorService = getDefaultExecutorService();
this.metadataFilter = getDefaultMetadataFilter(serviceLoader);
- this.fetcher = getDefaultFetcher(serviceLoader);
- this.emitter = getDefaultEmitter(serviceLoader);
+ this.fetcherManager = getDefaultFetcherManager(serviceLoader);
+ this.emitterManager = getDefaultEmitterManager(serviceLoader);
this.fetchIterator = getDefaultFetchIterator(serviceLoader);
} else {
ServiceLoader tmpServiceLoader = new ServiceLoader();
@@ -322,8 +324,8 @@ public class TikaConfig {
this.translator = translatorLoader.loadOverall(element, mimeTypes, serviceLoader);
this.executorService = executorLoader.loadOverall(element, mimeTypes, serviceLoader);
this.metadataFilter = metadataFilterXmlLoader.loadOverall(element, mimeTypes, serviceLoader);
- this.fetcher = fetcherXmlLoader.loadOverall(element, mimeTypes, serviceLoader);
- this.emitter = emitterXmlLoader.loadOverall(element, mimeTypes, serviceLoader);
+ this.fetcherManager = fetcherXmlLoader.loadOverall(element, mimeTypes, serviceLoader);
+ this.emitterManager = emitterXmlLoader.loadOverall(element, mimeTypes, serviceLoader);
this.fetchIterator = fetchIteratorXmlLoader.loadOverall(element, mimeTypes, serviceLoader);
} catch (SAXException e) {
throw new TikaException(
@@ -418,6 +420,7 @@ public class TikaConfig {
/**
* Returns the configured encoding detector instance
+ *
* @return configured encoding detector
*/
public EncodingDetector getEncodingDetector() {
@@ -432,19 +435,19 @@ public class TikaConfig {
public Translator getTranslator() {
return translator;
}
-
+
public ExecutorService getExecutorService() {
return executorService;
}
- public MimeTypes getMimeRepository(){
+ public MimeTypes getMimeRepository() {
return mimeTypes;
}
public MediaTypeRegistry getMediaTypeRegistry() {
return mimeTypes.getMediaTypeRegistry();
}
-
+
public ServiceLoader getServiceLoader() {
return serviceLoader;
}
@@ -453,17 +456,18 @@ public class TikaConfig {
return metadataFilter;
}
- public Fetcher getFetcher() {
- return fetcher;
+ public FetcherManager getFetcherManager() {
+ return fetcherManager;
}
- public Emitter getEmitter() {
- return emitter;
+ public EmitterManager getEmitterManager() {
+ return emitterManager;
}
public FetchIterator getFetchIterator() {
return fetchIterator;
}
+
/**
* Provides a default configuration (TikaConfig). Currently creates a
* new instance each time it's called; we may be able to have it
@@ -494,23 +498,23 @@ public class TikaConfig {
}
return null;
}
- private static List<Element> getTopLevelElementChildren(Element element,
- String parentName, String childrenName) throws TikaException {
+
+ private static List<Element> getTopLevelElementChildren(Element element,
+ String parentName, String childrenName) throws TikaException {
Node parentNode = null;
if (parentName != null) {
// Should be only zero or one <parsers> / <detectors> etc tag
NodeList nodes = element.getElementsByTagName(parentName);
if (nodes.getLength() > 1) {
- throw new TikaException("Properties may not contain multiple "+parentName+" entries");
- }
- else if (nodes.getLength() == 1) {
+ throw new TikaException("Properties may not contain multiple " + parentName + " entries");
+ } else if (nodes.getLength() == 1) {
parentNode = nodes.item(0);
}
} else {
// All children directly on the master element
parentNode = element;
}
-
+
if (parentNode != null) {
// Find only the direct child parser/detector objects
NodeList nodes = parentNode.getChildNodes();
@@ -518,7 +522,7 @@ public class TikaConfig {
for (int i = 0; i < nodes.getLength(); i++) {
Node node = nodes.item(i);
if (node instanceof Element) {
- Element nodeE = (Element)node;
+ Element nodeE = (Element) node;
if (childrenName.equals(nodeE.getTagName())) {
elements.add(nodeE);
}
@@ -540,16 +544,16 @@ public class TikaConfig {
return getDefaultMimeTypes(null);
}
}
-
+
private static Set<MediaType> mediaTypesListFromDomElement(
- Element node, String tag)
+ Element node, String tag)
throws TikaException, IOException {
Set<MediaType> types = null;
NodeList children = node.getChildNodes();
- for (int i=0; i<children.getLength(); i++) {
+ for (int i = 0; i < children.getLength(); i++) {
Node cNode = children.item(i);
if (cNode instanceof Element) {
- Element cElement = (Element)cNode;
+ Element cElement = (Element) cNode;
if (tag.equals(cElement.getTagName())) {
String mime = getText(cElement);
MediaType type = MediaType.parse(mime);
@@ -566,7 +570,7 @@ public class TikaConfig {
if (types != null) return types;
return Collections.emptySet();
}
-
+
private static ServiceLoader serviceLoaderFromDomElement(Element element, ClassLoader loader) throws TikaConfigException {
Element serviceLoaderElement = getChild(element, "service-loader");
ServiceLoader serviceLoader;
@@ -575,9 +579,9 @@ public class TikaConfig {
boolean dynamic = Boolean.parseBoolean(serviceLoaderElement.getAttribute("dynamic"));
LoadErrorHandler loadErrorHandler = LoadErrorHandler.IGNORE;
String loadErrorHandleConfig = serviceLoaderElement.getAttribute("loadErrorHandler");
- if(LoadErrorHandler.WARN.toString().equalsIgnoreCase(loadErrorHandleConfig)) {
+ if (LoadErrorHandler.WARN.toString().equalsIgnoreCase(loadErrorHandleConfig)) {
loadErrorHandler = LoadErrorHandler.WARN;
- } else if(LoadErrorHandler.THROW.toString().equalsIgnoreCase(loadErrorHandleConfig)) {
+ } else if (LoadErrorHandler.THROW.toString().equalsIgnoreCase(loadErrorHandleConfig)) {
loadErrorHandler = LoadErrorHandler.THROW;
}
InitializableProblemHandler initializableProblemHandler = getInitializableProblemHandler(serviceLoaderElement.getAttribute("initializableProblemHandler"));
@@ -586,7 +590,7 @@ public class TikaConfig {
loader = ServiceLoader.getContextClassLoader();
}
serviceLoader = new ServiceLoader(loader, loadErrorHandler, initializableProblemHandler, dynamic);
- } else if(loader != null) {
+ } else if (loader != null) {
serviceLoader = new ServiceLoader(loader);
} else {
serviceLoader = new ServiceLoader();
@@ -595,7 +599,7 @@ public class TikaConfig {
}
private static InitializableProblemHandler getInitializableProblemHandler(String initializableProblemHandler)
- throws TikaConfigException {
+ throws TikaConfigException {
if (initializableProblemHandler == null || initializableProblemHandler.length() == 0) {
return InitializableProblemHandler.DEFAULT;
}
@@ -614,36 +618,46 @@ public class TikaConfig {
}
- private static abstract class XmlLoader<CT,T> {
+ private static abstract class XmlLoader<CT, T> {
protected static final String PARAMS_TAG_NAME = "params";
+
abstract boolean supportsComposite();
+
abstract String getParentTagName(); // eg parsers
+
abstract String getLoaderTagName(); // eg parser
+
abstract Class<? extends T> getLoaderClass(); // Generics workaround
+
abstract boolean isComposite(T loaded);
+
abstract boolean isComposite(Class<? extends T> loadedClass);
- abstract T preLoadOne(Class<? extends T> loadedClass, String classname,
- MimeTypes mimeTypes) throws TikaException;
+
+ abstract T preLoadOne(Class<? extends T> loadedClass, String classname,
+ MimeTypes mimeTypes) throws TikaException;
+
abstract CT createDefault(MimeTypes mimeTypes, ServiceLoader loader);
+
abstract CT createComposite(List<T> loaded, MimeTypes mimeTypes, ServiceLoader loader);
- abstract T createComposite(Class<? extends T> compositeClass,
- List<T> children, Set<Class<? extends T>> excludeChildren,
- Map<String, Param> params, MimeTypes mimeTypes, ServiceLoader loader)
+
+ abstract T createComposite(Class<? extends T> compositeClass,
+ List<T> children, Set<Class<? extends T>> excludeChildren,
+ Map<String, Param> params, MimeTypes mimeTypes, ServiceLoader loader)
throws InvocationTargetException, IllegalAccessException, InstantiationException;
- abstract T decorate(T created, Element element)
+
+ abstract T decorate(T created, Element element)
throws IOException, TikaException; // eg explicit mime types
-
+
@SuppressWarnings("unchecked")
- CT loadOverall(Element element, MimeTypes mimeTypes,
- ServiceLoader loader) throws TikaException, IOException {
+ CT loadOverall(Element element, MimeTypes mimeTypes,
+ ServiceLoader loader) throws TikaException, IOException {
List<T> loaded = new ArrayList<T>();
-
+
// Find the children of the parent tag, if any
- for (Element le : getTopLevelElementChildren(element, getParentTagName(), getLoaderTagName())) {
- T loadedChild = loadOne(le, mimeTypes, loader);
+ for (Element le : getTopLevelElementChildren(element, getParentTagName(), getLoaderTagName())) {T loadedChild = loadOne(le, mimeTypes, loader);
if (loadedChild != null) loaded.add(loadedChild);
}
-
+
// Build the classes, and wrap as needed
if (loaded.isEmpty()) {
// Nothing defined, create a Default
@@ -652,28 +666,36 @@ public class TikaConfig {
T single = loaded.get(0);
if (isComposite(single)) {
// Single Composite defined, use that
- return (CT)single;
+ return (CT) single;
}
- } else if (! supportsComposite()) {
+ } else if (!supportsComposite()) {
// No composite support, just return the first one
- return (CT)loaded.get(0);
+ if (loaded.size() == 1) {
+ return (CT) loaded.get(0);
+ } else if (loaded.size() > 1) {
+ throw new TikaConfigException("Composite not supported for " +
+ getParentTagName() +
+ ". Must specify only one child!");
+ } else {
+ //throw exception if empty?
+ }
}
// Wrap the defined parsers/detectors up in a Composite
return createComposite(loaded, mimeTypes, loader);
}
- T loadOne(Element element, MimeTypes mimeTypes, ServiceLoader loader)
+ T loadOne(Element element, MimeTypes mimeTypes, ServiceLoader loader)
throws TikaException, IOException {
String name = element.getAttribute("class");
if (name == null) {
- throw new TikaConfigException("class attribute must not be null: "+element);
+ throw new TikaConfigException("class attribute must not be null: " + element);
}
String initProbHandler = element.getAttribute("initializableProblemHandler");
InitializableProblemHandler initializableProblemHandler;
if (initProbHandler == null || initProbHandler.length() == 0) {
initializableProblemHandler = loader.getInitializableProblemHandler();
} else {
- initializableProblemHandler =
+ initializableProblemHandler =
getInitializableProblemHandler(initProbHandler);
}
@@ -687,7 +709,7 @@ public class TikaConfig {
//TODO : allow duplicate instances with different configurations
loaded = preLoadOne(loadedClass, name, mimeTypes);
if (loaded != null) return loaded;
-
+
// Get any parameters / settings for the parser
Map<String, Param> params = null;
try {
@@ -695,7 +717,7 @@ public class TikaConfig {
} catch (Exception e) {
throw new TikaConfigException(e.getMessage(), e);
}
-
+
// Is this a composite or decorated class? If so, support recursion
if (isComposite(loadedClass)) {
// Get the child objects for it
@@ -703,28 +725,28 @@ public class TikaConfig {
NodeList childNodes = element.getElementsByTagName(getLoaderTagName());
if (childNodes.getLength() > 0) {
for (int i = 0; i < childNodes.getLength(); i++) {
- T loadedChild = loadOne((Element)childNodes.item(i),
- mimeTypes, loader);
+ T loadedChild = loadOne((Element) childNodes.item(i),
+ mimeTypes, loader);
if (loadedChild != null) children.add(loadedChild);
}
}
-
+
// Get the list of children to exclude
Set<Class<? extends T>> excludeChildren = new HashSet<Class<? extends T>>();
- NodeList excludeChildNodes = element.getElementsByTagName(getLoaderTagName()+"-exclude");
+ NodeList excludeChildNodes = element.getElementsByTagName(getLoaderTagName() + "-exclude");
if (excludeChildNodes.getLength() > 0) {
for (int i = 0; i < excludeChildNodes.getLength(); i++) {
- Element excl = (Element)excludeChildNodes.item(i);
+ Element excl = (Element) excludeChildNodes.item(i);
String exclName = excl.getAttribute("class");
try {
excludeChildren.add(loader.getServiceClass(getLoaderClass(), exclName));
} catch (ClassNotFoundException e) {
//TIKA-3268 -- This should stop the world.
- throw new TikaConfigException("Class now found in -exclude list: " +exclName);
+ throw new TikaConfigException("Class now found in -exclude list: " + exclName);
}
}
}
-
+
// Create the Composite
loaded = createComposite(loadedClass, children, excludeChildren, params, mimeTypes, loader);
@@ -752,24 +774,24 @@ public class TikaConfig {
} catch (ClassNotFoundException e) {
if (loader.getLoadErrorHandler() == LoadErrorHandler.THROW) {
// Use a different exception signature here
- throw new TikaException(
- "Unable to find a "+getLoaderTagName()+" class: " + name, e);
+ throw new TikaConfigException(
+ "Unable to find a " + getLoaderTagName() + " class: " + name, e);
}
// Report the problem
loader.getLoadErrorHandler().handleLoadError(name, e);
return null;
} catch (IllegalAccessException e) {
throw new TikaException(
- "Unable to access a "+getLoaderTagName()+" class: " + name, e);
+ "Unable to access a " + getLoaderTagName() + " class: " + name, e);
} catch (InvocationTargetException e) {
throw new TikaException(
- "Unable to create a "+getLoaderTagName()+" class: " + name, e);
+ "Unable to create a " + getLoaderTagName() + " class: " + name, e);
} catch (InstantiationException e) {
throw new TikaException(
- "Unable to instantiate a "+getLoaderTagName()+" class: " + name, e);
+ "Unable to instantiate a " + getLoaderTagName() + " class: " + name, e);
} catch (NoSuchMethodException e) {
throw new TikaException(
- "Unable to find the right constructor for "+getLoaderTagName()+" class: " + name, e);
+ "Unable to find the right constructor for " + getLoaderTagName() + " class: " + name, e);
}
}
@@ -782,19 +804,20 @@ public class TikaConfig {
/**
* Gets parameters from a given
+ *
* @param el xml node which has {@link #PARAMS_TAG_NAME} child
* @return Map of key values read from xml
*/
- Map<String, Param> getParams(Element el) throws TikaException {
+ Map<String, Param> getParams(Element el) throws TikaException {
Map<String, Param> params = new HashMap<>();
for (Node child = el.getFirstChild(); child != null;
- child = child.getNextSibling()){
- if (PARAMS_TAG_NAME.equals(child.getNodeName())){ //found the node
+ child = child.getNextSibling()) {
+ if (PARAMS_TAG_NAME.equals(child.getNodeName())) { //found the node
if (child.hasChildNodes()) {//it has children
NodeList childNodes = child.getChildNodes();
for (int i = 0; i < childNodes.getLength(); i++) {
Node item = childNodes.item(i);
- if (item.getNodeType() == Node.ELEMENT_NODE){
+ if (item.getNodeType() == Node.ELEMENT_NODE) {
Param<?> param = Param.load(item);
params.put(param.getName(), param);
}
@@ -807,113 +830,133 @@ public class TikaConfig {
}
}
- private static class ParserXmlLoader extends XmlLoader<CompositeParser,Parser> {
+
+ private static class ParserXmlLoader extends XmlLoader<CompositeParser, Parser> {
private final EncodingDetector encodingDetector;
- boolean supportsComposite() { return true; }
- String getParentTagName() { return "parsers"; }
- String getLoaderTagName() { return "parser"; }
+ boolean supportsComposite() {
+ return true;
+ }
+
+ String getParentTagName() {
+ return "parsers";
+ }
+
+ String getLoaderTagName() {
+ return "parser";
+ }
private ParserXmlLoader(EncodingDetector encodingDetector) {
this.encodingDetector = encodingDetector;
}
+
@Override
Class<? extends Parser> getLoaderClass() {
return Parser.class;
}
+
@Override
- Parser preLoadOne(Class<? extends Parser> loadedClass, String classname,
+ Parser preLoadOne(Class<? extends Parser> loadedClass, String classname,
MimeTypes mimeTypes) throws TikaException {
// Check for classes which can't be set in config
if (AutoDetectParser.class.isAssignableFrom(loadedClass)) {
// https://issues.apache.org/jira/browse/TIKA-866
throw new TikaException(
"AutoDetectParser not supported in a <parser>"
- + " configuration element: " + classname);
+ + " configuration element: " + classname);
}
// Continue with normal loading
return null;
}
+
@Override
boolean isComposite(Parser loaded) {
return loaded instanceof CompositeParser;
}
+
@Override
boolean isComposite(Class<? extends Parser> loadedClass) {
if (CompositeParser.class.isAssignableFrom(loadedClass) ||
- AbstractMultipleParser.class.isAssignableFrom(loadedClass) ||
- ParserDecorator.class.isAssignableFrom(loadedClass)) {
+ AbstractMultipleParser.class.isAssignableFrom(loadedClass) ||
+ ParserDecorator.class.isAssignableFrom(loadedClass)) {
return true;
}
return false;
}
+
@Override
CompositeParser createDefault(MimeTypes mimeTypes, ServiceLoader loader) {
return getDefaultParser(mimeTypes, loader, encodingDetector);
}
+
@Override
CompositeParser createComposite(List<Parser> parsers, MimeTypes mimeTypes, ServiceLoader loader) {
MediaTypeRegistry registry = mimeTypes.getMediaTypeRegistry();
return new CompositeParser(registry, parsers);
}
+
@Override
Parser createComposite(Class<? extends Parser> parserClass,
- List<Parser> childParsers, Set<Class<? extends Parser>> excludeParsers,
- Map<String, Param> params, MimeTypes mimeTypes, ServiceLoader loader)
+ List<Parser> childParsers, Set<Class<? extends Parser>> excludeParsers,
+ Map<String, Param> params, MimeTypes mimeTypes, ServiceLoader loader)
throws InvocationTargetException, IllegalAccessException, InstantiationException {
Parser parser = null;
Constructor<? extends Parser> c = null;
MediaTypeRegistry registry = mimeTypes.getMediaTypeRegistry();
-
+
// Try the possible default and composite parser constructors
if (parser == null) {
try {
c = parserClass.getConstructor(MediaTypeRegistry.class,
ServiceLoader.class, Collection.class, EncodingDetector.class);
parser = c.newInstance(registry, loader, excludeParsers, encodingDetector);
+ } catch (NoSuchMethodException me) {
}
- catch (NoSuchMethodException me) {}
}
if (parser == null) {
try {
c = parserClass.getConstructor(MediaTypeRegistry.class, ServiceLoader.class, Collection.class);
parser = c.newInstance(registry, loader, excludeParsers);
- }
- catch (NoSuchMethodException me) {}
+ } catch (NoSuchMethodException me) {
+ }
}
if (parser == null) {
try {
c = parserClass.getConstructor(MediaTypeRegistry.class, List.class, Collection.class);
parser = c.newInstance(registry, childParsers, excludeParsers);
- } catch (NoSuchMethodException me) {}
+ } catch (NoSuchMethodException me) {
+ }
}
if (parser == null) {
try {
c = parserClass.getConstructor(MediaTypeRegistry.class, Collection.class, Map.class);
parser = c.newInstance(registry, childParsers, params);
- } catch (NoSuchMethodException me) {}
+ } catch (NoSuchMethodException me) {
+ }
}
if (parser == null) {
try {
c = parserClass.getConstructor(MediaTypeRegistry.class, List.class);
parser = c.newInstance(registry, childParsers);
- } catch (NoSuchMethodException me) {}
+ } catch (NoSuchMethodException me) {
+ }
}
-
+
// Create as a Parser Decorator
if (parser == null && ParserDecorator.class.isAssignableFrom(parserClass)) {
try {
CompositeParser cp = null;
if (childParsers.size() == 1 && excludeParsers.size() == 0 &&
childParsers.get(0) instanceof CompositeParser) {
- cp = (CompositeParser)childParsers.get(0);
+ cp = (CompositeParser) childParsers.get(0);
} else {
cp = new CompositeParser(registry, childParsers, excludeParsers);
}
c = parserClass.getConstructor(Parser.class);
parser = c.newInstance(cp);
- } catch (NoSuchMethodException me) {}
+ } catch (NoSuchMethodException me) {
+ }
}
return parser;
}
@@ -931,34 +974,44 @@ public class TikaConfig {
@Override
Parser decorate(Parser created, Element element) throws IOException, TikaException {
Parser parser = created;
-
+
// Is there an explicit list of mime types for this to handle?
Set<MediaType> parserTypes = mediaTypesListFromDomElement(element, "mime");
- if (! parserTypes.isEmpty()) {
+ if (!parserTypes.isEmpty()) {
parser = ParserDecorator.withTypes(parser, parserTypes);
}
// Is there an explicit list of mime types this shouldn't handle?
Set<MediaType> parserExclTypes = mediaTypesListFromDomElement(element, "mime-exclude");
- if (! parserExclTypes.isEmpty()) {
+ if (!parserExclTypes.isEmpty()) {
parser = ParserDecorator.withoutTypes(parser, parserExclTypes);
}
-
+
// All done with decoration
return parser;
}
}
- private static class DetectorXmlLoader extends XmlLoader<CompositeDetector,Detector> {
- boolean supportsComposite() { return true; }
- String getParentTagName() { return "detectors"; }
- String getLoaderTagName() { return "detector"; }
-
+
+ private static class DetectorXmlLoader extends XmlLoader<CompositeDetector, Detector> {
+ boolean supportsComposite() {
+ return true;
+ }
+
+ String getParentTagName() {
+ return "detectors";
+ }
+
+ String getLoaderTagName() {
+ return "detector";
+ }
+
@Override
Class<? extends Detector> getLoaderClass() {
return Detector.class;
}
+
@Override
- Detector preLoadOne(Class<? extends Detector> loadedClass, String classname,
+ Detector preLoadOne(Class<? extends Detector> loadedClass, String classname,
MimeTypes mimeTypes) throws TikaException {
// If they asked for the mime types as a detector, give
// them the one we've already created. TIKA-1708
@@ -968,112 +1021,141 @@ public class TikaConfig {
// Continue with normal loading
return null;
}
+
@Override
boolean isComposite(Detector loaded) {
return loaded instanceof CompositeDetector;
}
+
@Override
boolean isComposite(Class<? extends Detector> loadedClass) {
return CompositeDetector.class.isAssignableFrom(loadedClass);
}
+
@Override
CompositeDetector createDefault(MimeTypes mimeTypes, ServiceLoader loader) {
return getDefaultDetector(mimeTypes, loader);
}
+
@Override
CompositeDetector createComposite(List<Detector> detectors, MimeTypes mimeTypes, ServiceLoader loader) {
MediaTypeRegistry registry = mimeTypes.getMediaTypeRegistry();
return new CompositeDetector(registry, detectors);
}
+
@Override
Detector createComposite(Class<? extends Detector> detectorClass,
- List<Detector> childDetectors,
- Set<Class<? extends Detector>> excludeDetectors,
- Map<String, Param> params, MimeTypes mimeTypes, ServiceLoader loader)
+ List<Detector> childDetectors,
+ Set<Class<? extends Detector>> excludeDetectors,
+ Map<String, Param> params, MimeTypes mimeTypes, ServiceLoader loader)
throws InvocationTargetException, IllegalAccessException,
InstantiationException {
Detector detector = null;
Constructor<? extends Detector> c;
MediaTypeRegistry registry = mimeTypes.getMediaTypeRegistry();
-
+
// Try the possible default and composite detector constructors
if (detector == null) {
try {
c = detectorClass.getConstructor(MimeTypes.class, ServiceLoader.class, Collection.class);
detector = c.newInstance(mimeTypes, loader, excludeDetectors);
- }
- catch (NoSuchMethodException me) {}
+ } catch (NoSuchMethodException me) {
+ }
}
if (detector == null) {
try {
c = detectorClass.getConstructor(MediaTypeRegistry.class, List.class, Collection.class);
detector = c.newInstance(registry, childDetectors, excludeDetectors);
- } catch (NoSuchMethodException me) {}
+ } catch (NoSuchMethodException me) {
+ }
}
if (detector == null) {
try {
c = detectorClass.getConstructor(MediaTypeRegistry.class, List.class);
detector = c.newInstance(registry, childDetectors);
- } catch (NoSuchMethodException me) {}
+ } catch (NoSuchMethodException me) {
+ }
}
if (detector == null) {
try {
c = detectorClass.getConstructor(List.class);
detector = c.newInstance(childDetectors);
- } catch (NoSuchMethodException me) {}
+ } catch (NoSuchMethodException me) {
+ }
}
-
+
return detector;
}
+
@Override
Detector decorate(Detector created, Element element) {
return created; // No decoration of Detectors
}
}
- private static class TranslatorXmlLoader extends XmlLoader<Translator,Translator> {
- boolean supportsComposite() { return false; }
- String getParentTagName() { return null; }
- String getLoaderTagName() { return "translator"; }
-
+
+ private static class TranslatorXmlLoader extends XmlLoader<Translator, Translator> {
+ boolean supportsComposite() {
+ return false;
+ }
+
+ String getParentTagName() {
+ return null;
+ }
+
+ String getLoaderTagName() {
+ return "translator";
+ }
+
@Override
Class<? extends Translator> getLoaderClass() {
return Translator.class;
}
+
@Override
- Translator preLoadOne(Class<? extends Translator> loadedClass, String classname,
+ Translator preLoadOne(Class<? extends Translator> loadedClass, String classname,
MimeTypes mimeTypes) throws TikaException {
// Continue with normal loading
return null;
}
+
@Override
- boolean isComposite(Translator loaded) { return false; }
+ boolean isComposite(Translator loaded) {
+ return false;
+ }
+
@Override
- boolean isComposite(Class<? extends Translator> loadedClass) { return false; }
+ boolean isComposite(Class<? extends Translator> loadedClass) {
+ return false;
+ }
+
@Override
Translator createDefault(MimeTypes mimeTypes, ServiceLoader loader) {
return getDefaultTranslator(loader);
}
+
@Override
Translator createComposite(List<Translator> loaded,
- MimeTypes mimeTypes, ServiceLoader loader) {
+ MimeTypes mimeTypes, ServiceLoader loader) {
return loaded.get(0);
}
+
@Override
Translator createComposite(Class<? extends Translator> compositeClass,
- List<Translator> children,
- Set<Class<? extends Translator>> excludeChildren,
- Map<String, Param> params, MimeTypes mimeTypes, ServiceLoader loader)
+ List<Translator> children,
+ Set<Class<? extends Translator>> excludeChildren,
+ Map<String, Param> params, MimeTypes mimeTypes, ServiceLoader loader)
throws InvocationTargetException, IllegalAccessException,
InstantiationException {
throw new InstantiationException("Only one translator supported");
}
+
@Override
Translator decorate(Translator created, Element element) {
return created; // No decoration of Translators
- }
+ }
}
-
- private static class ExecutorServiceXmlLoader extends XmlLoader<ConfigurableThreadPoolExecutor,ConfigurableThreadPoolExecutor> {
+
+ private static class ExecutorServiceXmlLoader extends XmlLoader<ConfigurableThreadPoolExecutor, ConfigurableThreadPoolExecutor> {
@Override
ConfigurableThreadPoolExecutor createComposite(
Class<? extends ConfigurableThreadPoolExecutor> compositeClass,
@@ -1084,61 +1166,69 @@ public class TikaConfig {
InstantiationException {
throw new InstantiationException("Only one executor service supported");
}
-
+
@Override
ConfigurableThreadPoolExecutor createComposite(List<ConfigurableThreadPoolExecutor> loaded,
- MimeTypes mimeTypes, ServiceLoader loader) {
+ MimeTypes mimeTypes, ServiceLoader loader) {
return loaded.get(0);
}
-
+
@Override
ConfigurableThreadPoolExecutor createDefault(MimeTypes mimeTypes, ServiceLoader loader) {
return getDefaultExecutorService();
}
-
+
@Override
ConfigurableThreadPoolExecutor decorate(ConfigurableThreadPoolExecutor created, Element element)
throws IOException, TikaException {
-
+
Element maxThreadElement = getChild(element, "max-threads");
- if(maxThreadElement != null)
- {
+ if (maxThreadElement != null) {
created.setMaximumPoolSize(Integer.parseInt(getText(maxThreadElement)));
}
-
+
Element coreThreadElement = getChild(element, "core-threads");
- if(coreThreadElement != null)
- {
+ if (coreThreadElement != null) {
created.setCorePoolSize(Integer.parseInt(getText(coreThreadElement)));
}
return created;
}
-
+
@Override
Class<? extends ConfigurableThreadPoolExecutor> getLoaderClass() {
return ConfigurableThreadPoolExecutor.class;
}
-
+
@Override
ConfigurableThreadPoolExecutor loadOne(Element element, MimeTypes mimeTypes,
- ServiceLoader loader) throws TikaException, IOException {
+ ServiceLoader loader) throws TikaException, IOException {
return super.loadOne(element, mimeTypes, loader);
}
@Override
- boolean supportsComposite() {return false;}
+ boolean supportsComposite() {
+ return false;
+ }
@Override
- String getParentTagName() {return null;}
+ String getParentTagName() {
+ return null;
+ }
@Override
- String getLoaderTagName() {return "executor-service";}
+ String getLoaderTagName() {
+ return "executor-service";
+ }
@Override
- boolean isComposite(ConfigurableThreadPoolExecutor loaded) {return false;}
+ boolean isComposite(ConfigurableThreadPoolExecutor loaded) {
+ return false;
+ }
@Override
- boolean isComposite(Class<? extends ConfigurableThreadPoolExecutor> loadedClass) {return false;}
+ boolean isComposite(Class<? extends ConfigurableThreadPoolExecutor> loadedClass) {
+ return false;
+ }
@Override
ConfigurableThreadPoolExecutor preLoadOne(
@@ -1268,7 +1358,7 @@ public class TikaConfig {
@Override
MetadataFilter preLoadOne(Class<? extends MetadataFilter> loadedClass,
- String classname, MimeTypes mimeTypes) throws TikaException {
+ String classname, MimeTypes mimeTypes) throws TikaException {
// Check for classes which can't be set in config
// Continue with normal loading
return null;
@@ -1287,9 +1377,9 @@ public class TikaConfig {
@Override
MetadataFilter createComposite(Class<? extends MetadataFilter> metadataFilterClass,
- List<MetadataFilter> childMetadataFilters,
- Set<Class<? extends MetadataFilter>> excludeFilters,
- Map<String, Param> params, MimeTypes mimeTypes, ServiceLoader loader)
+ List<MetadataFilter> childMetadataFilters,
+ Set<Class<? extends MetadataFilter>> excludeFilters,
+ Map<String, Param> params, MimeTypes mimeTypes, ServiceLoader loader)
throws InvocationTargetException, IllegalAccessException,
InstantiationException {
MetadataFilter metadataFilter = null;
@@ -1323,8 +1413,9 @@ public class TikaConfig {
}
private static class FetcherXmlLoader extends
- XmlLoader<Fetcher, Fetcher> {
+ XmlLoader<FetcherManager, Fetcher> {
+ //FetcherManager is composite
boolean supportsComposite() {
return true;
}
@@ -1345,61 +1436,42 @@ public class TikaConfig {
@Override
boolean isComposite(Fetcher loaded) {
- return loaded instanceof DefaultFetcher;
+ return false;
}
@Override
boolean isComposite(Class<? extends Fetcher> loadedClass) {
- return DefaultFetcher.class.isAssignableFrom(loadedClass);
+ return false;
}
@Override
Fetcher preLoadOne(Class<? extends Fetcher> loadedClass,
- String classname, MimeTypes mimeTypes) throws TikaException {
+ String classname, MimeTypes mimeTypes) throws TikaException {
// Check for classes which can't be set in config
// Continue with normal loading
return null;
}
@Override
- Fetcher createDefault(MimeTypes mimeTypes, ServiceLoader loader) {
- return getDefaultFetcher(loader);
+ FetcherManager createDefault(MimeTypes mimeTypes, ServiceLoader loader) {
+ return getDefaultFetcherManager(loader);
}
//this ignores the service loader
@Override
- Fetcher createComposite(List<Fetcher> loaded, MimeTypes mimeTypes, ServiceLoader loader) {
- return new DefaultFetcher(loaded);
+ FetcherManager createComposite(List<Fetcher> loaded, MimeTypes mimeTypes, ServiceLoader loader) {
+ return new FetcherManager(loaded);
}
@Override
Fetcher createComposite(Class<? extends Fetcher> fetcherClass,
- List<Fetcher> childFetchers,
- Set<Class<? extends Fetcher>> excludeFilters,
- Map<String, Param> params, MimeTypes mimeTypes, ServiceLoader loader)
+ List<Fetcher> childFetchers,
+ Set<Class<? extends Fetcher>> excludeFilters,
+ Map<String, Param> params, MimeTypes mimeTypes, ServiceLoader loader)
throws InvocationTargetException, IllegalAccessException,
InstantiationException {
- Fetcher fetcher = null;
- Constructor<? extends Fetcher> c;
+ throw new InstantiationException("Only one fetcher manager supported");
- // Try the possible default and composite detector constructors
- if (fetcher == null) {
- try {
- c = fetcherClass.getConstructor(ServiceLoader.class, Collection.class);
- fetcher = c.newInstance(loader, excludeFilters);
- } catch (NoSuchMethodException me) {
- me.printStackTrace();
- }
- }
- if (fetcher == null) {
- try {
- c = fetcherClass.getConstructor(List.class);
- fetcher = c.newInstance(childFetchers);
- } catch (NoSuchMethodException me) {
- me.printStackTrace();
- }
- }
- return fetcher;
}
@Override
@@ -1409,7 +1481,7 @@ public class TikaConfig {
}
private static class EmitterXmlLoader extends
- XmlLoader<Emitter, Emitter> {
+ XmlLoader<EmitterManager, Emitter> {
boolean supportsComposite() {
return true;
@@ -1431,12 +1503,12 @@ public class TikaConfig {
@Override
boolean isComposite(Emitter loaded) {
- return loaded instanceof DefaultEmitter;
+ return false;
}
@Override
boolean isComposite(Class<? extends Emitter> loadedClass) {
- return DefaultEmitter.class.isAssignableFrom(loadedClass);
+ return false;
}
@Override
@@ -1448,19 +1520,18 @@ public class TikaConfig {
}
@Override
- Emitter createDefault(MimeTypes mimeTypes, ServiceLoader loader) {
+ EmitterManager createDefault(MimeTypes mimeTypes, ServiceLoader loader) {
return getDefaultEmitter(loader);
}
- private Emitter getDefaultEmitter(ServiceLoader loader) {
- //TODO: should we allow service loading?
- return new DefaultEmitter(Collections.EMPTY_LIST);
+ private EmitterManager getDefaultEmitter(ServiceLoader loader) {
+ return new EmitterManager(Collections.EMPTY_LIST);
}
//this ignores the service loader
@Override
- Emitter createComposite(List<Emitter> loaded, MimeTypes mimeTypes, ServiceLoader loader) {
- return new DefaultEmitter(loaded);
+ EmitterManager createComposite(List<Emitter> loaded, MimeTypes mimeTypes, ServiceLoader loader) {
+ return new EmitterManager(loaded);
}
@Override
@@ -1504,7 +1575,7 @@ public class TikaConfig {
XmlLoader<FetchIterator, FetchIterator> {
boolean supportsComposite() {
- return true;
+ return false;
}
String getParentTagName() {
@@ -1520,7 +1591,6 @@ public class TikaConfig {
return FetchIterator.class;
}
-
@Override
boolean isComposite(FetchIterator loaded) {
return false;
@@ -1533,7 +1603,7 @@ public class TikaConfig {
@Override
FetchIterator preLoadOne(Class<? extends FetchIterator> loadedClass,
- String classname, MimeTypes mimeTypes) throws TikaException {
+ String classname, MimeTypes mimeTypes) throws TikaException {
// Check for classes which can't be set in config
// Continue with normal loading
return null;
@@ -1541,44 +1611,28 @@ public class TikaConfig {
@Override
FetchIterator createDefault(MimeTypes mimeTypes, ServiceLoader loader) {
- return new DefaultFetchIterator(Collections.EMPTY_LIST);
+ return new EmptyFetchIterator();
}
//this ignores the service loader
@Override
- FetchIterator createComposite(List<FetchIterator> loaded, MimeTypes mimeTypes, ServiceLoader loader) {
- return new DefaultFetchIterator(loaded);
+ FetchIterator createComposite(List<FetchIterator> loaded,
+ MimeTypes mimeTypes, ServiceLoader loader) {
+ if (loaded.size() > 1) {
+ throw new IllegalArgumentException(
+ "Can only specify one fetch iterator per config file");
+ }
+ return loaded.get(0);
}
@Override
FetchIterator createComposite(Class<? extends FetchIterator> fetchIteratorClass,
- List<FetchIterator> fetchIteratorChildren,
- Set<Class<? extends FetchIterator>> excludeFetchIterators,
- Map<String, Param> params, MimeTypes mimeTypes, ServiceLoader loader)
+ List<FetchIterator> fetchIteratorChildren,
+ Set<Class<? extends FetchIterator>> excludeFetchIterators,
+ Map<String, Param> params, MimeTypes mimeTypes, ServiceLoader loader)
throws InvocationTargetException, IllegalAccessException,
InstantiationException {
- FetchIterator fetchIterator = null;
- Constructor<? extends FetchIterator> c;
-
- // Try the possible default and composite detector constructors
- if (fetchIterator == null) {
- try {
- c = fetchIteratorClass.getConstructor(ServiceLoader.class, Collection.class);
- fetchIterator = c.newInstance(loader, excludeFetchIterators);
- } catch (NoSuchMethodException me) {
- me.printStackTrace();
- }
- }
- if (fetchIterator == null) {
- try {
- c = fetchIteratorClass.getConstructor(List.class);
- fetchIterator = c.newInstance(fetchIteratorChildren);
- } catch (NoSuchMethodException me) {
- me.printStackTrace();
- }
- }
-
- return fetchIterator;
+ return null;
}
@Override
@@ -1587,4 +1641,5 @@ public class TikaConfig {
}
}
+
}
diff --git a/tika-core/src/main/java/org/apache/tika/emitter/Emitter.java b/tika-core/src/main/java/org/apache/tika/emitter/Emitter.java
deleted file mode 100644
index 23021e3..0000000
--- a/tika-core/src/main/java/org/apache/tika/emitter/Emitter.java
+++ /dev/null
@@ -1,20 +0,0 @@
-package org.apache.tika.emitter;
-
-import org.apache.tika.exception.TikaException;
-import org.apache.tika.metadata.Metadata;
-
-import java.io.IOException;
-import java.util.List;
-import java.util.Set;
-
-public interface Emitter {
-
- Set<String> getSupported();
-
- //TODO: do we need a key or can we pass that in metadatalist?
- //If we do need it, how do we populate it?
- void emit(String emitterName, List<Metadata> metadataList) throws IOException, TikaException;
- //TODO we can add this later?
- //void emit(String txt, Metadata metadata) throws IOException, TikaException;
-
-}
diff --git a/tika-core/src/main/java/org/apache/tika/emitter/TikaEmitterException.java b/tika-core/src/main/java/org/apache/tika/emitter/TikaEmitterException.java
deleted file mode 100644
index 6926b78..0000000
--- a/tika-core/src/main/java/org/apache/tika/emitter/TikaEmitterException.java
+++ /dev/null
@@ -1,9 +0,0 @@
-package org.apache.tika.emitter;
-
-import org.apache.tika.exception.TikaException;
-
-public class TikaEmitterException extends TikaException {
- public TikaEmitterException(String msg) {
- super(msg);
- }
-}
diff --git a/tika-core/src/main/java/org/apache/tika/fetcher/DefaultFetchIterator.java b/tika-core/src/main/java/org/apache/tika/fetcher/DefaultFetchIterator.java
deleted file mode 100644
index a579314..0000000
--- a/tika-core/src/main/java/org/apache/tika/fetcher/DefaultFetchIterator.java
+++ /dev/null
@@ -1,57 +0,0 @@
-package org.apache.tika.fetcher;
-
-import org.apache.tika.config.Field;
-import org.apache.tika.config.Initializable;
-import org.apache.tika.config.InitializableProblemHandler;
-import org.apache.tika.config.Param;
-import org.apache.tika.exception.TikaConfigException;
-
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.TimeoutException;
-
-public class DefaultFetchIterator extends FetchIterator implements Initializable {
-
- private final Map<String, FetchIterator> fetchIterators = new HashMap<>();
- private String iteratorName = "";
- public DefaultFetchIterator(List<FetchIterator> fetchIterators) {
- super("default");
- for (FetchIterator fetchIterator : fetchIterators) {
- if (this.fetchIterators.containsKey(fetchIterator.getName())) {
- throw new RuntimeException(new TikaConfigException("Multiple fetchIterators cannot have the same name: "
- + fetchIterator.getName()));
- }
- this.fetchIterators.put(fetchIterator.getName(), fetchIterator);
- }
- }
-
- @Override
- protected void enqueue() throws IOException, TimeoutException {
- if (fetchIterators.size() == 0) {
- return;
- } else if (fetchIterators.size() == 1) {
- for (FetchIterator fetchIterator : fetchIterators.values()) {
- fetchIterator.enqueue();
- }
- }
- }
-
- @Field
- public void setIteratorName(String iteratorName) {
- this.iteratorName = iteratorName;
- }
-
- @Override
- public void initialize(Map<String, Param> params) throws TikaConfigException {
- //no-op
- }
-
- @Override
- public void checkInitialization(InitializableProblemHandler problemHandler) throws TikaConfigException {
- if (this.fetchIterators.size() > 1 && this.iteratorName == null) {
- throw new TikaConfigException("Must set desired iteratorName, if multiple iterators are defined");
- }
- }
-}
diff --git a/tika-core/src/main/java/org/apache/tika/fetcher/FetchIterator.java b/tika-core/src/main/java/org/apache/tika/fetcher/FetchIterator.java
deleted file mode 100644
index 84e74d3..0000000
--- a/tika-core/src/main/java/org/apache/tika/fetcher/FetchIterator.java
+++ /dev/null
@@ -1,141 +0,0 @@
-/*
- * 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.fetcher;
-
-import org.apache.tika.config.Field;
-
-import java.io.IOException;
-import java.util.Iterator;
-import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.Callable;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-
-/**
- * Abstract class that handles the testing for timeouts/thread safety
- * issues. Concrete classes implement {@link #enqueue()}.
- *
- * This must be "called" and managed with an ExecutorService, etc.
- * for the iterable to work.
- */
-public abstract class FetchIterator implements Callable<Integer>,
- Iterable<FetchMetadataPair> {
-
- public static final long DEFAULT_MAX_WAIT_MS = 300_000;
-
- static final FetchMetadataPair POISON =
- new FetchMetadataPair(null, null);
-
- private final int queueSize = 1000;
- private long maxWaitMs = DEFAULT_MAX_WAIT_MS;
- private final ArrayBlockingQueue<FetchMetadataPair> queue = new ArrayBlockingQueue<>(queueSize);
- private String name;
-
- public FetchIterator(String name) {
- this.name = name;
- }
-
- @Field
- public void setName(String name) {
- this.name = name;
- }
-
- public String getName() {
- return name;
- }
-
- @Override
- public Integer call() throws Exception {
- enqueue();
- System.out.println("finished");
- return 1;
- }
-
- protected abstract void enqueue() throws IOException, TimeoutException;
-
- void tryToAdd(FetchMetadataPair p) throws InterruptedException, TimeoutException {
- System.out.println("trying to add: "+p + " "+ queue.size());
- boolean offered = queue.offer(p, maxWaitMs, TimeUnit.MILLISECONDS);
- System.out.println("added: "+p + " "+ queue.size());
- if (! offered) {
- throw new TimeoutException("timed out while offering");
- }
- }
-
- @Override
- public Iterator<FetchMetadataPair> iterator() {
- return new InternalIterator();
- }
-
- private class InternalIterator implements Iterator<FetchMetadataPair> {
- //Object[] is recommended by FindBugs as a lock object
- private Object[] lock = new Object[0];
- private FetchMetadataPair next = null;
- volatile boolean initialized = false;
- InternalIterator() {
-
- }
-
- @Override
- public boolean hasNext() {
- System.out.println("hasNExt");
- if (!initialized) {
- next = getNext();
- initialized = true;
- }
-
- return next != POISON;
- }
-
- /**
- *
- * @return next FetcherStringMetadataPair; if {@link #hasNext()} returns
- * false, this will return null.
- */
- @Override
- public FetchMetadataPair next() {
-
- System.out.println("in next");
-
- if (next == POISON) {
- return null;
- }
- FetchMetadataPair ret = next;
- next = getNext();
- return ret;
- }
-
- private FetchMetadataPair getNext() {
- FetchMetadataPair p = null;
- System.out.println("in get next: " + queue.size());
- try {
- //System.out.println("peek: " + queue.peek());
- p = queue.poll(maxWaitMs, TimeUnit.MILLISECONDS);
- //System.out.println("peek2: " + queue.peek() + " :: "+p);
- } catch (InterruptedException e) {
- throw new RuntimeException(e);
- } catch (Exception e) {
- e.printStackTrace();
- }
-
- if (p == null) {
- throw new RuntimeException(new TimeoutException(""));
- }
- return p;
- }
- }
-}
diff --git a/tika-core/src/main/java/org/apache/tika/fetcher/FetchPrefixKeyPair.java b/tika-core/src/main/java/org/apache/tika/fetcher/FetchPrefixKeyPair.java
deleted file mode 100644
index 2a068c5..0000000
--- a/tika-core/src/main/java/org/apache/tika/fetcher/FetchPrefixKeyPair.java
+++ /dev/null
@@ -1,53 +0,0 @@
-/*
- * 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.fetcher;
-
-public class FetchPrefixKeyPair {
- private final String prefix;
- private final String key;
-
- private FetchPrefixKeyPair(String prefix, String key) {
- this.prefix = prefix;
- this.key = key;
- }
-
- public static FetchPrefixKeyPair create(String fetcherString) throws FetcherStringException {
- int prefixIndex = fetcherString.indexOf(":");
- if (prefixIndex < 0) {
- throw new FetcherStringException("Can't find fetcher prefix, e.g. the 's3' in s3:/myfile");
- }
- String prefix = fetcherString.substring(0, prefixIndex);
- String key = fetcherString.substring(prefixIndex+1);
- return new FetchPrefixKeyPair(prefix, key);
- }
-
- public String getPrefix() {
- return prefix;
- }
-
- public String getKey() {
- return key;
- }
-
- @Override
- public String toString() {
- return "FetchPrefixKeyPair{" +
- "prefix='" + prefix + '\'' +
- ", key='" + key + '\'' +
- '}';
- }
-}
diff --git a/tika-core/src/main/java/org/apache/tika/fetcher/FileSystemFetcher.java b/tika-core/src/main/java/org/apache/tika/fetcher/FileSystemFetcher.java
deleted file mode 100644
index 63c4cba..0000000
--- a/tika-core/src/main/java/org/apache/tika/fetcher/FileSystemFetcher.java
+++ /dev/null
@@ -1,79 +0,0 @@
-/*
- * 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.fetcher;
-
-import org.apache.tika.config.Field;
-import org.apache.tika.exception.TikaException;
-import org.apache.tika.io.TikaInputStream;
-import org.apache.tika.metadata.Metadata;
-import org.apache.tika.metadata.TikaCoreProperties;
-
-import java.io.FileNotFoundException;
-import java.io.IOException;
-import java.io.InputStream;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.Paths;
-import java.util.Collections;
-import java.util.Set;
-
-public class FileSystemFetcher implements Fetcher {
-
- private static String PREFIX = "fs";
- private static final Set<String> SUPPORTED = Collections.singleton(PREFIX);
- private Path basePath = null;
- @Override
- public Set<String> getSupportedPrefixes() {
- return SUPPORTED;
- }
-
- @Override
- public InputStream fetch(String fetcherString, Metadata metadata)
- throws IOException, TikaException {
- FetchPrefixKeyPair fetchPrefixKeyPair = FetchPrefixKeyPair.create(fetcherString);
- metadata.set(TikaCoreProperties.SOURCE_PATH, fetchPrefixKeyPair.getKey());
- Path p = null;
- if (basePath != null) {
- p = basePath.resolve(fetchPrefixKeyPair.getKey());
- if (!Files.isRegularFile(p)) {
- if (!Files.isDirectory(basePath)) {
- throw new IOException("BasePath is not a directory: "+basePath);
- } else {
- throw new FileNotFoundException(p.toAbsolutePath().toString());
- }
- }
- } else {
- p = Paths.get(fetchPrefixKeyPair.getKey());
- if (!Files.isRegularFile(p)) {
- throw new FileNotFoundException(p.toAbsolutePath().toString());
- }
- }
- return TikaInputStream.get(p, metadata);
- }
-
- /**
- * If clients will send in relative paths, this
- * must be set to allow this fetcher to fetch the
- * full path.
- *
- * @param basePath
- */
- @Field
- public void setBasePath(String basePath) {
- this.basePath = Paths.get(basePath);
- }
-}
diff --git a/tika-core/src/main/java/org/apache/tika/fetcher/FetcherStringException.java b/tika-core/src/main/java/org/apache/tika/pipes/emitter/AbstractEmitter.java
similarity index 72%
copy from tika-core/src/main/java/org/apache/tika/fetcher/FetcherStringException.java
copy to tika-core/src/main/java/org/apache/tika/pipes/emitter/AbstractEmitter.java
index 9f1e93b..0aa9e7f 100644
--- a/tika-core/src/main/java/org/apache/tika/fetcher/FetcherStringException.java
+++ b/tika-core/src/main/java/org/apache/tika/pipes/emitter/AbstractEmitter.java
@@ -14,16 +14,20 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.tika.fetcher;
+package org.apache.tika.pipes.emitter;
-import org.apache.tika.exception.TikaException;
+import org.apache.tika.config.Field;
-/**
- * If something goes wrong in parsing the fetcher string
- */
-public class FetcherStringException extends TikaException {
+public abstract class AbstractEmitter implements Emitter {
+
+ private String name;
- public FetcherStringException(String msg) {
- super(msg);
+ @Field
+ public void setName(String name) {
+ this.name = name;
+ }
+ @Override
+ public String getName() {
+ return name;
}
}
diff --git a/tika-core/src/main/java/org/apache/tika/fetcher/FetcherStringException.java b/tika-core/src/main/java/org/apache/tika/pipes/emitter/Emitter.java
similarity index 68%
copy from tika-core/src/main/java/org/apache/tika/fetcher/FetcherStringException.java
copy to tika-core/src/main/java/org/apache/tika/pipes/emitter/Emitter.java
index 9f1e93b..8f31311 100644
--- a/tika-core/src/main/java/org/apache/tika/fetcher/FetcherStringException.java
+++ b/tika-core/src/main/java/org/apache/tika/pipes/emitter/Emitter.java
@@ -14,16 +14,21 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.tika.fetcher;
+package org.apache.tika.pipes.emitter;
import org.apache.tika.exception.TikaException;
+import org.apache.tika.metadata.Metadata;
-/**
- * If something goes wrong in parsing the fetcher string
- */
-public class FetcherStringException extends TikaException {
+import java.io.IOException;
+import java.util.List;
+import java.util.Set;
+
+public interface Emitter {
+
+ String getName();
+
+ void emit(List<Metadata> metadataList) throws IOException, TikaException;
+ //TODO we can add this later?
+ //void emit(String txt, Metadata metadata) throws IOException, TikaException;
- public FetcherStringException(String msg) {
- super(msg);
- }
}
diff --git a/tika-core/src/main/java/org/apache/tika/emitter/DefaultEmitter.java b/tika-core/src/main/java/org/apache/tika/pipes/emitter/EmitterManager.java
similarity index 61%
rename from tika-core/src/main/java/org/apache/tika/emitter/DefaultEmitter.java
rename to tika-core/src/main/java/org/apache/tika/pipes/emitter/EmitterManager.java
index 80fe314..3c6ec70 100644
--- a/tika-core/src/main/java/org/apache/tika/emitter/DefaultEmitter.java
+++ b/tika-core/src/main/java/org/apache/tika/pipes/emitter/EmitterManager.java
@@ -14,16 +14,8 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.tika.emitter;
+package org.apache.tika.pipes.emitter;
-import org.apache.tika.config.ServiceLoader;
-import org.apache.tika.exception.TikaException;
-import org.apache.tika.fetcher.FetchPrefixKeyPair;
-import org.apache.tika.fetcher.Fetcher;
-import org.apache.tika.metadata.Metadata;
-
-import java.io.IOException;
-import java.io.InputStream;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -35,48 +27,34 @@ import java.util.concurrent.ConcurrentHashMap;
*
* This does not allow multiple fetchers supporting the same prefix.
*/
-public class DefaultEmitter implements Emitter {
+public class EmitterManager {
private final Map<String, Emitter> emitterMap = new ConcurrentHashMap<>();
- private static List<Emitter> getDefaultFilters(
- ServiceLoader loader) {
- return loader.loadStaticServiceProviders(Emitter.class);
- }
-
- public DefaultEmitter(ServiceLoader serviceLoader) {
- this(getDefaultFilters(serviceLoader));
- }
-
- public DefaultEmitter(List<Emitter> emitters) {
+ public EmitterManager(List<Emitter> emitters) {
for (Emitter emitter : emitters) {
- for (String name : emitter.getSupported()) {
- if (emitterMap.containsKey(name)) {
+ if (emitterMap.containsKey(emitter.getName())) {
throw new IllegalArgumentException(
"Multiple emitters cannot support the same name: "
- + name);
+ + emitter.getName());
}
- emitterMap.put(name, emitter);
- }
+ emitterMap.put(emitter.getName(), emitter);
+
}
}
- @Override
public Set<String> getSupported() {
return emitterMap.keySet();
}
- @Override
- public void emit(String emitterName, List<Metadata> metadata)
- throws IOException, TikaException {
-
+ public Emitter getEmitter(String emitterName) {
Emitter emitter = emitterMap.get(emitterName);
if (emitter == null) {
throw new IllegalArgumentException("Can't find emitter for prefix: "+
emitterName);
}
- emitter.emit(emitterName, metadata);
+ return emitter;
}
}
diff --git a/tika-core/src/main/java/org/apache/tika/fetcher/FetchMetadataPair.java b/tika-core/src/main/java/org/apache/tika/pipes/emitter/EmptyEmitter.java
similarity index 57%
copy from tika-core/src/main/java/org/apache/tika/fetcher/FetchMetadataPair.java
copy to tika-core/src/main/java/org/apache/tika/pipes/emitter/EmptyEmitter.java
index 9711c1f..9a1a10a 100644
--- a/tika-core/src/main/java/org/apache/tika/fetcher/FetchMetadataPair.java
+++ b/tika-core/src/main/java/org/apache/tika/pipes/emitter/EmptyEmitter.java
@@ -14,33 +14,22 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.tika.fetcher;
+package org.apache.tika.pipes.emitter;
+import org.apache.tika.exception.TikaException;
import org.apache.tika.metadata.Metadata;
-public class FetchMetadataPair {
+import java.io.IOException;
+import java.util.List;
- private final String fetcherString;
- private final Metadata metadata;
-
- public FetchMetadataPair(String fetcherString, Metadata metadata) {
- this.fetcherString = fetcherString;
- this.metadata = metadata;
- }
-
- public String getFetcherString() {
- return fetcherString;
- }
-
- public Metadata getMetadata() {
- return metadata;
+public class EmptyEmitter implements Emitter {
+ @Override
+ public String getName() {
+ return "empty";
}
@Override
- public String toString() {
- return "FetcherStringMetadataPair{" +
- "fetcherString='" + fetcherString + '\'' +
- ", metadata=" + metadata +
- '}';
+ public void emit(List<Metadata> metadataList) throws IOException, TikaException {
+
}
}
diff --git a/tika-core/src/main/java/org/apache/tika/fetcher/FetcherStringException.java b/tika-core/src/main/java/org/apache/tika/pipes/emitter/TikaEmitterException.java
similarity index 80%
copy from tika-core/src/main/java/org/apache/tika/fetcher/FetcherStringException.java
copy to tika-core/src/main/java/org/apache/tika/pipes/emitter/TikaEmitterException.java
index 9f1e93b..00a5197 100644
--- a/tika-core/src/main/java/org/apache/tika/fetcher/FetcherStringException.java
+++ b/tika-core/src/main/java/org/apache/tika/pipes/emitter/TikaEmitterException.java
@@ -14,16 +14,12 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.tika.fetcher;
+package org.apache.tika.pipes.emitter;
import org.apache.tika.exception.TikaException;
-/**
- * If something goes wrong in parsing the fetcher string
- */
-public class FetcherStringException extends TikaException {
-
- public FetcherStringException(String msg) {
+public class TikaEmitterException extends TikaException {
+ public TikaEmitterException(String msg) {
super(msg);
}
}
diff --git a/tika-core/src/main/java/org/apache/tika/fetcher/FetcherStringException.java b/tika-core/src/main/java/org/apache/tika/pipes/fetcher/AbstractFetcher.java
similarity index 65%
copy from tika-core/src/main/java/org/apache/tika/fetcher/FetcherStringException.java
copy to tika-core/src/main/java/org/apache/tika/pipes/fetcher/AbstractFetcher.java
index 9f1e93b..0b417e3 100644
--- a/tika-core/src/main/java/org/apache/tika/fetcher/FetcherStringException.java
+++ b/tika-core/src/main/java/org/apache/tika/pipes/fetcher/AbstractFetcher.java
@@ -14,16 +14,31 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.tika.fetcher;
+package org.apache.tika.pipes.fetcher;
-import org.apache.tika.exception.TikaException;
+import org.apache.tika.config.Field;
-/**
- * If something goes wrong in parsing the fetcher string
- */
-public class FetcherStringException extends TikaException {
- public FetcherStringException(String msg) {
- super(msg);
+public abstract class AbstractFetcher implements Fetcher {
+
+ private String name;
+
+ public AbstractFetcher() {
+
+ }
+
+ public AbstractFetcher(String name) {
+ this.name = name;
}
+
+ @Override
+ public String getName() {
+ return name;
+ }
+
+ @Field
+ public void setName(String name) {
+ this.name = name;
+ }
+
}
diff --git a/tika-core/src/main/java/org/apache/tika/fetcher/FetcherStringException.java b/tika-core/src/main/java/org/apache/tika/pipes/fetcher/EmptyFetcher.java
similarity index 68%
copy from tika-core/src/main/java/org/apache/tika/fetcher/FetcherStringException.java
copy to tika-core/src/main/java/org/apache/tika/pipes/fetcher/EmptyFetcher.java
index 9f1e93b..5bc84b0 100644
--- a/tika-core/src/main/java/org/apache/tika/fetcher/FetcherStringException.java
+++ b/tika-core/src/main/java/org/apache/tika/pipes/fetcher/EmptyFetcher.java
@@ -14,16 +14,23 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.tika.fetcher;
+package org.apache.tika.pipes.fetcher;
import org.apache.tika.exception.TikaException;
+import org.apache.tika.metadata.Metadata;
-/**
- * If something goes wrong in parsing the fetcher string
- */
-public class FetcherStringException extends TikaException {
+import java.io.IOException;
+import java.io.InputStream;
+
+public class EmptyFetcher implements Fetcher {
+
+ @Override
+ public String getName() {
+ return "empty";
+ }
- public FetcherStringException(String msg) {
- super(msg);
+ @Override
+ public InputStream fetch(String fetchKey, Metadata metadata) throws TikaException, IOException {
+ return null;
}
}
diff --git a/tika-core/src/main/java/org/apache/tika/pipes/fetcher/FetchId.java b/tika-core/src/main/java/org/apache/tika/pipes/fetcher/FetchId.java
new file mode 100644
index 0000000..a327fc6
--- /dev/null
+++ b/tika-core/src/main/java/org/apache/tika/pipes/fetcher/FetchId.java
@@ -0,0 +1,65 @@
+/*
+ * 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;
+
+/**
+ * Pair of fetcherName (which fetcher to call) and the key
+ * to send to that fetcher to retrieve a specific file.
+ */
+public class FetchId {
+ private final String fetcherName;
+ private final String fetchKey;
+
+ public FetchId(String fetcherName, String fetchKey) {
+ this.fetcherName = fetcherName;
+ this.fetchKey = fetchKey;
+ }
+
+ public String getFetcherName() {
+ return fetcherName;
+ }
+
+ public String getFetchKey() {
+ return fetchKey;
+ }
+
+ @Override
+ public String toString() {
+ return "FetcherKeyPair{" +
+ "fetcherName='" + fetcherName + '\'' +
+ ", fetchKey='" + fetchKey + '\'' +
+ '}';
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ FetchId fetchId = (FetchId) o;
+
+ if (fetcherName != null ? !fetcherName.equals(fetchId.fetcherName) : fetchId.fetcherName != null) return false;
+ return fetchKey != null ? fetchKey.equals(fetchId.fetchKey) : fetchId.fetchKey == null;
+ }
+
+ @Override
+ public int hashCode() {
+ int result = fetcherName != null ? fetcherName.hashCode() : 0;
+ result = 31 * result + (fetchKey != null ? fetchKey.hashCode() : 0);
+ return result;
+ }
+}
diff --git a/tika-core/src/main/java/org/apache/tika/fetcher/FetchMetadataPair.java b/tika-core/src/main/java/org/apache/tika/pipes/fetcher/FetchIdMetadataPair.java
similarity index 53%
rename from tika-core/src/main/java/org/apache/tika/fetcher/FetchMetadataPair.java
rename to tika-core/src/main/java/org/apache/tika/pipes/fetcher/FetchIdMetadataPair.java
index 9711c1f..25d89db 100644
--- a/tika-core/src/main/java/org/apache/tika/fetcher/FetchMetadataPair.java
+++ b/tika-core/src/main/java/org/apache/tika/pipes/fetcher/FetchIdMetadataPair.java
@@ -14,33 +14,51 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.tika.fetcher;
+package org.apache.tika.pipes.fetcher;
import org.apache.tika.metadata.Metadata;
-public class FetchMetadataPair {
+public class FetchIdMetadataPair {
- private final String fetcherString;
+ private final FetchId fetchId;
private final Metadata metadata;
- public FetchMetadataPair(String fetcherString, Metadata metadata) {
- this.fetcherString = fetcherString;
+ public FetchIdMetadataPair(FetchId fetchId, Metadata metadata) {
+ this.fetchId = fetchId;
this.metadata = metadata;
}
- public String getFetcherString() {
- return fetcherString;
- }
-
public Metadata getMetadata() {
return metadata;
}
+ public FetchId getFetchId() {
+ return fetchId;
+ }
+
@Override
public String toString() {
- return "FetcherStringMetadataPair{" +
- "fetcherString='" + fetcherString + '\'' +
+ return "FetchIdMetadataPair{" +
+ "fetchId=" + fetchId +
", metadata=" + metadata +
'}';
}
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ FetchIdMetadataPair that = (FetchIdMetadataPair) o;
+
+ if (fetchId != null ? !fetchId.equals(that.fetchId) : that.fetchId != null) return false;
+ return metadata != null ? metadata.equals(that.metadata) : that.metadata == null;
+ }
+
+ @Override
+ public int hashCode() {
+ int result = fetchId != null ? fetchId.hashCode() : 0;
+ result = 31 * result + (metadata != null ? metadata.hashCode() : 0);
+ return result;
+ }
}
diff --git a/tika-core/src/main/java/org/apache/tika/fetcher/Fetcher.java b/tika-core/src/main/java/org/apache/tika/pipes/fetcher/Fetcher.java
similarity index 76%
rename from tika-core/src/main/java/org/apache/tika/fetcher/Fetcher.java
rename to tika-core/src/main/java/org/apache/tika/pipes/fetcher/Fetcher.java
index 95a795c..0450737 100644
--- a/tika-core/src/main/java/org/apache/tika/fetcher/Fetcher.java
+++ b/tika-core/src/main/java/org/apache/tika/pipes/fetcher/Fetcher.java
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.tika.fetcher;
+package org.apache.tika.pipes.fetcher;
import org.apache.tika.exception.TikaException;
import org.apache.tika.metadata.Metadata;
@@ -30,12 +30,10 @@ import java.util.Set;
*
* Implementations of Fetcher must be thread safe.
*
- * The fetcherString must start with a prefix that can be
- * used to uniquely select the fetcher, e.g. file:my_file.pdf, s3:bucket/path/to/my_file
- *
- * Each fetcher must specify which prefixes it can handle.
*/
public interface Fetcher {
- Set<String> getSupportedPrefixes();
- InputStream fetch(String fetcherString, Metadata metadata) throws TikaException, IOException;
+
+ String getName();
+
+ InputStream fetch(String fetchKey, Metadata metadata) throws TikaException, IOException;
}
diff --git a/tika-core/src/main/java/org/apache/tika/fetcher/DefaultFetcher.java b/tika-core/src/main/java/org/apache/tika/pipes/fetcher/FetcherManager.java
similarity index 58%
rename from tika-core/src/main/java/org/apache/tika/fetcher/DefaultFetcher.java
rename to tika-core/src/main/java/org/apache/tika/pipes/fetcher/FetcherManager.java
index 085e021..1ea46fd 100644
--- a/tika-core/src/main/java/org/apache/tika/fetcher/DefaultFetcher.java
+++ b/tika-core/src/main/java/org/apache/tika/pipes/fetcher/FetcherManager.java
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.tika.fetcher;
+package org.apache.tika.pipes.fetcher;
import org.apache.tika.exception.TikaException;
import org.apache.tika.metadata.Metadata;
@@ -27,43 +27,41 @@ import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
/**
- * Utility class that will apply the appropriate fetcher
- * to the fetcherString based on the prefix.
- *
+ * Utility class to hold multiple fetchers.
+ * <p>
* This forbids multiple fetchers supporting the same prefix.
*/
-public class DefaultFetcher implements Fetcher {
+public class FetcherManager {
private final Map<String, Fetcher> fetcherMap = new ConcurrentHashMap<>();
- public DefaultFetcher(List<Fetcher> fetchers) {
+ public FetcherManager(List<Fetcher> fetchers) {
for (Fetcher fetcher : fetchers) {
- for (String supportedPrefix : fetcher.getSupportedPrefixes()) {
- if (fetcherMap.containsKey(supportedPrefix)) {
- throw new IllegalArgumentException(
- "Multiple fetchers cannot support the same prefix: "
- + supportedPrefix);
- }
- fetcherMap.put(supportedPrefix, fetcher);
+ String name = fetcher.getName();
+ if (name == null || name.trim().length() == 0) {
+ throw new IllegalArgumentException("fetcher name must not be blank");
+ }
+ if (fetcherMap.containsKey(fetcher.getName())) {
+ throw new IllegalArgumentException(
+ "Multiple fetchers cannot support the same prefix: "
+ + fetcher.getName());
}
+ fetcherMap.put(fetcher.getName(), fetcher);
}
}
- @Override
- public Set<String> getSupportedPrefixes() {
- return fetcherMap.keySet();
- }
-
- @Override
- public InputStream fetch(String fetcherString, Metadata metadata)
+ public Fetcher getFetcher(String fetcherName)
throws IOException, TikaException {
- FetchPrefixKeyPair fetchPrefixKeyPair = FetchPrefixKeyPair.create(fetcherString);
-
- Fetcher fetcher = fetcherMap.get(fetchPrefixKeyPair.getPrefix());
+ Fetcher fetcher = fetcherMap.get(fetcherName);
if (fetcher == null) {
- throw new IllegalArgumentException("Can't find fetcher for prefix: "+
- fetchPrefixKeyPair.getPrefix());
+ throw new IllegalArgumentException("Can't find fetcher for fetcherName: " +
+ fetcherName +
+ ". I've loaded: "+fetcherMap.keySet());
}
- return fetcher.fetch(fetcherString, metadata);
+ return fetcher;
+ }
+
+ public Set<String> getSupported() {
+ return fetcherMap.keySet();
}
}
diff --git a/tika-core/src/main/java/org/apache/tika/fetcher/FetcherStringException.java b/tika-core/src/main/java/org/apache/tika/pipes/fetcher/FetcherStringException.java
similarity index 96%
copy from tika-core/src/main/java/org/apache/tika/fetcher/FetcherStringException.java
copy to tika-core/src/main/java/org/apache/tika/pipes/fetcher/FetcherStringException.java
index 9f1e93b..a07439a 100644
--- a/tika-core/src/main/java/org/apache/tika/fetcher/FetcherStringException.java
+++ b/tika-core/src/main/java/org/apache/tika/pipes/fetcher/FetcherStringException.java
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.tika.fetcher;
+package org.apache.tika.pipes.fetcher;
import org.apache.tika.exception.TikaException;
diff --git a/tika-core/src/main/java/org/apache/tika/pipes/fetcher/FileSystemFetcher.java b/tika-core/src/main/java/org/apache/tika/pipes/fetcher/FileSystemFetcher.java
new file mode 100644
index 0000000..7bdf56d
--- /dev/null
+++ b/tika-core/src/main/java/org/apache/tika/pipes/fetcher/FileSystemFetcher.java
@@ -0,0 +1,110 @@
+/*
+ * 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;
+
+import org.apache.tika.config.Field;
+import org.apache.tika.config.Initializable;
+import org.apache.tika.config.InitializableProblemHandler;
+import org.apache.tika.config.Param;
+import org.apache.tika.exception.TikaConfigException;
+import org.apache.tika.exception.TikaException;
+import org.apache.tika.io.TikaInputStream;
+import org.apache.tika.metadata.Metadata;
+import org.apache.tika.metadata.TikaCoreProperties;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Set;
+
+public class FileSystemFetcher extends AbstractFetcher implements Initializable {
+
+ private Path basePath = null;
+
+ @Override
+ public InputStream fetch(String fetchKey, Metadata metadata)
+ throws IOException, TikaException {
+ if (basePath == null) {
+ throw new IllegalStateException("must set 'basePath' before calling fetch");
+
+ }
+ if (fetchKey.contains("\u0000")) {
+ throw new IllegalArgumentException("Path must not contain \u0000. " +
+ "Please review the life decisions that led you to requesting " +
+ "a file name with this character in it.");
+ }
+ Path p = basePath.resolve(fetchKey);
+ if (!p.toRealPath().startsWith(basePath.toRealPath())) {
+ throw new IllegalArgumentException("fetchKey must resolve to be a" +
+ " descendant of the 'basePath'");
+ }
+
+ metadata.set(TikaCoreProperties.SOURCE_PATH, fetchKey);
+
+ if (!Files.isRegularFile(p)) {
+ if (!Files.isDirectory(basePath)) {
+ throw new IOException("BasePath is not a directory: " + basePath);
+ } else {
+ throw new FileNotFoundException(p.toAbsolutePath().toString());
+ }
+ }
+
+ return TikaInputStream.get(p, metadata);
+ }
+
+ /**
+ * If clients will send in relative paths, this
+ * must be set to allow this fetcher to fetch the
+ * full path.
+ *
+ * @param basePath
+ */
+ @Field
+ public void setBasePath(String basePath) {
+ this.basePath = Paths.get(basePath);
+ }
+
+ static boolean isDescendant(Path root, Path descendant) {
+ return descendant.toAbsolutePath().normalize().startsWith(
+ root.toAbsolutePath().normalize());
+ }
+
+ public Path getBasePath() {
+ return basePath;
+ }
+
+ @Override
+ public void initialize(Map<String, Param> params) throws TikaConfigException {
+ //no-op
+ }
+
+ @Override
+ public void checkInitialization(InitializableProblemHandler problemHandler) throws TikaConfigException {
+ if (basePath == null || basePath.toString().trim().length() == 0) {
+ throw new IllegalArgumentException("'basePath' must be specified");
+ }
+ if (basePath.toAbsolutePath().toString().contains("\u0000")) {
+ throw new IllegalArgumentException("base path must not contain \u0000. " +
+ "Seriously, what were you thinking?");
+ }
+ }
+}
diff --git a/tika-core/src/main/java/org/apache/tika/fetcher/SimpleUrlFetcher.java b/tika-core/src/main/java/org/apache/tika/pipes/fetcher/SimpleUrlFetcher.java
similarity index 78%
rename from tika-core/src/main/java/org/apache/tika/fetcher/SimpleUrlFetcher.java
rename to tika-core/src/main/java/org/apache/tika/pipes/fetcher/SimpleUrlFetcher.java
index a15b78c..40faadd 100644
--- a/tika-core/src/main/java/org/apache/tika/fetcher/SimpleUrlFetcher.java
+++ b/tika-core/src/main/java/org/apache/tika/pipes/fetcher/SimpleUrlFetcher.java
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.tika.fetcher;
+package org.apache.tika.pipes.fetcher;
import org.apache.tika.exception.TikaException;
import org.apache.tika.io.TikaInputStream;
@@ -35,21 +35,19 @@ import java.util.Set;
* This is limited to http: and https: urls. This does
* not support the file:/// protocol. See {@link FileSystemFetcher}.
*/
-public class SimpleUrlFetcher implements Fetcher {
+public class SimpleUrlFetcher extends AbstractFetcher {
- private static String PREFIX = "url";
- private static final Set<String> SUPPORTED = Collections.singleton(PREFIX);
+ private static String NAME = "url";
- @Override
- public Set<String> getSupportedPrefixes() {
- return SUPPORTED;
+ public SimpleUrlFetcher() {
+ super(NAME);
}
+
@Override
- public InputStream fetch(String fetcherString, Metadata metadata)
+ public InputStream fetch(String fetchKey, Metadata metadata)
throws IOException, TikaException {
- FetchPrefixKeyPair fetchPrefixKeyPair = FetchPrefixKeyPair.create(fetcherString);
- URL url = new URL(fetchPrefixKeyPair.getKey());
+ URL url = new URL(fetchKey);
if (! url.getProtocol().equals("http") &&
! url.getProtocol().equals("https") &&
! url.getProtocol().equals("ftp")) {
diff --git a/tika-core/src/main/java/org/apache/tika/fetcher/FetcherStringException.java b/tika-core/src/main/java/org/apache/tika/pipes/fetchiterator/EmptyFetchIterator.java
similarity index 64%
copy from tika-core/src/main/java/org/apache/tika/fetcher/FetcherStringException.java
copy to tika-core/src/main/java/org/apache/tika/pipes/fetchiterator/EmptyFetchIterator.java
index 9f1e93b..1a2d40e 100644
--- a/tika-core/src/main/java/org/apache/tika/fetcher/FetcherStringException.java
+++ b/tika-core/src/main/java/org/apache/tika/pipes/fetchiterator/EmptyFetchIterator.java
@@ -14,16 +14,22 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.tika.fetcher;
+package org.apache.tika.pipes.fetchiterator;
-import org.apache.tika.exception.TikaException;
+import org.apache.tika.config.InitializableProblemHandler;
+import org.apache.tika.config.Param;
+import org.apache.tika.exception.TikaConfigException;
-/**
- * If something goes wrong in parsing the fetcher string
- */
-public class FetcherStringException extends TikaException {
+import java.io.IOException;
+
+import java.util.Map;
+import java.util.concurrent.TimeoutException;
+
+public class EmptyFetchIterator extends FetchIterator {
+
+
+ @Override
+ protected void enqueue() throws IOException, TimeoutException, InterruptedException {
- public FetcherStringException(String msg) {
- super(msg);
}
}
diff --git a/tika-core/src/main/java/org/apache/tika/pipes/fetchiterator/FetchIterator.java b/tika-core/src/main/java/org/apache/tika/pipes/fetchiterator/FetchIterator.java
new file mode 100644
index 0000000..b11615f
--- /dev/null
+++ b/tika-core/src/main/java/org/apache/tika/pipes/fetchiterator/FetchIterator.java
@@ -0,0 +1,135 @@
+/*
+ * 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.fetchiterator;
+
+import org.apache.tika.config.Field;
+import org.apache.tika.config.Initializable;
+import org.apache.tika.config.InitializableProblemHandler;
+import org.apache.tika.config.Param;
+import org.apache.tika.exception.TikaConfigException;
+import org.apache.tika.pipes.fetcher.FetchIdMetadataPair;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.Callable;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+/**
+ * Abstract class that handles the testing for timeouts/thread safety
+ * issues. Concrete classes implement the blocking {@link #enqueue()}.
+ * <p>
+ * This must be "called", obviously...
+ */
+public abstract class FetchIterator implements Callable<Integer>, Initializable {
+
+ public static final long DEFAULT_MAX_WAIT_MS = 300_000;
+ public static final int DEFAULT_QUEUE_SIZE = 1000;
+ public static final FetchIdMetadataPair COMPLETED_SEMAPHORE =
+ new FetchIdMetadataPair(null, null);
+
+ private long maxWaitMs = DEFAULT_MAX_WAIT_MS;
+ private int numConsumers = -1;
+ private ArrayBlockingQueue<FetchIdMetadataPair> queue = null;
+ private int queueSize = DEFAULT_QUEUE_SIZE;
+ private String fetcherName;
+ private int added = 0;
+ public FetchIterator() {
+
+ }
+
+ public FetchIterator(String fetcherName) {
+ this.fetcherName = fetcherName;
+ }
+
+ /**
+ * This must be called before 'calling' this object.
+ * @param numConsumers
+ */
+ public ArrayBlockingQueue<FetchIdMetadataPair> init(int numConsumers) {
+ this.queue = new ArrayBlockingQueue<>(queueSize);
+ this.numConsumers = numConsumers;
+ return queue;
+ }
+
+ @Field
+ public void setFetcherName(String fetcherName) {
+ this.fetcherName = fetcherName;
+ }
+
+ public String getFetcherName() {
+ return fetcherName;
+ }
+
+
+ @Field
+ public void setMaxWaitMs(long maxWaitMs) {
+ this.maxWaitMs = maxWaitMs;
+ }
+
+ @Field
+ public void setQueueSize(int queueSize) {
+ this.queueSize = queueSize;
+ }
+ @Override
+ public Integer call() throws Exception {
+ if (queue == null || numConsumers < 0) {
+ throw new IllegalStateException("Must call 'init' before calling this object");
+ }
+
+ enqueue();
+ for (int i = 0; i < numConsumers; i++) {
+ try {
+ tryToAdd(COMPLETED_SEMAPHORE);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ return added;
+ }
+
+ protected abstract void enqueue() throws IOException, TimeoutException, InterruptedException;
+
+ protected void tryToAdd(FetchIdMetadataPair p) throws InterruptedException, TimeoutException {
+ if (p != COMPLETED_SEMAPHORE) {
+ added++;
+ }
+ boolean offered = queue.offer(p, maxWaitMs, TimeUnit.MILLISECONDS);
+ if (!offered) {
+ throw new TimeoutException("timed out while offering");
+ }
+ }
+
+ @Override
+ public void initialize(Map<String, Param> params) throws TikaConfigException {
+ //no-op
+ }
+
+ @Override
+ public void checkInitialization(InitializableProblemHandler problemHandler) throws TikaConfigException {
+ //no-op
+ }
+
+ protected static void mustNotBeEmpty(String paramName, String paramValue) {
+ if (paramValue == null || paramValue.trim().equals("")) {
+ throw new IllegalArgumentException("parameter '"+
+ paramName + "' must be set in the config file");
+ }
+ }
+
+}
diff --git a/tika-core/src/main/java/org/apache/tika/fetcher/FileSystemFetchIterator.java b/tika-core/src/main/java/org/apache/tika/pipes/fetchiterator/FileSystemFetchIterator.java
similarity index 69%
rename from tika-core/src/main/java/org/apache/tika/fetcher/FileSystemFetchIterator.java
rename to tika-core/src/main/java/org/apache/tika/pipes/fetchiterator/FileSystemFetchIterator.java
index 5edf83d..35e2ea8 100644
--- a/tika-core/src/main/java/org/apache/tika/fetcher/FileSystemFetchIterator.java
+++ b/tika-core/src/main/java/org/apache/tika/pipes/fetchiterator/FileSystemFetchIterator.java
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.tika.fetcher;
+package org.apache.tika.pipes.fetchiterator;
import org.apache.tika.config.Field;
import org.apache.tika.config.Initializable;
@@ -22,6 +22,8 @@ import org.apache.tika.config.InitializableProblemHandler;
import org.apache.tika.config.Param;
import org.apache.tika.exception.TikaConfigException;
import org.apache.tika.metadata.Metadata;
+import org.apache.tika.pipes.fetcher.FetchId;
+import org.apache.tika.pipes.fetcher.FetchIdMetadataPair;
import java.io.IOException;
import java.nio.file.FileVisitResult;
@@ -30,33 +32,21 @@ import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.attribute.BasicFileAttributes;
-import java.rmi.ServerRuntimeException;
import java.util.Map;
import java.util.concurrent.TimeoutException;
-public class FileSystemFetchIterator extends FetchIterator implements Initializable {
+public class FileSystemFetchIterator
+ extends FetchIterator implements Initializable {
- private static final String NAME = "fs";
private Path basePath;
- private String fetchPrefix;
public FileSystemFetchIterator() {
- super(NAME);
- }
- public FileSystemFetchIterator(String fetchPrefix, Path basePath) {
- super(NAME);
- this.fetchPrefix = fetchPrefix;
- this.basePath = basePath;
}
- /**
- * fetchPrefix not including the colon (:), e.g. "fs"
- * @param fetchPrefix
- */
- @Field
- public void setFetchPrefix(String fetchPrefix) {
- this.fetchPrefix = fetchPrefix;
+ public FileSystemFetchIterator(String fetcherName, Path basePath) {
+ super(fetcherName);
+ this.basePath = basePath;
}
@Field
@@ -65,10 +55,14 @@ public class FileSystemFetchIterator extends FetchIterator implements Initializa
}
@Override
- protected void enqueue() throws IOException, TimeoutException {
-
+ protected void enqueue() throws InterruptedException, IOException, TimeoutException {
+ if (! Files.isDirectory(basePath)) {
+ throw new IllegalArgumentException("\"basePath\" directory does not exist: " +
+ basePath.toAbsolutePath());
+ }
+ String fetcherName = getFetcherName();
try {
- Files.walkFileTree(basePath, new FSFileVisitor());
+ Files.walkFileTree(basePath, new FSFileVisitor(fetcherName));
} catch (IOException e) {
Throwable cause = e.getCause();
if (cause != null && cause instanceof TimeoutException) {
@@ -76,38 +70,28 @@ public class FileSystemFetchIterator extends FetchIterator implements Initializa
}
throw e;
}
- try {
- tryToAdd(POISON);
- } catch (InterruptedException e) {
- throw new RuntimeException(e);
- }
}
- @Override
- public void initialize(Map<String, Param> params) throws TikaConfigException {
- //no-op
- }
@Override
public void checkInitialization(InitializableProblemHandler problemHandler) throws TikaConfigException {
+
+ //ignore problem handler. These should all be fatal.
if (basePath == null) {
throw new TikaConfigException("Must specify a \"basePath\"");
}
- if (! Files.isDirectory(basePath)) {
- throw new TikaConfigException("\"root\" directory does not exist: " +
- basePath.toAbsolutePath());
- }
- if (fetchPrefix == null || fetchPrefix.trim().length() == 0) {
- throw new TikaConfigException("\"fetchPrefix\" must be specified and must be not blank");
- }
- if (fetchPrefix.contains(":")) {
- throw new TikaConfigException("\"fetchPrefix\" must not contain a colon (:)");
+ if (getFetcherName() == null || getFetcherName().trim().length() == 0) {
+ throw new TikaConfigException("\"fetcherName\" must be specified and must be not blank");
}
}
private class FSFileVisitor implements FileVisitor<Path> {
+ private final String fetcherName;
+ private FSFileVisitor(String fetcherName) {
+ this.fetcherName = fetcherName;
+ }
@Override
public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) throws IOException {
return FileVisitResult.CONTINUE;
@@ -116,10 +100,10 @@ public class FileSystemFetchIterator extends FetchIterator implements Initializa
@Override
public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
String relPath = basePath.relativize(file).toString();
- String fetcherString = fetchPrefix + ":" + relPath;
try {
- tryToAdd(new FetchMetadataPair(fetcherString, new Metadata()));
+ tryToAdd(new FetchIdMetadataPair(
+ new FetchId(fetcherName, relPath), new Metadata()));
} catch (TimeoutException e) {
throw new IOException(e);
} catch (InterruptedException e) {
diff --git a/tika-core/src/test/java/org/apache/tika/config/TikaConfigTest.java b/tika-core/src/test/java/org/apache/tika/config/TikaConfigTest.java
index 5ebfdf5..f2cbf73 100644
--- a/tika-core/src/test/java/org/apache/tika/config/TikaConfigTest.java
+++ b/tika-core/src/test/java/org/apache/tika/config/TikaConfigTest.java
@@ -43,6 +43,13 @@ import org.apache.tika.parser.ErrorParser;
import org.apache.tika.parser.Parser;
import org.apache.tika.parser.ParserDecorator;
import org.apache.tika.parser.multiple.FallbackParser;
+import org.apache.tika.pipes.emitter.Emitter;
+import org.apache.tika.pipes.emitter.EmitterManager;
+import org.apache.tika.pipes.fetcher.Fetcher;
+import org.apache.tika.pipes.fetcher.FetcherManager;
+import org.apache.tika.pipes.fetcher.FileSystemFetcher;
+import org.apache.tika.pipes.fetchiterator.FetchIterator;
+import org.apache.tika.pipes.fetchiterator.FileSystemFetchIterator;
import org.apache.tika.utils.XMLReaderUtils;
import org.junit.Test;
@@ -355,4 +362,64 @@ public class TikaConfigTest extends AbstractTikaConfigTest {
}
+ @Test
+ public void testFetchers() throws Exception {
+ TikaConfig config = getConfig("fetchers-config.xml");
+ FetcherManager m = config.getFetcherManager();
+ Fetcher f1 = m.getFetcher("fs1");
+ assertEquals(Paths.get("/my/base/path1"),
+ ((FileSystemFetcher)f1).getBasePath()
+ );
+
+ Fetcher f2 = m.getFetcher("fs2");
+ assertEquals(Paths.get("/my/base/path2"),
+ ((FileSystemFetcher)f2).getBasePath()
+ );
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testDuplicateFetchers() throws Exception {
+ //can't have two fetchers with the same name
+ TikaConfig config = getConfig("fetchers-duplicate-config.xml");
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testNoNameFetchers() throws Exception {
+ //can't have two fetchers with the same name
+ TikaConfig config = getConfig("fetchers-noname-config.xml");
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testNoBasePathFetchers() throws Exception {
+ //can't have an fs fetcher with no basepath specified
+ TikaConfig config = getConfig("fetchers-nobasepath-config.xml");
+ }
+
+ @Test
+ public void testEmitters() throws Exception {
+ EmitterManager emitterManager = getConfig("emitters-config.xml").getEmitterManager();
+ Emitter em1 = emitterManager.getEmitter("em1");
+ assertNotNull(em1);
+ Emitter em2 = emitterManager.getEmitter("em2");
+ assertNotNull(em2);
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testDuplicateEmitters() throws Exception {
+ getConfig("emitters-duplicate-config.xml").getEmitterManager();
+ }
+
+ @Test
+ public void testFetchIterator() throws Exception {
+ FetchIterator f = getConfig("fetch-iterator-config.xml")
+ .getFetchIterator();
+ assertEquals("fs1", f.getFetcherName());
+ }
+
+ @Test(expected = TikaConfigException.class)
+ public void testMultipleFetchIterators() throws Exception {
+ FetchIterator f = getConfig("fetch-iterator-multiple-config.xml")
+ .getFetchIterator();
+ assertEquals("fs1", f.getFetcherName());
+ }
}
\ No newline at end of file
diff --git a/tika-core/src/main/java/org/apache/tika/fetcher/FetcherStringException.java b/tika-core/src/test/java/org/apache/tika/pipes/emitter/MockEmitter.java
similarity index 75%
rename from tika-core/src/main/java/org/apache/tika/fetcher/FetcherStringException.java
rename to tika-core/src/test/java/org/apache/tika/pipes/emitter/MockEmitter.java
index 9f1e93b..f00d815 100644
--- a/tika-core/src/main/java/org/apache/tika/fetcher/FetcherStringException.java
+++ b/tika-core/src/test/java/org/apache/tika/pipes/emitter/MockEmitter.java
@@ -14,16 +14,18 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.tika.fetcher;
+package org.apache.tika.pipes.emitter;
import org.apache.tika.exception.TikaException;
+import org.apache.tika.metadata.Metadata;
-/**
- * If something goes wrong in parsing the fetcher string
- */
-public class FetcherStringException extends TikaException {
+import java.io.IOException;
+import java.util.List;
+
+public class MockEmitter extends AbstractEmitter {
+
+ @Override
+ public void emit(List<Metadata> metadataList) throws IOException, TikaException {
- public FetcherStringException(String msg) {
- super(msg);
}
}
diff --git a/tika-core/src/test/java/org/apache/tika/pipes/fetcher/FileSystemFetcherTest.java b/tika-core/src/test/java/org/apache/tika/pipes/fetcher/FileSystemFetcherTest.java
new file mode 100644
index 0000000..60e14e3
--- /dev/null
+++ b/tika-core/src/test/java/org/apache/tika/pipes/fetcher/FileSystemFetcherTest.java
@@ -0,0 +1,54 @@
+/*
+ * 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;
+
+import org.apache.tika.config.InitializableProblemHandler;
+import org.junit.Test;
+
+
+import java.nio.file.InvalidPathException;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+
+public class FileSystemFetcherTest {
+
+ @Test
+ public void testDescendant() throws Exception {
+
+ Path root = Paths.get("/ab/cd/");
+ Path descendant = root.resolve("ef/gh/ij.pdf");
+ assertTrue(FileSystemFetcher.isDescendant(root, descendant));
+
+ descendant = Paths.get("/cd/ef.pdf");
+ assertFalse(FileSystemFetcher.isDescendant(root, descendant));
+
+ descendant = root.resolve("../../ij.pdf");
+ assertFalse(FileSystemFetcher.isDescendant(root, descendant));
+ }
+
+ @Test(expected = InvalidPathException.class)
+ public void testNullByte() throws Exception {
+ FileSystemFetcher f = new FileSystemFetcher();
+ f.setBasePath("bad\u0000path");
+ f.setName("fs");
+ f.checkInitialization(InitializableProblemHandler.IGNORE);
+ }
+}
diff --git a/tika-core/src/test/java/org/apache/tika/fetcher/FileSystemFetchIteratorTest.java b/tika-core/src/test/java/org/apache/tika/pipes/fetchiterator/FileSystemFetchIteratorTest.java
similarity index 77%
rename from tika-core/src/test/java/org/apache/tika/fetcher/FileSystemFetchIteratorTest.java
rename to tika-core/src/test/java/org/apache/tika/pipes/fetchiterator/FileSystemFetchIteratorTest.java
index 5349fcb..f23489a 100644
--- a/tika-core/src/test/java/org/apache/tika/fetcher/FileSystemFetchIteratorTest.java
+++ b/tika-core/src/test/java/org/apache/tika/pipes/fetchiterator/FileSystemFetchIteratorTest.java
@@ -14,8 +14,9 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.tika.fetcher;
+package org.apache.tika.pipes.fetchiterator;
+import org.apache.tika.pipes.fetcher.FetchIdMetadataPair;
import org.junit.Test;
import java.io.IOException;
@@ -25,6 +26,7 @@ import java.nio.file.Paths;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
+import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -37,28 +39,38 @@ import static org.junit.Assert.assertEquals;
public class FileSystemFetchIteratorTest {
- @Test
+ @Test(timeout = 30000)
public void testBasic() throws Exception {
Path root = Paths.get(".");
- String fetchPrefix = "fs";
+
+ List<Path> files = listFiles(root);
+ Set<String> truthSet = new HashSet<>();
+ for (Path p : files) {
+ String fetchString = root.relativize(p).toString();
+ truthSet.add(fetchString);
+ }
+
+ String fetcherName = "fs";
ExecutorService es = Executors.newFixedThreadPool(1);
ExecutorCompletionService cs = new ExecutorCompletionService(es);
- FetchIterator it = new FileSystemFetchIterator(fetchPrefix, root);
+ FetchIterator it = new FileSystemFetchIterator(fetcherName, root);
+ it.setQueueSize(20000);
+ ArrayBlockingQueue<FetchIdMetadataPair> q = it.init(1);
cs.submit(it);
- Set<String> iteratorSet = new HashSet<>();
- int i = 0;
- for (FetchMetadataPair p : it) {
- iteratorSet.add(p.getFetcherString());
- }
+
+
Future f = cs.take();
f.get();
- List<Path> files = listFiles(root);
- Set<String> truthSet = new HashSet<>();
- for (Path p : files) {
- String fetchString = fetchPrefix+":"+root.relativize(p);
- truthSet.add(fetchString);
+
+ Set<String> iteratorSet = new HashSet<>();
+ for (FetchIdMetadataPair p : q) {
+ if (p == FetchIterator.COMPLETED_SEMAPHORE) {
+ break;
+ }
+ iteratorSet.add(p.getFetchId().getFetchKey());
}
+
assertEquals(truthSet, iteratorSet);
}
diff --git a/tika-fetchers/tika-fetcher-s3/src/test/resources/tika-config-s3.xml b/tika-core/src/test/resources/org/apache/tika/config/emitters-config.xml
similarity index 69%
copy from tika-fetchers/tika-fetcher-s3/src/test/resources/tika-config-s3.xml
copy to tika-core/src/test/resources/org/apache/tika/config/emitters-config.xml
index 39a2ce8..73218c4 100644
--- a/tika-fetchers/tika-fetcher-s3/src/test/resources/tika-config-s3.xml
+++ b/tika-core/src/test/resources/org/apache/tika/config/emitters-config.xml
@@ -16,12 +16,16 @@
limitations under the License.
-->
<properties>
- <fetchers>
- <fetcher class="org.apache.tika.fetcher.s3.S3Fetcher">
+ <emitters>
+ <emitter class="org.apache.tika.pipes.emitter.MockEmitter">
<params>
- <param name="region" type="string">us-east-1</param>
- <param name="profile" type="string">my_profile</param>
+ <param name="name" type="string">em1</param>
</params>
- </fetcher>
- </fetchers>
-</properties>
\ No newline at end of file
+ </emitter>
+ <emitter class="org.apache.tika.pipes.emitter.MockEmitter">
+ <params>
+ <param name="name" type="string">em2</param>
+ </params>
+ </emitter>
+ </emitters>
+</properties>
diff --git a/tika-fetchers/tika-fetcher-s3/src/test/resources/tika-config-s3.xml b/tika-core/src/test/resources/org/apache/tika/config/emitters-duplicate-config.xml
similarity index 69%
copy from tika-fetchers/tika-fetcher-s3/src/test/resources/tika-config-s3.xml
copy to tika-core/src/test/resources/org/apache/tika/config/emitters-duplicate-config.xml
index 39a2ce8..25dd7ce 100644
--- a/tika-fetchers/tika-fetcher-s3/src/test/resources/tika-config-s3.xml
+++ b/tika-core/src/test/resources/org/apache/tika/config/emitters-duplicate-config.xml
@@ -16,12 +16,16 @@
limitations under the License.
-->
<properties>
- <fetchers>
- <fetcher class="org.apache.tika.fetcher.s3.S3Fetcher">
+ <emitters>
+ <emitter class="org.apache.tika.pipes.emitter.MockEmitter">
<params>
- <param name="region" type="string">us-east-1</param>
- <param name="profile" type="string">my_profile</param>
+ <param name="name" type="string">em1</param>
</params>
- </fetcher>
- </fetchers>
-</properties>
\ No newline at end of file
+ </emitter>
+ <emitter class="org.apache.tika.pipes.emitter.MockEmitter">
+ <params>
+ <param name="name" type="string">em1</param>
+ </params>
+ </emitter>
+ </emitters>
+</properties>
diff --git a/tika-fetchers/tika-fetcher-s3/src/test/resources/tika-config-s3.xml b/tika-core/src/test/resources/org/apache/tika/config/fetch-iterator-config.xml
similarity index 73%
copy from tika-fetchers/tika-fetcher-s3/src/test/resources/tika-config-s3.xml
copy to tika-core/src/test/resources/org/apache/tika/config/fetch-iterator-config.xml
index 39a2ce8..7f3dd41 100644
--- a/tika-fetchers/tika-fetcher-s3/src/test/resources/tika-config-s3.xml
+++ b/tika-core/src/test/resources/org/apache/tika/config/fetch-iterator-config.xml
@@ -16,12 +16,12 @@
limitations under the License.
-->
<properties>
- <fetchers>
- <fetcher class="org.apache.tika.fetcher.s3.S3Fetcher">
+ <fetchIterators>
+ <fetchIterator class="org.apache.tika.pipes.fetchiterator.FileSystemFetchIterator">
<params>
- <param name="region" type="string">us-east-1</param>
- <param name="profile" type="string">my_profile</param>
+ <param name="fetcherName" type="string">fs1</param>
+ <param name="basePath" type="string">/my/base/path1</param>
</params>
- </fetcher>
- </fetchers>
-</properties>
\ No newline at end of file
+ </fetchIterator>
+ </fetchIterators>
+</properties>
diff --git a/tika-fetchers/tika-fetcher-s3/src/test/resources/tika-config-s3.xml b/tika-core/src/test/resources/org/apache/tika/config/fetch-iterator-multiple-config.xml
similarity index 58%
copy from tika-fetchers/tika-fetcher-s3/src/test/resources/tika-config-s3.xml
copy to tika-core/src/test/resources/org/apache/tika/config/fetch-iterator-multiple-config.xml
index 39a2ce8..4417fbc 100644
--- a/tika-fetchers/tika-fetcher-s3/src/test/resources/tika-config-s3.xml
+++ b/tika-core/src/test/resources/org/apache/tika/config/fetch-iterator-multiple-config.xml
@@ -16,12 +16,18 @@
limitations under the License.
-->
<properties>
- <fetchers>
- <fetcher class="org.apache.tika.fetcher.s3.S3Fetcher">
+ <fetchIterators>
+ <fetchIterator class="org.apache.tika.pipes.fetchiterator.FileSystemFetchIterator">
<params>
- <param name="region" type="string">us-east-1</param>
- <param name="profile" type="string">my_profile</param>
+ <param name="fetcherName" type="string">fs1</param>
+ <param name="basePath" type="string">/my/base/path1</param>
</params>
- </fetcher>
- </fetchers>
-</properties>
\ No newline at end of file
+ </fetchIterator>
+ <fetchIterator class="org.apache.tika.pipes.fetchiterator.FileSystemFetchIterator">
+ <params>
+ <param name="fetcherName" type="string">fs2</param>
+ <param name="basePath" type="string">/my/base/path2</param>
+ </params>
+ </fetchIterator>
+ </fetchIterators>
+</properties>
diff --git a/tika-fetchers/tika-fetcher-s3/src/test/resources/tika-config-s3.xml b/tika-core/src/test/resources/org/apache/tika/config/fetchers-config.xml
similarity index 65%
copy from tika-fetchers/tika-fetcher-s3/src/test/resources/tika-config-s3.xml
copy to tika-core/src/test/resources/org/apache/tika/config/fetchers-config.xml
index 39a2ce8..647c676 100644
--- a/tika-fetchers/tika-fetcher-s3/src/test/resources/tika-config-s3.xml
+++ b/tika-core/src/test/resources/org/apache/tika/config/fetchers-config.xml
@@ -17,11 +17,17 @@
-->
<properties>
<fetchers>
- <fetcher class="org.apache.tika.fetcher.s3.S3Fetcher">
+ <fetcher class="org.apache.tika.pipes.fetcher.FileSystemFetcher">
<params>
- <param name="region" type="string">us-east-1</param>
- <param name="profile" type="string">my_profile</param>
+ <param name="name" type="string">fs1</param>
+ <param name="basePath" type="string">/my/base/path1</param>
+ </params>
+ </fetcher>
+ <fetcher class="org.apache.tika.pipes.fetcher.FileSystemFetcher">
+ <params>
+ <param name="name" type="string">fs2</param>
+ <param name="basePath" type="string">/my/base/path2</param>
</params>
</fetcher>
</fetchers>
-</properties>
\ No newline at end of file
+</properties>
diff --git a/tika-fetchers/tika-fetcher-s3/src/test/resources/tika-config-s3.xml b/tika-core/src/test/resources/org/apache/tika/config/fetchers-duplicate-config.xml
similarity index 65%
copy from tika-fetchers/tika-fetcher-s3/src/test/resources/tika-config-s3.xml
copy to tika-core/src/test/resources/org/apache/tika/config/fetchers-duplicate-config.xml
index 39a2ce8..314ed35 100644
--- a/tika-fetchers/tika-fetcher-s3/src/test/resources/tika-config-s3.xml
+++ b/tika-core/src/test/resources/org/apache/tika/config/fetchers-duplicate-config.xml
@@ -17,11 +17,17 @@
-->
<properties>
<fetchers>
- <fetcher class="org.apache.tika.fetcher.s3.S3Fetcher">
+ <fetcher class="org.apache.tika.pipes.fetcher.FileSystemFetcher">
<params>
- <param name="region" type="string">us-east-1</param>
- <param name="profile" type="string">my_profile</param>
+ <param name="name" type="string">fs1</param>
+ <param name="basePath" type="string">/my/base/path1</param>
+ </params>
+ </fetcher>
+ <fetcher class="org.apache.tika.pipes.fetcher.FileSystemFetcher">
+ <params>
+ <param name="name" type="string">fs1</param>
+ <param name="basePath" type="string">/my/base/path2</param>
</params>
</fetcher>
</fetchers>
-</properties>
\ No newline at end of file
+</properties>
diff --git a/tika-fetchers/tika-fetcher-s3/src/test/resources/tika-config-s3.xml b/tika-core/src/test/resources/org/apache/tika/config/fetchers-nobasepath-config.xml
similarity index 68%
copy from tika-fetchers/tika-fetcher-s3/src/test/resources/tika-config-s3.xml
copy to tika-core/src/test/resources/org/apache/tika/config/fetchers-nobasepath-config.xml
index 39a2ce8..de954cd 100644
--- a/tika-fetchers/tika-fetcher-s3/src/test/resources/tika-config-s3.xml
+++ b/tika-core/src/test/resources/org/apache/tika/config/fetchers-nobasepath-config.xml
@@ -17,11 +17,16 @@
-->
<properties>
<fetchers>
- <fetcher class="org.apache.tika.fetcher.s3.S3Fetcher">
+ <fetcher class="org.apache.tika.pipes.fetcher.FileSystemFetcher">
<params>
- <param name="region" type="string">us-east-1</param>
- <param name="profile" type="string">my_profile</param>
+ <param name="name" type="string">fs1</param>
+ <param name="basePath" type="string">/my/base/path1</param>
+ </params>
+ </fetcher>
+ <fetcher class="org.apache.tika.pipes.fetcher.FileSystemFetcher">
+ <params>
+ <param name="name" type="string">fs2</param>
</params>
</fetcher>
</fetchers>
-</properties>
\ No newline at end of file
+</properties>
diff --git a/tika-fetchers/tika-fetcher-s3/src/test/resources/tika-config-s3.xml b/tika-core/src/test/resources/org/apache/tika/config/fetchers-noname-config.xml
similarity index 68%
copy from tika-fetchers/tika-fetcher-s3/src/test/resources/tika-config-s3.xml
copy to tika-core/src/test/resources/org/apache/tika/config/fetchers-noname-config.xml
index 39a2ce8..bb92c42 100644
--- a/tika-fetchers/tika-fetcher-s3/src/test/resources/tika-config-s3.xml
+++ b/tika-core/src/test/resources/org/apache/tika/config/fetchers-noname-config.xml
@@ -17,11 +17,16 @@
-->
<properties>
<fetchers>
- <fetcher class="org.apache.tika.fetcher.s3.S3Fetcher">
+ <fetcher class="org.apache.tika.pipes.fetcher.FileSystemFetcher">
<params>
- <param name="region" type="string">us-east-1</param>
- <param name="profile" type="string">my_profile</param>
+ <param name="name" type="string">fs1</param>
+ <param name="basePath" type="string">/my/base/path1</param>
+ </params>
+ </fetcher>
+ <fetcher class="org.apache.tika.pipes.fetcher.FileSystemFetcher">
+ <params>
+ <param name="basePath" type="string">/my/base/path2</param>
</params>
</fetcher>
</fetchers>
-</properties>
\ No newline at end of file
+</properties>
diff --git a/tika-fetchers/tika-fetcher-jdbc/pom.xml b/tika-fetchers/tika-fetcher-jdbc/pom.xml
deleted file mode 100644
index 06f79d5..0000000
--- a/tika-fetchers/tika-fetcher-jdbc/pom.xml
+++ /dev/null
@@ -1,122 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
- 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.
--->
-<project xmlns="http://maven.apache.org/POM/4.0.0"
- xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
- xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
- <parent>
- <artifactId>tika-fetchers</artifactId>
- <groupId>org.apache.tika</groupId>
- <version>2.0.0-SNAPSHOT</version>
- </parent>
- <modelVersion>4.0.0</modelVersion>
-
- <artifactId>jdbc-fetcher</artifactId>
-
- <dependencies>
- <dependency>
- <groupId>${project.groupId}</groupId>
- <artifactId>tika-core</artifactId>
- <version>${project.version}</version>
- <scope>provided</scope>
- </dependency>
- <dependency>
- <groupId>com.h2database</groupId>
- <artifactId>h2</artifactId>
- <version>${h2.version}</version>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>junit</groupId>
- <artifactId>junit</artifactId>
- <scope>test</scope>
- </dependency>
- </dependencies>
- <build>
- <plugins>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-jar-plugin</artifactId>
- <configuration>
- <archive>
- <manifestEntries>
- <Automatic-Module-Name>org.apache.tika.fetcher.jdbc</Automatic-Module-Name>
- </manifestEntries>
- </archive>
- </configuration>
- <executions>
- <execution>
- <goals>
- <goal>test-jar</goal>
- </goals>
- </execution>
- </executions>
- </plugin>
- <plugin>
- <artifactId>maven-shade-plugin</artifactId>
- <version>${maven.shade.version}</version>
- <executions>
- <execution>
- <phase>package</phase>
- <goals>
- <goal>shade</goal>
- </goals>
- <configuration>
- <createDependencyReducedPom>
- false
- </createDependencyReducedPom>
- <!-- <filters> -->
- <filters>
- <filter>
- <artifact>*:*</artifact>
- <excludes>
- <exclude>META-INF/*</exclude>
- <exclude>LICENSE.txt</exclude>
- <exclude>NOTICE.txt</exclude>
- </excludes>
- </filter>
- </filters>
- <transformers>
- <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
- <mainClass>org.apache.tika.eval.app.TikaEvalCLI</mainClass>
- </transformer>
-
- <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer" />
- <transformer implementation="org.apache.maven.plugins.shade.resource.IncludeResourceTransformer">
- <resource>META-INF/LICENSE</resource>
- <file>target/classes/META-INF/LICENSE</file>
- </transformer>
- <transformer implementation="org.apache.maven.plugins.shade.resource.IncludeResourceTransformer">
- <resource>META-INF/NOTICE</resource>
- <file>target/classes/META-INF/NOTICE</file>
- </transformer>
- <transformer implementation="org.apache.maven.plugins.shade.resource.IncludeResourceTransformer">
- <resource>META-INF/DEPENDENCIES</resource>
- <file>target/classes/META-INF/DEPENDENCIES</file>
- </transformer>
- </transformers>
- </configuration>
- </execution>
- </executions>
- </plugin>
-
- </plugins>
- </build>
-
-</project>
\ No newline at end of file
diff --git a/tika-fetchers/tika-fetcher-jdbc/src/main/java/org/apache/tika/fetcher/jdbc/JDBCFetcher.java b/tika-fetchers/tika-fetcher-jdbc/src/main/java/org/apache/tika/fetcher/jdbc/JDBCFetcher.java
deleted file mode 100644
index 25152a9..0000000
--- a/tika-fetchers/tika-fetcher-jdbc/src/main/java/org/apache/tika/fetcher/jdbc/JDBCFetcher.java
+++ /dev/null
@@ -1,100 +0,0 @@
-/*
- * 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.fetcher.jdbc;
-
-import org.apache.tika.config.Field;
-import org.apache.tika.config.Initializable;
-import org.apache.tika.config.InitializableProblemHandler;
-import org.apache.tika.config.Param;
-import org.apache.tika.exception.TikaConfigException;
-import org.apache.tika.exception.TikaException;
-import org.apache.tika.fetcher.Fetcher;
-import org.apache.tika.metadata.Metadata;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
-public class JDBCFetcher implements Fetcher, Initializable {
-
- private int requestPoolSize = 10;
- private String table;
- private List<String> metadataColumns = new ArrayList<>();
- private String binaryField;
- private String connectionString;
-
-
- @Override
- public void initialize(Map<String, Param> params) throws TikaConfigException {
- //TODO: init prepared
- }
-
- @Override
- public void checkInitialization(InitializableProblemHandler problemHandler) throws TikaConfigException {
- //no-op
- }
-
- @Override
- public Set<String> getSupportedPrefixes() {
- return null;
- }
-
- @Override
- public InputStream fetch(String fetcherString, Metadata metadata) throws TikaException, IOException {
- return null;
- }
-
- /**
- * The prepared inserts are pooled. The default is 10.
- * This sets how big this pool should be.
- * @param requestPoolSize
- */
- @Field
- public void setRequestPoolSize(int requestPoolSize) {
- this.requestPoolSize = requestPoolSize;
- }
-
- @Field
- public void setJDBCConnectionString(String connectionString) {
- this.connectionString = connectionString;
- }
-
- @Field
- public void setTable(String table) {
- this.table = table;
- }
-
- @Field
- public void setMetadataFields(List<String> cols) {
-
- }
-
- /**
- * If there's a blob or text field that you
- * want Tika to parse.
- *
- * This is optional.
- * @param binaryField
- */
- @Field
- public void setBinaryField(String binaryField) {
-
- }
-}
diff --git a/tika-pipes/pom.xml b/tika-pipes/pom.xml
new file mode 100644
index 0000000..b90d3c3
--- /dev/null
+++ b/tika-pipes/pom.xml
@@ -0,0 +1,41 @@
+<!--
+ 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <packaging>pom</packaging>
+
+
+ <parent>
+ <groupId>org.apache.tika</groupId>
+ <artifactId>tika-parent</artifactId>
+ <version>2.0.0-SNAPSHOT</version>
+ <relativePath>../tika-parent/pom.xml</relativePath>
+ </parent>
+
+ <artifactId>tika-pipes</artifactId>
+ <name>Apache Tika pipes</name>
+ <url>http://tika.apache.org/</url>
+
+ <modules>
+ <module>tika-httpclient-commons</module>
+ <module>tika-fetchers</module>
+ <module>tika-emitters</module>
+ <module>tika-fetch-iterators</module>
+ <module>tika-pipes-integration-tests</module>
+ </modules>
+
+</project>
diff --git a/tika-emitters/pom.xml b/tika-pipes/tika-emitters/pom.xml
similarity index 89%
copy from tika-emitters/pom.xml
copy to tika-pipes/tika-emitters/pom.xml
index 42bdf6b..bfb55c8 100644
--- a/tika-emitters/pom.xml
+++ b/tika-pipes/tika-emitters/pom.xml
@@ -22,9 +22,9 @@
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<groupId>org.apache.tika</groupId>
- <artifactId>tika-parent</artifactId>
+ <artifactId>tika-pipes</artifactId>
<version>2.0.0-SNAPSHOT</version>
- <relativePath>../tika-parent/pom.xml</relativePath>
+ <relativePath>../pom.xml</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
@@ -35,9 +35,7 @@
<modules>
<module>tika-emitter-fs</module>
- <module>tika-httpclient-commons</module>
<module>tika-emitter-solr</module>
-
</modules>
diff --git a/tika-emitters/tika-emitter-fs/pom.xml b/tika-pipes/tika-emitters/tika-emitter-fs/pom.xml
similarity index 100%
rename from tika-emitters/tika-emitter-fs/pom.xml
rename to tika-pipes/tika-emitters/tika-emitter-fs/pom.xml
diff --git a/tika-emitters/tika-emitter-fs/src/main/java/org/apache/tika/emitter/fs/FileSystemEmitter.java b/tika-pipes/tika-emitters/tika-emitter-fs/src/main/java/org/apache/tika/emitter/fs/FileSystemEmitter.java
similarity index 69%
rename from tika-emitters/tika-emitter-fs/src/main/java/org/apache/tika/emitter/fs/FileSystemEmitter.java
rename to tika-pipes/tika-emitters/tika-emitter-fs/src/main/java/org/apache/tika/emitter/fs/FileSystemEmitter.java
index 0acae53..e42b57c 100644
--- a/tika-emitters/tika-emitter-fs/src/main/java/org/apache/tika/emitter/fs/FileSystemEmitter.java
+++ b/tika-pipes/tika-emitters/tika-emitter-fs/src/main/java/org/apache/tika/emitter/fs/FileSystemEmitter.java
@@ -1,8 +1,24 @@
+/*
+ * 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.emitter.fs;
import org.apache.tika.config.Field;
-import org.apache.tika.emitter.Emitter;
-import org.apache.tika.emitter.TikaEmitterException;
+import org.apache.tika.pipes.emitter.Emitter;
+import org.apache.tika.pipes.emitter.TikaEmitterException;
import org.apache.tika.exception.TikaException;
import org.apache.tika.metadata.Metadata;
import org.apache.tika.metadata.TikaCoreProperties;
@@ -25,12 +41,12 @@ public class FileSystemEmitter implements Emitter {
private String fileExtension = "json";
@Override
- public Set<String> getSupported() {
- return Collections.singleton(name);
+ public String getName() {
+ return name;
}
@Override
- public void emit(String emitterName, List<Metadata> metadataList) throws IOException, TikaException {
+ public void emit(List<Metadata> metadataList) throws IOException, TikaException {
Path output;
if (metadataList == null || metadataList.size() == 0) {
throw new TikaEmitterException("metadata list must not be null or of size 0");
diff --git a/tika-emitters/tika-emitter-solr/pom.xml b/tika-pipes/tika-emitters/tika-emitter-solr/pom.xml
similarity index 100%
copy from tika-emitters/tika-emitter-solr/pom.xml
copy to tika-pipes/tika-emitters/tika-emitter-solr/pom.xml
diff --git a/tika-emitters/tika-emitter-solr/src/main/java/org/apache/tika/emitter/solr/SolrEmitter.java b/tika-pipes/tika-emitters/tika-emitter-solr/src/main/java/org/apache/tika/emitter/solr/SolrEmitter.java
similarity index 87%
rename from tika-emitters/tika-emitter-solr/src/main/java/org/apache/tika/emitter/solr/SolrEmitter.java
rename to tika-pipes/tika-emitters/tika-emitter-solr/src/main/java/org/apache/tika/emitter/solr/SolrEmitter.java
index e77519e..df78bb7 100644
--- a/tika-emitters/tika-emitter-solr/src/main/java/org/apache/tika/emitter/solr/SolrEmitter.java
+++ b/tika-pipes/tika-emitters/tika-emitter-solr/src/main/java/org/apache/tika/emitter/solr/SolrEmitter.java
@@ -1,3 +1,19 @@
+/*
+ * 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.emitter.solr;
import com.google.gson.Gson;
@@ -8,7 +24,7 @@ import org.apache.tika.config.Field;
import org.apache.tika.config.Initializable;
import org.apache.tika.config.InitializableProblemHandler;
import org.apache.tika.config.Param;
-import org.apache.tika.emitter.Emitter;
+import org.apache.tika.pipes.emitter.Emitter;
import org.apache.tika.exception.TikaConfigException;
import org.apache.tika.exception.TikaException;
import org.apache.tika.metadata.Metadata;
@@ -19,7 +35,6 @@ import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.Map;
-import java.util.Set;
import java.util.UUID;
public class SolrEmitter implements Emitter, Initializable {
@@ -43,7 +58,12 @@ public class SolrEmitter implements Emitter, Initializable {
private int commitWithin = 100;
@Override
- public void emit(String emitterName, List<Metadata> metadataList) throws IOException,
+ public String getName() {
+ return name;
+ }
+
+ @Override
+ public void emit(List<Metadata> metadataList) throws IOException,
TikaException {
if (metadataList == null || metadataList.size() == 0) {
LOG.warn("metadataList is null or empty");
@@ -200,7 +220,7 @@ public class SolrEmitter implements Emitter, Initializable {
@Override
public void initialize(Map<String, Param> params) throws TikaConfigException {
- //TODO: build the client here
+ //TODO: build the client here?
}
@@ -209,9 +229,4 @@ public class SolrEmitter implements Emitter, Initializable {
}
- @Override
- public Set<String> getSupported() {
- return Collections.singleton(name);
- }
-
}
diff --git a/tika-emitters/tika-emitter-solr/src/test/java/org/apache/tika/emitter/solr/TestBasic.java b/tika-pipes/tika-emitters/tika-emitter-solr/src/test/java/org/apache/tika/emitter/solr/TestBasic.java
similarity index 58%
rename from tika-emitters/tika-emitter-solr/src/test/java/org/apache/tika/emitter/solr/TestBasic.java
rename to tika-pipes/tika-emitters/tika-emitter-solr/src/test/java/org/apache/tika/emitter/solr/TestBasic.java
index b374476..3df4c4a 100644
--- a/tika-emitters/tika-emitter-solr/src/test/java/org/apache/tika/emitter/solr/TestBasic.java
+++ b/tika-pipes/tika-emitters/tika-emitter-solr/src/test/java/org/apache/tika/emitter/solr/TestBasic.java
@@ -1,25 +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.emitter.solr;
import org.apache.tika.config.TikaConfig;
-import org.apache.tika.emitter.Emitter;
+import org.apache.tika.pipes.emitter.Emitter;
import org.apache.tika.metadata.Metadata;
import org.apache.tika.metadata.TikaCoreProperties;
import org.apache.tika.metadata.filter.MetadataFilter;
import org.apache.tika.sax.AbstractRecursiveParserWrapperHandler;
+import org.junit.Ignore;
import org.junit.Test;
-import java.lang.reflect.Array;
import java.util.ArrayList;
import java.util.List;
+@Ignore("requires solr to be up and running")
public class TestBasic {
@Test
public void testBasic() throws Exception {
TikaConfig tikaConfig = new TikaConfig(
TestBasic.class.getResourceAsStream("/tika-config-simple-emitter.xml"));
- Emitter emitter = tikaConfig.getEmitter();
+ Emitter emitter = tikaConfig.getEmitterManager().getEmitter("solr1");
List<Metadata> metadataList = new ArrayList<>();
Metadata m1 = new Metadata();
m1.set("id", "1");
@@ -38,6 +55,6 @@ public class TestBasic {
metadataList.add(m1);
metadataList.add(m2);
- emitter.emit("solr1", metadataList);
+ emitter.emit(metadataList);
}
}
diff --git a/tika-emitters/tika-emitter-solr/src/test/resources/log4j.properties b/tika-pipes/tika-emitters/tika-emitter-solr/src/test/resources/log4j.properties
similarity index 100%
copy from tika-emitters/tika-emitter-solr/src/test/resources/log4j.properties
copy to tika-pipes/tika-emitters/tika-emitter-solr/src/test/resources/log4j.properties
diff --git a/tika-emitters/tika-emitter-solr/src/test/resources/tika-config-simple-emitter.xml b/tika-pipes/tika-emitters/tika-emitter-solr/src/test/resources/tika-config-simple-emitter.xml
similarity index 100%
rename from tika-emitters/tika-emitter-solr/src/test/resources/tika-config-simple-emitter.xml
rename to tika-pipes/tika-emitters/tika-emitter-solr/src/test/resources/tika-config-simple-emitter.xml
diff --git a/tika-emitters/pom.xml b/tika-pipes/tika-fetch-iterators/pom.xml
similarity index 72%
rename from tika-emitters/pom.xml
rename to tika-pipes/tika-fetch-iterators/pom.xml
index 42bdf6b..38603af 100644
--- a/tika-emitters/pom.xml
+++ b/tika-pipes/tika-fetch-iterators/pom.xml
@@ -22,23 +22,22 @@
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<groupId>org.apache.tika</groupId>
- <artifactId>tika-parent</artifactId>
+ <artifactId>tika-pipes</artifactId>
<version>2.0.0-SNAPSHOT</version>
- <relativePath>../tika-parent/pom.xml</relativePath>
+ <relativePath>../pom.xml</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
- <artifactId>tika-emitters</artifactId>
+ <artifactId>tika-fetch-iterators</artifactId>
<packaging>pom</packaging>
- <name>Apache Tika emitters</name>
+ <name>Apache Tika fetch iterators</name>
<url>http://tika.apache.org/</url>
+ <!-- see also org.apache.tika.pipes.fetchiterator.FileSystemIterator
+ in tika-core if you want a file system directory crawler -->
<modules>
- <module>tika-emitter-fs</module>
- <module>tika-httpclient-commons</module>
- <module>tika-emitter-solr</module>
-
+ <module>tika-fetch-iterator-csv</module>
+ <module>tika-fetch-iterator-jdbc</module>
+ <module>tika-fetch-iterator-s3</module>
</modules>
-
-
-</project>
\ No newline at end of file
+</project>
diff --git a/tika-emitters/tika-emitter-solr/pom.xml b/tika-pipes/tika-fetch-iterators/tika-fetch-iterator-csv/pom.xml
similarity index 79%
copy from tika-emitters/tika-emitter-solr/pom.xml
copy to tika-pipes/tika-fetch-iterators/tika-fetch-iterator-csv/pom.xml
index 6aefb89..bac6ace 100644
--- a/tika-emitters/tika-emitter-solr/pom.xml
+++ b/tika-pipes/tika-fetch-iterators/tika-fetch-iterator-csv/pom.xml
@@ -21,14 +21,18 @@
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
- <artifactId>tika-emitters</artifactId>
<groupId>org.apache.tika</groupId>
+ <artifactId>tika-fetch-iterators</artifactId>
<version>2.0.0-SNAPSHOT</version>
+ <relativePath>../pom.xml</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
- <artifactId>tika-emitter-solr</artifactId>
-
+ <artifactId>tika-fetch-iterator-csv</artifactId>
+
+ <name>Apache Tika Fetch Iterator - CSV</name>
+ <url>http://tika.apache.org/</url>
+
<dependencies>
<dependency>
<groupId>${project.groupId}</groupId>
@@ -37,13 +41,9 @@
<scope>provided</scope>
</dependency>
<dependency>
- <groupId>${project.groupId}</groupId>
- <artifactId>tika-httpclient-commons</artifactId>
- <version>${project.version}</version>
- </dependency>
- <dependency>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-log4j12</artifactId>
+ <groupId>org.apache.commons</groupId>
+ <artifactId>commons-csv</artifactId>
+ <version>${commons.csv.version}</version>
</dependency>
<dependency>
<groupId>junit</groupId>
@@ -51,5 +51,4 @@
<scope>test</scope>
</dependency>
</dependencies>
-
-</project>
\ No newline at end of file
+</project>
diff --git a/tika-pipes/tika-fetch-iterators/tika-fetch-iterator-csv/src/main/java/org/apache/tika/pipes/fetchiterator/csv/CSVFetchIterator.java b/tika-pipes/tika-fetch-iterators/tika-fetch-iterator-csv/src/main/java/org/apache/tika/pipes/fetchiterator/csv/CSVFetchIterator.java
new file mode 100644
index 0000000..e87f730
--- /dev/null
+++ b/tika-pipes/tika-fetch-iterators/tika-fetch-iterator-csv/src/main/java/org/apache/tika/pipes/fetchiterator/csv/CSVFetchIterator.java
@@ -0,0 +1,121 @@
+/*
+ * 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.fetchiterator.csv;
+
+import org.apache.commons.csv.CSVFormat;
+import org.apache.commons.csv.CSVRecord;
+import org.apache.tika.config.Field;
+import org.apache.tika.config.Initializable;
+import org.apache.tika.config.InitializableProblemHandler;
+import org.apache.tika.exception.TikaConfigException;
+import org.apache.tika.metadata.Metadata;
+import org.apache.tika.pipes.fetcher.FetchId;
+import org.apache.tika.pipes.fetcher.FetchIdMetadataPair;
+import org.apache.tika.pipes.fetchiterator.FetchIterator;
+
+import java.io.IOException;
+import java.io.Reader;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeoutException;
+
+public class CSVFetchIterator extends FetchIterator implements Initializable {
+
+ private Charset charset = StandardCharsets.UTF_8;
+ private Path csvPath;
+ private String fetchKeyColumn;
+
+
+ @Field
+ public void setCsvPath(String csvPath) {
+ setCsvPath(Paths.get(csvPath));
+ }
+
+ @Field
+ public void setFetchKeyColumn(String fetchKeyColumn) {
+ this.fetchKeyColumn = fetchKeyColumn;
+ }
+
+ public void setCsvPath(Path csvPath) {
+ this.csvPath = csvPath;
+ }
+
+ @Override
+ protected void enqueue() throws InterruptedException, IOException, TimeoutException {
+ String fetcherName = getFetcherName();
+ try (Reader reader = Files.newBufferedReader(csvPath, charset)) {
+ Iterable<CSVRecord> records = CSVFormat.EXCEL.parse(reader);
+ int fetchKeyIndex = -1;
+ List<String> headers = new ArrayList<>();
+ for (CSVRecord record : records) {
+ fetchKeyIndex = loadHeaders(record, headers);
+ break;
+ }
+ //check that if a user set a fetchKeyColumn, but we didn't find it
+ if (fetchKeyIndex < 0 && (fetchKeyColumn != null)) {
+ throw new IllegalArgumentException("Can't find " + fetchKeyColumn +
+ " in the csv. I see:"+
+ headers);
+ }
+ for (CSVRecord record : records) {
+ String fetchKey = "";
+ if (fetchKeyIndex > -1) {
+ fetchKey = record.get(fetchKeyIndex);
+ }
+ Metadata metadata = loadMetadata(fetchKeyIndex, headers, record);
+ tryToAdd(new FetchIdMetadataPair(new FetchId(fetcherName, fetchKey), metadata));
+ }
+ }
+ }
+
+ private Metadata loadMetadata(int fetchKeyIndex, List<String> headers, CSVRecord record) {
+ Metadata metadata = new Metadata();
+ for (int i = 0; i < record.size(); i++) {
+ if (fetchKeyIndex == i) {
+ continue;
+ }
+ metadata.set(headers.get(i), record.get(i));
+ }
+ return metadata;
+ }
+
+
+ private int loadHeaders(CSVRecord record, List<String> headers) {
+ int fetchKeyColumnIndex = -1;
+
+ for (int col = 0; col < record.size(); col++) {
+ String header = record.get(col);
+ headers.add(header);
+ if (header.equals(fetchKeyColumn)) {
+ fetchKeyColumnIndex = col;
+ }
+ }
+ return fetchKeyColumnIndex;
+ }
+
+ @Override
+ public void checkInitialization(InitializableProblemHandler problemHandler)
+ throws TikaConfigException {
+ super.checkInitialization(problemHandler);
+ mustNotBeEmpty("csvPath", this.csvPath.toString());
+ }
+}
diff --git a/tika-pipes/tika-fetch-iterators/tika-fetch-iterator-csv/src/test/java/TestCSVFetchIterator.java b/tika-pipes/tika-fetch-iterators/tika-fetch-iterator-csv/src/test/java/TestCSVFetchIterator.java
new file mode 100644
index 0000000..2a6ab52
--- /dev/null
+++ b/tika-pipes/tika-fetch-iterators/tika-fetch-iterator-csv/src/test/java/TestCSVFetchIterator.java
@@ -0,0 +1,129 @@
+/*
+ * 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.
+ */
+import org.apache.tika.metadata.Metadata;
+import org.apache.tika.pipes.fetcher.FetchId;
+import org.apache.tika.pipes.fetcher.FetchIdMetadataPair;
+import org.apache.tika.pipes.fetchiterator.FetchIterator;
+import org.apache.tika.pipes.fetchiterator.csv.CSVFetchIterator;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorCompletionService;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertEquals;
+
+public class TestCSVFetchIterator {
+
+
+ @Test
+ public void testSimple() throws Exception {
+ Path p = get("test-simple.csv");
+ CSVFetchIterator it = new CSVFetchIterator();
+ it.setFetcherName("fs");
+ it.setCsvPath(p);
+ it.setFetchKeyColumn("fetchKey");
+ int numConsumers = 2;
+ ExecutorService es = Executors.newFixedThreadPool(numConsumers+1);
+ ArrayBlockingQueue<FetchIdMetadataPair> queue = it.init(numConsumers);
+ ExecutorCompletionService c = new ExecutorCompletionService(es);
+ c.submit(it);
+ List<MockFetcher> fetchers = new ArrayList<>();
+ for (int i = 0; i < numConsumers; i++) {
+ MockFetcher f = new MockFetcher(queue);
+ fetchers.add(f);
+ c.submit(f);
+ }
+ int finished = 0;
+ int completed = 0;
+ try {
+ while (finished++ < numConsumers+1) {
+ Future<Integer> f = c.take();
+ completed += f.get();
+ }
+ } finally {
+ es.shutdownNow();
+ }
+ assertEquals(10, completed);
+ for (MockFetcher f : fetchers) {
+ for (FetchIdMetadataPair pair : f.pairs) {
+ String id = pair.getMetadata().get("id");
+ assertEquals("path/to/my/file"+id,
+ pair.getFetchId().getFetchKey());
+ assertEquals("project"+
+ (Integer.parseInt(id) % 2 == 1 ? "a" : "b"),
+ pair.getMetadata().get("project"));
+ }
+ }
+ }
+
+ @Test(expected = ExecutionException.class)
+ public void testBadFetchKeyCol() throws Exception {
+ Path p = get("test-simple.csv");
+ CSVFetchIterator it = new CSVFetchIterator();
+ it.setFetcherName("fs");
+ it.setCsvPath(p);
+ it.setFetchKeyColumn("fetchKeyDoesntExist");
+ ExecutorService es = Executors.newFixedThreadPool(2);
+ ExecutorCompletionService c = new ExecutorCompletionService(es);
+ c.submit(it);
+ c.submit(new MockFetcher(it.init(1)));
+ int finished = 0;
+ try {
+ while (finished++ < 2) {
+ Future f = c.take();
+ f.get();
+ }
+ } finally {
+ es.shutdownNow();
+ }
+
+ }
+
+ private Path get(String testFileName) throws Exception {
+ return Paths.get(TestCSVFetchIterator.class.getResource("/"+testFileName).toURI());
+ }
+
+ private static class MockFetcher implements Callable<Integer> {
+ private final ArrayBlockingQueue<FetchIdMetadataPair> queue;
+ private final List<FetchIdMetadataPair> pairs = new ArrayList<>();
+ private MockFetcher(ArrayBlockingQueue<FetchIdMetadataPair> queue) {
+ this.queue = queue;
+ }
+
+ @Override
+ public Integer call() throws Exception {
+ while (true) {
+ FetchIdMetadataPair p = queue.poll(1, TimeUnit.HOURS);
+ if (p == FetchIterator.COMPLETED_SEMAPHORE) {
+ return pairs.size();
+ }
+ pairs.add(p);
+ }
+ }
+ }
+}
diff --git a/tika-pipes/tika-fetch-iterators/tika-fetch-iterator-csv/src/test/resources/test-simple.csv b/tika-pipes/tika-fetch-iterators/tika-fetch-iterator-csv/src/test/resources/test-simple.csv
new file mode 100644
index 0000000..22630ef
--- /dev/null
+++ b/tika-pipes/tika-fetch-iterators/tika-fetch-iterator-csv/src/test/resources/test-simple.csv
@@ -0,0 +1,6 @@
+id,project,fetchKey
+1,projecta,path/to/my/file1
+2,projectb,path/to/my/file2
+3,projecta,path/to/my/file3
+4,projectb,path/to/my/file4
+5,projecta,path/to/my/file5
diff --git a/tika-emitters/tika-emitter-solr/pom.xml b/tika-pipes/tika-fetch-iterators/tika-fetch-iterator-jdbc/pom.xml
similarity index 79%
copy from tika-emitters/tika-emitter-solr/pom.xml
copy to tika-pipes/tika-fetch-iterators/tika-fetch-iterator-jdbc/pom.xml
index 6aefb89..14cd6f3 100644
--- a/tika-emitters/tika-emitter-solr/pom.xml
+++ b/tika-pipes/tika-fetch-iterators/tika-fetch-iterator-jdbc/pom.xml
@@ -21,14 +21,18 @@
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
- <artifactId>tika-emitters</artifactId>
<groupId>org.apache.tika</groupId>
+ <artifactId>tika-fetch-iterators</artifactId>
<version>2.0.0-SNAPSHOT</version>
+ <relativePath>../pom.xml</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
- <artifactId>tika-emitter-solr</artifactId>
-
+ <artifactId>tika-fetch-iterator-jdbc</artifactId>
+
+ <name>Apache Tika Fetch Iterator - jdbc</name>
+ <url>http://tika.apache.org/</url>
+
<dependencies>
<dependency>
<groupId>${project.groupId}</groupId>
@@ -37,13 +41,10 @@
<scope>provided</scope>
</dependency>
<dependency>
- <groupId>${project.groupId}</groupId>
- <artifactId>tika-httpclient-commons</artifactId>
- <version>${project.version}</version>
- </dependency>
- <dependency>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-log4j12</artifactId>
+ <groupId>com.h2database</groupId>
+ <artifactId>h2</artifactId>
+ <version>${h2.version}</version>
+ <scope>test</scope>
</dependency>
<dependency>
<groupId>junit</groupId>
@@ -51,5 +52,4 @@
<scope>test</scope>
</dependency>
</dependencies>
-
-</project>
\ No newline at end of file
+</project>
diff --git a/tika-pipes/tika-fetch-iterators/tika-fetch-iterator-jdbc/src/main/java/org/apache/tika/pipes/fetchiterator/jdbc/JDBCFetchIterator.java b/tika-pipes/tika-fetch-iterators/tika-fetch-iterator-jdbc/src/main/java/org/apache/tika/pipes/fetchiterator/jdbc/JDBCFetchIterator.java
new file mode 100644
index 0000000..9908681
--- /dev/null
+++ b/tika-pipes/tika-fetch-iterators/tika-fetch-iterator-jdbc/src/main/java/org/apache/tika/pipes/fetchiterator/jdbc/JDBCFetchIterator.java
@@ -0,0 +1,166 @@
+/*
+ * 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.fetchiterator.jdbc;
+
+import org.apache.tika.config.Field;
+import org.apache.tika.config.Initializable;
+import org.apache.tika.config.InitializableProblemHandler;
+import org.apache.tika.config.Param;
+import org.apache.tika.exception.TikaConfigException;
+import org.apache.tika.metadata.Metadata;
+import org.apache.tika.pipes.fetcher.FetchId;
+import org.apache.tika.pipes.fetcher.FetchIdMetadataPair;
+import org.apache.tika.pipes.fetchiterator.FetchIterator;
+import org.apache.tika.utils.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeoutException;
+
+public class JDBCFetchIterator extends FetchIterator implements Initializable {
+
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(JDBCFetchIterator.class);
+
+ private String fetchKeyColumn;
+ private String connection;
+ private String select;
+
+ private Connection db;
+
+ @Field
+ public void setFetchKeyColumn(String fetchKeyColumn) {
+ this.fetchKeyColumn = fetchKeyColumn;
+ }
+
+ @Field
+ public void setConnection(String connection) {
+ this.connection = connection;
+ }
+
+ @Field
+ public void setSelect(String select) {
+ this.select = select;
+ }
+
+ @Override
+ protected void enqueue() throws InterruptedException, IOException, TimeoutException {
+ String fetcherName = getFetcherName();
+ int fetchKeyIndex = -1;
+ List<String> headers = new ArrayList<>();
+ int rowCount = 0;
+ LOGGER.debug("select: {}", select);
+ try (Statement st = db.createStatement()) {
+ try (ResultSet rs = st.executeQuery(select)) {
+ while (rs.next()) {
+ if (headers.size() == 0) {
+ fetchKeyIndex = loadHeaders(rs.getMetaData(), headers);
+ }
+ try {
+ processRow(fetcherName, headers, fetchKeyIndex, rs);
+ } catch (SQLException e) {
+ LOGGER.warn("Failed to insert: "+rs, e);
+ }
+ rowCount++;
+ if (rowCount % 1000 == 0) {
+ LOGGER.info("added "+rowCount + " rows to the queue");
+ }
+ }
+ }
+ } catch (SQLException e) {
+ LOGGER.warn("problem initializing connection and selecting", e);
+ throw new IOException(e);
+ } finally {
+ try {
+ db.close();
+ } catch (SQLException e){
+ LOGGER.warn("failed to close connection", e);
+ }
+ }
+ }
+
+ private void processRow(String fetcherName, List<String> headers,
+ int fetchKeyIndex, ResultSet rs)
+ throws SQLException, TimeoutException, InterruptedException {
+ Metadata metadata = new Metadata();
+ String fetchKey = null;
+ for (int i = 1; i <= rs.getMetaData().getColumnCount(); i++) {
+ if (i == fetchKeyIndex) {
+ fetchKey = getString(i, rs);
+ fetchKey = (fetchKey == null) ? "" : fetchKey;
+ continue;
+ }
+ String val = getString(i, rs);
+ if (val != null) {
+ metadata.set(headers.get(i - 1), val);
+ }
+ }
+
+ tryToAdd(new FetchIdMetadataPair(new FetchId(fetcherName, fetchKey), metadata));
+ }
+
+ private String getString(int i, ResultSet rs) throws SQLException {
+ //TODO: improve this later with special handling for numerals/dates/timestamps, etc
+ String val = rs.getString(i);
+ if (rs.wasNull()) {
+ return null;
+ }
+ return val;
+ }
+
+
+ private int loadHeaders(ResultSetMetaData metaData, List<String> headers) throws SQLException {
+ int fetchKeyIndex = -1;
+ for (int i = 1; i <= metaData.getColumnCount(); i++) {
+ if (metaData.getColumnLabel(i).equalsIgnoreCase(fetchKeyColumn)) {
+ fetchKeyIndex = i;
+ }
+ headers.add(metaData.getColumnLabel(i));
+ }
+ return fetchKeyIndex;
+ }
+
+ @Override
+ public void initialize(Map<String, Param> params) throws TikaConfigException {
+ try {
+ db = DriverManager.getConnection(connection);
+ } catch (SQLException e) {
+ throw new TikaConfigException("couldn't connect to db", e);
+ }
+ }
+
+ @Override
+ public void checkInitialization(InitializableProblemHandler problemHandler)
+ throws TikaConfigException {
+ super.checkInitialization(problemHandler);
+ mustNotBeEmpty("connection", this.connection);
+ mustNotBeEmpty("select", this.select);
+ if (StringUtils.isEmpty(fetchKeyColumn)) {
+ LOGGER.info("no fetch key column has been specified");
+ }
+ }
+}
diff --git a/tika-pipes/tika-fetch-iterators/tika-fetch-iterator-jdbc/src/test/java/org/apache/tika/pipes/fetchiterator/jdbc/TestJDBCFetchIterator.java b/tika-pipes/tika-fetch-iterators/tika-fetch-iterator-jdbc/src/test/java/org/apache/tika/pipes/fetchiterator/jdbc/TestJDBCFetchIterator.java
new file mode 100644
index 0000000..fb0277b
--- /dev/null
+++ b/tika-pipes/tika-fetch-iterators/tika-fetch-iterator-jdbc/src/test/java/org/apache/tika/pipes/fetchiterator/jdbc/TestJDBCFetchIterator.java
@@ -0,0 +1,178 @@
+/*
+ * 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.fetchiterator.jdbc;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.tika.config.TikaConfig;
+import org.apache.tika.metadata.Metadata;
+import org.apache.tika.pipes.fetcher.FetchId;
+import org.apache.tika.pipes.fetcher.FetchIdMetadataPair;
+import org.apache.tika.pipes.fetchiterator.FetchIterator;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorCompletionService;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.fail;
+
+public class TestJDBCFetchIterator {
+
+ static final String TABLE = "fetchkeys";
+ static Connection CONNECTION;
+ static Path DB_DIR;
+ static final String db = "mydb";
+ private static final int NUM_ROWS = 1000;
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ DB_DIR = Files.createTempDirectory("tika-jdbc-fetchiterator-test-");
+
+ CONNECTION = DriverManager.getConnection("jdbc:h2:file:"+DB_DIR.toAbsolutePath()+"/"+db);
+ String sql = "create table "+TABLE +
+ " (id varchar(128), " +
+ "project varchar(128), " +
+ "fetchKey varchar(128))";
+ CONNECTION.createStatement().execute(sql);
+
+ for (int i = 0; i < NUM_ROWS; i++) {
+ sql = "insert into "+TABLE + " (id, project, fetchKey) values ('id"+i+"','project"+
+ (i%2 == 0 ? "a" : "b") +"','fk"+i+"')";
+ CONNECTION.createStatement().execute(sql);
+ }
+ sql = "select count(1) from "+ TABLE;
+ ResultSet rs = CONNECTION.createStatement().executeQuery(sql);
+ while (rs.next()) {
+ int cnt = rs.getInt(1);
+ assertEquals(NUM_ROWS, cnt);
+ }
+ }
+
+ @AfterClass
+ public static void tearDown() throws IOException {
+ FileUtils.deleteDirectory(DB_DIR.toFile());
+ }
+
+ @Test
+ public void testSimple() throws Exception {
+ TikaConfig tk = getConfig();
+ int numConsumers = 5;
+ FetchIterator fetchIterator = tk.getFetchIterator();
+ ExecutorService es = Executors.newFixedThreadPool(numConsumers+1);
+ ExecutorCompletionService<Integer> completionService = new ExecutorCompletionService<>(es);
+ ArrayBlockingQueue<FetchIdMetadataPair> queue = fetchIterator.init(numConsumers);
+ completionService.submit(fetchIterator);
+ List<MockFetcher> fetchers = new ArrayList<>();
+ for (int i = 0; i < numConsumers; i++) {
+ MockFetcher mockFetcher = new MockFetcher(queue);
+ fetchers.add(mockFetcher);
+ completionService.submit(mockFetcher);
+ }
+ int processed = 0;
+ int completed = 0;
+ while (completed < numConsumers+1) {
+ Future<Integer> f = completionService.take();
+ processed += f.get();
+ completed++;
+ }
+ //fetchiterator added + MockFetcher yields 2x
+ assertEquals(NUM_ROWS * 2, processed);
+ int cnt = 0;
+ Matcher m = Pattern.compile("fk(\\d+)").matcher("");
+ for (MockFetcher f : fetchers) {
+ for (FetchIdMetadataPair p : f.pairs) {
+ String k = p.getFetchId().getFetchKey();
+ String num = "";
+ if (m.reset(k).find()) {
+ num = m.group(1);
+ } else {
+ fail("failed to find key pattern: "+k);
+ }
+ String aOrB = Integer.parseInt(num) % 2 == 0 ? "a" : "b";
+ assertEquals("id"+num, p.getMetadata().get("MY_ID"));
+ assertEquals("project"+aOrB, p.getMetadata().get("MY_PROJECT"));
+ assertNull(p.getMetadata().get("fetchKey"));
+ assertNull(p.getMetadata().get("MY_FETCHKEY"));
+ cnt++;
+ }
+ }
+ assertEquals(NUM_ROWS, cnt);
+ }
+
+ private TikaConfig getConfig() throws Exception {
+ String config = "<?xml version=\"1.0\" encoding=\"UTF-8\" ?><properties>\n" +
+ " <fetchIterators>\n" +
+ " <fetchIterator class=\"org.apache.tika.pipes.fetchiterator.jdbc.JDBCFetchIterator\">\n" +
+ " <params>\n" +
+ " <param name=\"fetcherName\" type=\"string\">s3</param>\n" +
+ " <param name=\"queueSize\" type=\"int\">57</param>\n" +
+ " <param name=\"fetchKeyColumn\" type=\"string\">my_fetchkey</param>\n" +
+ " <param name=\"select\" type=\"string\">" +
+ "select id as my_id, project as my_project, fetchKey as my_fetchKey from fetchkeys</param>\n" +
+ " <param name=\"connection\" type=\"string\">jdbc:h2:file:"+
+ DB_DIR.toAbsolutePath()+"/"+db +"</param>\n" +
+ " </params>\n" +
+ " </fetchIterator>\n" +
+ " </fetchIterators>\n" +
+ "</properties>";
+ return new TikaConfig(new ByteArrayInputStream(config.getBytes(StandardCharsets.UTF_8)));
+ }
+
+ private static class MockFetcher implements Callable<Integer> {
+ private final ArrayBlockingQueue<FetchIdMetadataPair> queue;
+ private final List<FetchIdMetadataPair> pairs = new ArrayList<>();
+ private MockFetcher(ArrayBlockingQueue<FetchIdMetadataPair> queue) {
+ this.queue = queue;
+ }
+
+ @Override
+ public Integer call() throws Exception {
+ while (true) {
+ FetchIdMetadataPair p = queue.poll(1, TimeUnit.HOURS);
+ if (p == FetchIterator.COMPLETED_SEMAPHORE) {
+ return pairs.size();
+ }
+ pairs.add(p);
+ }
+ }
+ }
+}
diff --git a/tika-emitters/tika-emitter-solr/src/test/resources/log4j.properties b/tika-pipes/tika-fetch-iterators/tika-fetch-iterator-jdbc/src/test/resources/log4j.properties
similarity index 91%
copy from tika-emitters/tika-emitter-solr/src/test/resources/log4j.properties
copy to tika-pipes/tika-fetch-iterators/tika-fetch-iterator-jdbc/src/test/resources/log4j.properties
index 92b6d56..1bee240 100644
--- a/tika-emitters/tika-emitter-solr/src/test/resources/log4j.properties
+++ b/tika-pipes/tika-fetch-iterators/tika-fetch-iterator-jdbc/src/test/resources/log4j.properties
@@ -14,11 +14,11 @@
# limitations under the License.
#info,debug, error,fatal ...
-log4j.rootLogger=debug,stderr
+log4j.rootLogger=info,stderr
#console
log4j.appender.stderr=org.apache.log4j.ConsoleAppender
log4j.appender.stderr.layout=org.apache.log4j.PatternLayout
log4j.appender.stderr.Target=System.err
-log4j.appender.stderr.layout.ConversionPattern= %-5p %m%n
+log4j.appender.stderr.layout.ConversionPattern=%-5p [%t]: %m%n
diff --git a/tika-emitters/tika-emitter-solr/pom.xml b/tika-pipes/tika-fetch-iterators/tika-fetch-iterator-s3/pom.xml
similarity index 53%
copy from tika-emitters/tika-emitter-solr/pom.xml
copy to tika-pipes/tika-fetch-iterators/tika-fetch-iterator-s3/pom.xml
index 6aefb89..b0cce91 100644
--- a/tika-emitters/tika-emitter-solr/pom.xml
+++ b/tika-pipes/tika-fetch-iterators/tika-fetch-iterator-s3/pom.xml
@@ -21,13 +21,17 @@
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
- <artifactId>tika-emitters</artifactId>
<groupId>org.apache.tika</groupId>
+ <artifactId>tika-fetch-iterators</artifactId>
<version>2.0.0-SNAPSHOT</version>
+ <relativePath>../pom.xml</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
- <artifactId>tika-emitter-solr</artifactId>
+ <artifactId>tika-fetch-iterator-s3</artifactId>
+
+ <name>Apache Tika Fetch Iterator - S3</name>
+ <url>http://tika.apache.org/</url>
<dependencies>
<dependency>
@@ -37,19 +41,54 @@
<scope>provided</scope>
</dependency>
<dependency>
- <groupId>${project.groupId}</groupId>
- <artifactId>tika-httpclient-commons</artifactId>
- <version>${project.version}</version>
+ <groupId>com.amazonaws</groupId>
+ <artifactId>aws-java-sdk-s3</artifactId>
+ <version>${aws.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>commons-logging</groupId>
+ <artifactId>commons-logging</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-core</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ <version>${jackson.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>commons-logging</groupId>
+ <artifactId>commons-logging</artifactId>
+ <version>${commons.logging.version}</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
</dependency>
<dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>tika-core</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
-
-</project>
\ No newline at end of file
+</project>
diff --git a/tika-pipes/tika-fetch-iterators/tika-fetch-iterator-s3/src/main/java/org/apache/tika/pipes/fetchiterator/s3/S3FetchIterator.java b/tika-pipes/tika-fetch-iterators/tika-fetch-iterator-s3/src/main/java/org/apache/tika/pipes/fetchiterator/s3/S3FetchIterator.java
new file mode 100644
index 0000000..0fd88b6
--- /dev/null
+++ b/tika-pipes/tika-fetch-iterators/tika-fetch-iterator-s3/src/main/java/org/apache/tika/pipes/fetchiterator/s3/S3FetchIterator.java
@@ -0,0 +1,109 @@
+/*
+ * 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.fetchiterator.s3;
+
+import com.amazonaws.auth.profile.ProfileCredentialsProvider;
+import com.amazonaws.services.s3.AmazonS3;
+import com.amazonaws.services.s3.AmazonS3ClientBuilder;
+import com.amazonaws.services.s3.iterable.S3Objects;
+import com.amazonaws.services.s3.model.ObjectListing;
+import com.amazonaws.services.s3.model.S3ObjectSummary;
+import org.apache.tika.config.Field;
+import org.apache.tika.config.Initializable;
+import org.apache.tika.config.InitializableProblemHandler;
+import org.apache.tika.config.Param;
+import org.apache.tika.exception.TikaConfigException;
+import org.apache.tika.metadata.Metadata;
+import org.apache.tika.pipes.fetcher.FetchId;
+import org.apache.tika.pipes.fetcher.FetchIdMetadataPair;
+import org.apache.tika.pipes.fetchiterator.FetchIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.TimeoutException;
+
+public class S3FetchIterator extends FetchIterator implements Initializable {
+
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(S3FetchIterator.class);
+ private String s3PathPrefix = "";
+ private String region;
+ private String profile;
+ private String bucket;
+
+ private AmazonS3 s3Client;
+
+ @Field
+ public void setBucket(String bucket) {
+ this.bucket = bucket;
+ }
+
+ @Field
+ public void setRegion(String region) {
+ this.region = region;
+ }
+
+ @Field
+ public void setProfile(String profile) {
+ this.profile = profile;
+ }
+
+ @Field
+ public void setS3PathPrefix(String s3PathPrefix) {
+ this.s3PathPrefix = s3PathPrefix;
+ }
+
+ @Override
+ public void initialize(Map<String, Param> params) throws TikaConfigException {
+ //params have already been set
+ //ignore them
+
+ s3Client = AmazonS3ClientBuilder.standard()
+ .withRegion(region)
+ .withCredentials(new ProfileCredentialsProvider(profile))
+ .build();
+ }
+
+ @Override
+ public void checkInitialization(InitializableProblemHandler problemHandler)
+ throws TikaConfigException {
+ super.checkInitialization(problemHandler);
+ mustNotBeEmpty("bucket", this.bucket);
+ mustNotBeEmpty("profile", this.profile);
+ mustNotBeEmpty("region", this.region);
+ }
+
+ @Override
+ protected void enqueue() throws InterruptedException, IOException, TimeoutException {
+ String fetcherName = getFetcherName();
+ long start = System.currentTimeMillis();
+ int count = 0;
+ for (S3ObjectSummary summary : S3Objects.withPrefix(s3Client, bucket, s3PathPrefix)) {
+ long elapsed = System.currentTimeMillis() - start;
+ LOGGER.debug("adding ({}) {} in {} ms", count, bucket+"/"+summary.getKey(),
+ elapsed);
+ tryToAdd( new FetchIdMetadataPair(
+ new FetchId(fetcherName, bucket+"/"+summary.getKey()),
+ new Metadata()));
+ count++;
+ }
+ long elapsed = System.currentTimeMillis() - start;
+ LOGGER.info("finished enqueuing {} files in {} ms", count, elapsed);
+ }
+}
diff --git a/tika-pipes/tika-fetch-iterators/tika-fetch-iterator-s3/src/test/java/org/apache/tika/pipes/fetchiterator/s3/TestS3FetchIterator.java b/tika-pipes/tika-fetch-iterators/tika-fetch-iterator-s3/src/test/java/org/apache/tika/pipes/fetchiterator/s3/TestS3FetchIterator.java
new file mode 100644
index 0000000..1984577
--- /dev/null
+++ b/tika-pipes/tika-fetch-iterators/tika-fetch-iterator-s3/src/test/java/org/apache/tika/pipes/fetchiterator/s3/TestS3FetchIterator.java
@@ -0,0 +1,98 @@
+/*
+ * 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.fetchiterator.s3;
+import org.apache.tika.metadata.Metadata;
+import org.apache.tika.pipes.fetcher.FetchId;
+import org.apache.tika.pipes.fetcher.FetchIdMetadataPair;
+import org.apache.tika.pipes.fetchiterator.FetchIterator;
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Test;
+
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorCompletionService;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertEquals;
+
+@Ignore("turn into an actual unit test")
+public class TestS3FetchIterator {
+
+
+ @Test
+ public void testSimple() throws Exception {
+ S3FetchIterator it = new S3FetchIterator();
+ it.setFetcherName("s3");
+ it.setBucket("");//find one
+ it.setProfile("");//use one
+ it.setRegion("");//select one
+ it.initialize(Collections.EMPTY_MAP);
+ int numConsumers = 6;
+ ArrayBlockingQueue<FetchIdMetadataPair> queue = it.init(numConsumers);
+
+ ExecutorService es = Executors.newFixedThreadPool(numConsumers+1);
+ ExecutorCompletionService c = new ExecutorCompletionService(es);
+ c.submit(it);
+ List<MockFetcher> fetchers = new ArrayList<>();
+ for (int i = 0; i < numConsumers; i++) {
+ MockFetcher fetcher = new MockFetcher(queue);
+ fetchers.add(fetcher);
+ c.submit(fetcher);
+ }
+ int finished = 0;
+ int completed = 0;
+ try {
+ while (finished++ < numConsumers+1) {
+ Future<Integer> f = c.take();
+ completed += f.get();
+ }
+ } finally {
+ es.shutdownNow();
+ }
+ assertEquals(20000, completed);
+
+ }
+
+ private static class MockFetcher implements Callable<Integer> {
+ private final ArrayBlockingQueue<FetchIdMetadataPair> queue;
+ private final List<FetchIdMetadataPair> pairs = new ArrayList<>();
+ private MockFetcher(ArrayBlockingQueue<FetchIdMetadataPair> queue) {
+ this.queue = queue;
+ }
+
+ @Override
+ public Integer call() throws Exception {
+ while (true) {
+ FetchIdMetadataPair p = queue.poll(1, TimeUnit.HOURS);
+ if (p == FetchIterator.COMPLETED_SEMAPHORE) {
+ return pairs.size();
+ }
+ pairs.add(p);
+ }
+ }
+ }
+}
diff --git a/tika-emitters/tika-emitter-solr/src/test/resources/log4j.properties b/tika-pipes/tika-fetch-iterators/tika-fetch-iterator-s3/src/test/resources/log4j.properties
similarity index 91%
copy from tika-emitters/tika-emitter-solr/src/test/resources/log4j.properties
copy to tika-pipes/tika-fetch-iterators/tika-fetch-iterator-s3/src/test/resources/log4j.properties
index 92b6d56..1bee240 100644
--- a/tika-emitters/tika-emitter-solr/src/test/resources/log4j.properties
+++ b/tika-pipes/tika-fetch-iterators/tika-fetch-iterator-s3/src/test/resources/log4j.properties
@@ -14,11 +14,11 @@
# limitations under the License.
#info,debug, error,fatal ...
-log4j.rootLogger=debug,stderr
+log4j.rootLogger=info,stderr
#console
log4j.appender.stderr=org.apache.log4j.ConsoleAppender
log4j.appender.stderr.layout=org.apache.log4j.PatternLayout
log4j.appender.stderr.Target=System.err
-log4j.appender.stderr.layout.ConversionPattern= %-5p %m%n
+log4j.appender.stderr.layout.ConversionPattern=%-5p [%t]: %m%n
diff --git a/tika-fetchers/pom.xml b/tika-pipes/tika-fetchers/pom.xml
similarity index 90%
rename from tika-fetchers/pom.xml
rename to tika-pipes/tika-fetchers/pom.xml
index 833a56f..597c76b 100644
--- a/tika-fetchers/pom.xml
+++ b/tika-pipes/tika-fetchers/pom.xml
@@ -22,9 +22,9 @@
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<groupId>org.apache.tika</groupId>
- <artifactId>tika-parent</artifactId>
+ <artifactId>tika-pipes</artifactId>
<version>2.0.0-SNAPSHOT</version>
- <relativePath>../tika-parent/pom.xml</relativePath>
+ <relativePath>../pom.xml</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
@@ -34,7 +34,6 @@
<url>http://tika.apache.org/</url>
<modules>
- <module>tika-fetcher-jdbc</module>
<module>tika-fetcher-s3</module>
</modules>
diff --git a/tika-fetchers/tika-fetcher-s3/pom.xml b/tika-pipes/tika-fetchers/tika-fetcher-s3/pom.xml
similarity index 97%
rename from tika-fetchers/tika-fetcher-s3/pom.xml
rename to tika-pipes/tika-fetchers/tika-fetcher-s3/pom.xml
index 2ac83e1..1d561aa 100644
--- a/tika-fetchers/tika-fetcher-s3/pom.xml
+++ b/tika-pipes/tika-fetchers/tika-fetcher-s3/pom.xml
@@ -27,7 +27,7 @@
</parent>
<modelVersion>4.0.0</modelVersion>
- <artifactId>s3-fetcher</artifactId>
+ <artifactId>tika-fetcher-s3</artifactId>
<dependencies>
<dependency>
@@ -61,6 +61,10 @@
<version>${commons.logging.version}</version>
</dependency>
<dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ </dependency>
+ <dependency>
<groupId>${project.groupId}</groupId>
<artifactId>tika-core</artifactId>
<version>${project.version}</version>
diff --git a/tika-fetchers/tika-fetcher-s3/src/main/java/org/apache/tika/fetcher/s3/S3Fetcher.java b/tika-pipes/tika-fetchers/tika-fetcher-s3/src/main/java/org/apache/tika/pipes/fetcher/s3/S3Fetcher.java
similarity index 63%
rename from tika-fetchers/tika-fetcher-s3/src/main/java/org/apache/tika/fetcher/s3/S3Fetcher.java
rename to tika-pipes/tika-fetchers/tika-fetcher-s3/src/main/java/org/apache/tika/pipes/fetcher/s3/S3Fetcher.java
index 2757f17..126f8fa 100644
--- a/tika-fetchers/tika-fetcher-s3/src/main/java/org/apache/tika/fetcher/s3/S3Fetcher.java
+++ b/tika-pipes/tika-fetchers/tika-fetcher-s3/src/main/java/org/apache/tika/pipes/fetcher/s3/S3Fetcher.java
@@ -14,27 +14,35 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.tika.fetcher.s3;
+package org.apache.tika.pipes.fetcher.s3;
import com.amazonaws.auth.profile.ProfileCredentialsProvider;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.AmazonS3ClientBuilder;
import com.amazonaws.services.s3.model.GetObjectRequest;
import com.amazonaws.services.s3.model.S3Object;
+import org.apache.commons.io.IOUtils;
import org.apache.tika.config.Field;
import org.apache.tika.config.Initializable;
import org.apache.tika.config.InitializableProblemHandler;
import org.apache.tika.config.Param;
import org.apache.tika.exception.TikaConfigException;
import org.apache.tika.exception.TikaException;
-import org.apache.tika.fetcher.FetchPrefixKeyPair;
-import org.apache.tika.fetcher.Fetcher;
-import org.apache.tika.fetcher.FetcherStringException;
+import org.apache.tika.io.TemporaryResources;
+import org.apache.tika.pipes.fetcher.AbstractFetcher;
+import org.apache.tika.pipes.fetcher.FetchId;
+import org.apache.tika.pipes.fetcher.Fetcher;
+import org.apache.tika.pipes.fetcher.FetcherStringException;
import org.apache.tika.io.TikaInputStream;
import org.apache.tika.metadata.Metadata;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.io.InputStream;
+import java.nio.file.CopyOption;
+import java.nio.file.Path;
+import java.nio.file.StandardCopyOption;
import java.util.Collections;
import java.util.Map;
import java.util.Set;
@@ -43,39 +51,35 @@ import java.util.Set;
* Fetches files from s3. Example string: s3://my_bucket/path/to/my_file.pdf
* This will parse the bucket out of that string and retrieve the path.
*/
-public class S3Fetcher implements Fetcher, Initializable {
+public class S3Fetcher extends AbstractFetcher implements Initializable {
+ private static final Logger LOGGER = LoggerFactory.getLogger(S3Fetcher.class);
private static final String PREFIX = "s3";
- private static final Set<String> SUPPORTED = Collections.singleton(PREFIX);
-
private String region;
private String profile;
private boolean extractUserMetadata = true;
private AmazonS3 s3Client;
+ private boolean spoolToTemp = true;
@Override
- public Set<String> getSupportedPrefixes() {
- return SUPPORTED;
- }
-
- @Override
- public InputStream fetch(String fetcherString, Metadata metadata)
+ public InputStream fetch(String fetchKey, Metadata metadata)
throws TikaException, IOException {
- FetchPrefixKeyPair fetchPrefixKeyPair = FetchPrefixKeyPair.create(fetcherString);
- String bucketKey = fetchPrefixKeyPair.getKey();
- if (bucketKey.startsWith("//")) {
- bucketKey = bucketKey.substring(2);
- } else if (bucketKey.startsWith("/")) {
- bucketKey = bucketKey.substring(1);
+
+ LOGGER.debug("about to fetch fetchkey={}", fetchKey);
+ final String origFetchKey = fetchKey;
+ if (fetchKey.startsWith("//")) {
+ fetchKey = fetchKey.substring(2);
+ } else if (fetchKey.startsWith("/")) {
+ fetchKey = fetchKey.substring(1);
}
- int i = bucketKey.indexOf("/");
+ int i = fetchKey.indexOf("/");
if (i < 0) {
- throw new FetcherStringException("Couldn't find bucket:" + fetchPrefixKeyPair.getKey());
+ throw new FetcherStringException("Couldn't find bucket:" +
+ origFetchKey);
}
- String bucket = bucketKey.substring(0, i);
- String key = bucketKey.substring(i + 1);
- //should we cache this to a local file so that
- //we can close the s3Object?
+ String bucket = fetchKey.substring(0, i);
+ String key = fetchKey.substring(i + 1);
+ LOGGER.debug("about to fetch bucket: ({}); key: ({})", bucket, key);
S3Object fullObject = s3Client.getObject(new GetObjectRequest(bucket, key));
if (extractUserMetadata) {
for (Map.Entry<String, String> e :
@@ -83,8 +87,23 @@ public class S3Fetcher implements Fetcher, Initializable {
metadata.add(PREFIX + ":" + e.getKey(), e.getValue());
}
}
- return TikaInputStream.get(
- fullObject.getObjectContent());
+ if (! spoolToTemp) {
+ return TikaInputStream.get(
+ fullObject.getObjectContent());
+ } else {
+ long start = System.currentTimeMillis();
+ TikaInputStream tis = TikaInputStream.get(
+ fullObject.getObjectContent());
+ tis.getPath();
+ long elapsed = System.currentTimeMillis()-start;
+ LOGGER.debug("took {} ms to copy to local tmp file", elapsed);
+ return tis;
+ }
+ }
+
+ @Field
+ public void setSpoolToTemp(boolean spoolToTemp) {
+ this.spoolToTemp = spoolToTemp;
}
@Field
diff --git a/tika-fetchers/tika-fetcher-s3/src/test/java/org/apache/tika/fetcher/s3/TestS3Fetcher.java b/tika-pipes/tika-fetchers/tika-fetcher-s3/src/test/java/org/apache/tika/pipes/fetcher/s3/TestS3Fetcher.java
similarity index 93%
rename from tika-fetchers/tika-fetcher-s3/src/test/java/org/apache/tika/fetcher/s3/TestS3Fetcher.java
rename to tika-pipes/tika-fetchers/tika-fetcher-s3/src/test/java/org/apache/tika/pipes/fetcher/s3/TestS3Fetcher.java
index 1e352f9..450d933 100644
--- a/tika-fetchers/tika-fetcher-s3/src/test/java/org/apache/tika/fetcher/s3/TestS3Fetcher.java
+++ b/tika-pipes/tika-fetchers/tika-fetcher-s3/src/test/java/org/apache/tika/pipes/fetcher/s3/TestS3Fetcher.java
@@ -14,10 +14,9 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.tika.fetcher.s3;
-import com.amazonaws.regions.Regions;
+package org.apache.tika.pipes.fetcher.s3;
import org.apache.tika.config.TikaConfig;
-import org.apache.tika.fetcher.Fetcher;
+import org.apache.tika.pipes.fetcher.Fetcher;
import org.apache.tika.metadata.Metadata;
import org.junit.Ignore;
import org.junit.Test;
@@ -54,7 +53,7 @@ public class TestS3Fetcher {
TikaConfig config = new TikaConfig(
this.getClass().getResourceAsStream("/tika-config-s3.xml")
);
- Fetcher fetcher = config.getFetcher();
+ Fetcher fetcher = config.getFetcherManager().getFetcher("s3");
Metadata metadata = new Metadata();
try (InputStream is = fetcher.fetch(FETCH_STRING, metadata)) {
Files.copy(is, outputFile, StandardCopyOption.REPLACE_EXISTING);
diff --git a/tika-fetchers/tika-fetcher-s3/src/test/resources/tika-config-s3.xml b/tika-pipes/tika-fetchers/tika-fetcher-s3/src/test/resources/tika-config-s3.xml
similarity index 93%
rename from tika-fetchers/tika-fetcher-s3/src/test/resources/tika-config-s3.xml
rename to tika-pipes/tika-fetchers/tika-fetcher-s3/src/test/resources/tika-config-s3.xml
index 39a2ce8..5c1de6c 100644
--- a/tika-fetchers/tika-fetcher-s3/src/test/resources/tika-config-s3.xml
+++ b/tika-pipes/tika-fetchers/tika-fetcher-s3/src/test/resources/tika-config-s3.xml
@@ -19,6 +19,7 @@
<fetchers>
<fetcher class="org.apache.tika.fetcher.s3.S3Fetcher">
<params>
+ <param name="name" type="string">s3</param>
<param name="region" type="string">us-east-1</param>
<param name="profile" type="string">my_profile</param>
</params>
diff --git a/tika-emitters/tika-httpclient-commons/pom.xml b/tika-pipes/tika-httpclient-commons/pom.xml
similarity index 94%
rename from tika-emitters/tika-httpclient-commons/pom.xml
rename to tika-pipes/tika-httpclient-commons/pom.xml
index 8a5fa89..ef7797a 100644
--- a/tika-emitters/tika-httpclient-commons/pom.xml
+++ b/tika-pipes/tika-httpclient-commons/pom.xml
@@ -21,9 +21,10 @@
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
- <artifactId>tika-emitters</artifactId>
+ <artifactId>tika-pipes</artifactId>
<groupId>org.apache.tika</groupId>
<version>2.0.0-SNAPSHOT</version>
+ <relativePath>../pom.xml</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
diff --git a/tika-emitters/tika-httpclient-commons/src/main/java/org/apache/tika/client/HttpClientUtil.java b/tika-pipes/tika-httpclient-commons/src/main/java/org/apache/tika/client/HttpClientUtil.java
similarity index 100%
rename from tika-emitters/tika-httpclient-commons/src/main/java/org/apache/tika/client/HttpClientUtil.java
rename to tika-pipes/tika-httpclient-commons/src/main/java/org/apache/tika/client/HttpClientUtil.java
diff --git a/tika-emitters/tika-httpclient-commons/src/main/java/org/apache/tika/client/TikaClientException.java b/tika-pipes/tika-httpclient-commons/src/main/java/org/apache/tika/client/TikaClientException.java
similarity index 100%
rename from tika-emitters/tika-httpclient-commons/src/main/java/org/apache/tika/client/TikaClientException.java
rename to tika-pipes/tika-httpclient-commons/src/main/java/org/apache/tika/client/TikaClientException.java
diff --git a/tika-emitters/tika-emitter-solr/pom.xml b/tika-pipes/tika-pipes-integration-tests/pom.xml
similarity index 69%
rename from tika-emitters/tika-emitter-solr/pom.xml
rename to tika-pipes/tika-pipes-integration-tests/pom.xml
index 6aefb89..ccb138c 100644
--- a/tika-emitters/tika-emitter-solr/pom.xml
+++ b/tika-pipes/tika-pipes-integration-tests/pom.xml
@@ -21,30 +21,39 @@
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
- <artifactId>tika-emitters</artifactId>
+ <artifactId>tika-pipes</artifactId>
<groupId>org.apache.tika</groupId>
<version>2.0.0-SNAPSHOT</version>
+ <relativePath>../pom.xml</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
- <artifactId>tika-emitter-solr</artifactId>
+ <artifactId>tika-pipes-integration-tests</artifactId>
<dependencies>
<dependency>
- <groupId>${project.groupId}</groupId>
- <artifactId>tika-core</artifactId>
- <version>${project.version}</version>
- <scope>provided</scope>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ <scope>test</scope>
</dependency>
<dependency>
<groupId>${project.groupId}</groupId>
- <artifactId>tika-httpclient-commons</artifactId>
+ <artifactId>tika-core</artifactId>
<version>${project.version}</version>
+ <scope>test</scope>
</dependency>
- <dependency>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-log4j12</artifactId>
- </dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>tika-fetcher-s3</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>tika-fetch-iterator-s3</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
@@ -52,4 +61,4 @@
</dependency>
</dependencies>
-</project>
\ No newline at end of file
+</project>
diff --git a/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/PipeIntegrationTests.java b/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/PipeIntegrationTests.java
new file mode 100644
index 0000000..590a125
--- /dev/null
+++ b/tika-pipes/tika-pipes-integration-tests/src/test/java/org/apache/tika/pipes/PipeIntegrationTests.java
@@ -0,0 +1,162 @@
+/*
+ * 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;
+
+import com.amazonaws.auth.profile.ProfileCredentialsProvider;
+import com.amazonaws.services.s3.AmazonS3;
+import com.amazonaws.services.s3.AmazonS3Client;
+import com.amazonaws.services.s3.AmazonS3ClientBuilder;
+import com.amazonaws.services.s3.iterable.S3Objects;
+import com.amazonaws.services.s3.model.S3Object;
+import com.amazonaws.services.s3.model.S3ObjectSummary;
+import org.apache.tika.config.TikaConfig;
+import org.apache.tika.exception.TikaException;
+import org.apache.tika.pipes.emitter.Emitter;
+import org.apache.tika.pipes.fetcher.FetchIdMetadataPair;
+import org.apache.tika.pipes.fetcher.Fetcher;
+import org.apache.tika.pipes.fetchiterator.FetchIterator;
+import org.junit.Ignore;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorCompletionService;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+
+@Ignore("turn these into actual tests")
+public class PipeIntegrationTests {
+
+ private static final Path OUTDIR = Paths.get("");
+
+ @Test
+ public void testBruteForce() throws Exception {
+ String region = "";
+ String profile = "";
+ String bucket = "";
+ AmazonS3 s3Client = AmazonS3ClientBuilder.standard()
+ .withRegion(region)
+ .withCredentials(new ProfileCredentialsProvider(profile))
+ .build();
+ s3Client.listObjects(bucket);
+ int cnt = 0;
+ long sz = 0;
+
+ for (S3ObjectSummary summary : S3Objects.withPrefix(s3Client, bucket, "")) {
+ Path targ = OUTDIR.resolve(summary.getKey());
+ if (Files.isRegularFile(targ)) {
+ continue;
+ }
+ if (! Files.isDirectory(targ.getParent())) {
+ Files.createDirectories(targ.getParent());
+ }
+ System.out.println("id: " + cnt + " :: " + summary.getKey() + " : " + summary.getSize());
+ S3Object s3Object = s3Client.getObject(bucket, summary.getKey());
+ Files.copy(s3Object.getObjectContent(), targ);
+ summary.getSize();
+ cnt++;
+ sz += summary.getSize();
+ }
+ System.out.println("iterated: "+cnt + " sz: "+sz);
+ }
+
+ @Test
+ public void testS3ToFS() throws Exception {
+ TikaConfig tikaConfig = getConfig("tika-config-s3ToFs.xml");
+ FetchIterator it = tikaConfig.getFetchIterator();
+ int numConsumers = 1;
+ ExecutorService es = Executors.newFixedThreadPool(numConsumers + 1);
+ ExecutorCompletionService<Integer> completionService = new ExecutorCompletionService<>(es);
+ ArrayBlockingQueue<FetchIdMetadataPair> queue = it.init(numConsumers);
+ completionService.submit(it);
+ for (int i = 0; i < numConsumers; i++) {
+ completionService.submit(new FetcherEmitter(
+ queue, tikaConfig.getFetcherManager().getFetcher("s3"), null));
+ }
+ int finished = 0;
+ try {
+ while (finished++ < numConsumers+1) {
+ Future<Integer> future = completionService.take();
+ future.get();
+ }
+ } finally {
+ es.shutdownNow();
+ }
+
+
+ }
+
+ private TikaConfig getConfig(String fileName) throws Exception {
+ try (InputStream is =
+ PipeIntegrationTests.class.getResourceAsStream("/"+fileName)) {
+ return new TikaConfig(is);
+ }
+ }
+
+
+ private static class FetcherEmitter implements Callable<Integer> {
+ private static final AtomicInteger counter = new AtomicInteger(0);
+
+ private final Fetcher fetcher;
+ private final Emitter emitter;
+ private final ArrayBlockingQueue<FetchIdMetadataPair> queue;
+
+ FetcherEmitter(ArrayBlockingQueue<FetchIdMetadataPair> queue, Fetcher
+ fetcher, Emitter emitter) {
+ this.queue = queue;
+ this.fetcher = fetcher;
+ this.emitter = emitter;
+ }
+
+ @Override
+ public Integer call() throws Exception {
+
+ while (true) {
+ FetchIdMetadataPair p = queue.poll(5, TimeUnit.MINUTES);
+ if (p == null) {
+ throw new TimeoutException("");
+ }
+ if (p == FetchIterator.COMPLETED_SEMAPHORE) {
+ return 1;
+ }
+ process(p);
+ }
+ }
+
+ private void process(FetchIdMetadataPair p) throws IOException, TikaException {
+ Path targ = OUTDIR.resolve(p.getFetchId().getFetchKey());
+ if (Files.isRegularFile(targ)) {
+ return;
+ }
+ try (InputStream is = fetcher.fetch(p.getFetchId().getFetchKey(), p.getMetadata())) {
+ System.out.println(counter.getAndIncrement() + " : "+p );
+ Files.createDirectories(targ.getParent());
+ Files.copy(is, targ);
+ }
+ }
+ }
+}
diff --git a/tika-emitters/tika-emitter-solr/src/test/resources/log4j.properties b/tika-pipes/tika-pipes-integration-tests/src/test/resources/log4j.properties
similarity index 91%
rename from tika-emitters/tika-emitter-solr/src/test/resources/log4j.properties
rename to tika-pipes/tika-pipes-integration-tests/src/test/resources/log4j.properties
index 92b6d56..1bee240 100644
--- a/tika-emitters/tika-emitter-solr/src/test/resources/log4j.properties
+++ b/tika-pipes/tika-pipes-integration-tests/src/test/resources/log4j.properties
@@ -14,11 +14,11 @@
# limitations under the License.
#info,debug, error,fatal ...
-log4j.rootLogger=debug,stderr
+log4j.rootLogger=info,stderr
#console
log4j.appender.stderr=org.apache.log4j.ConsoleAppender
log4j.appender.stderr.layout=org.apache.log4j.PatternLayout
log4j.appender.stderr.Target=System.err
-log4j.appender.stderr.layout.ConversionPattern= %-5p %m%n
+log4j.appender.stderr.layout.ConversionPattern=%-5p [%t]: %m%n
diff --git a/tika-pipes/tika-pipes-integration-tests/src/test/resources/tika-config-s3ToFs.xml b/tika-pipes/tika-pipes-integration-tests/src/test/resources/tika-config-s3ToFs.xml
new file mode 100644
index 0000000..3e5bbc5
--- /dev/null
+++ b/tika-pipes/tika-pipes-integration-tests/src/test/resources/tika-config-s3ToFs.xml
@@ -0,0 +1,22 @@
+<?xml version="1.0" encoding="UTF-8" ?>
+<properties>
+ <fetchers>
+ <fetcher class="org.apache.tika.pipes.fetcher.s3.S3Fetcher">
+ <params>
+ <param name="name" type="string">s3</param>
+ <param name="region" type="string">us-east-1</param>
+ <param name="profile" type="string"><!-- fill in here --></param>
+ </params>
+ </fetcher>
+ </fetchers>
+ <fetchIterators>
+ <fetchIterator class="org.apache.tika.pipes.fetchiterator.s3.S3FetchIterator">
+ <params>
+ <param name="fetcherName" type="string">s3</param>
+ <param name="bucket" type="string"><!-- fill in here --></param>
+ <param name="region" type="string">us-east-1</param>
+ <param name="profile" type="string"><!-- fill in here --></param>
+ </params>
+ </fetchIterator>
+ </fetchIterators>
+</properties>
\ No newline at end of file
diff --git a/tika-server/tika-server-classic/src/test/java/org/apache/tika/server/classic/FetcherTest.java b/tika-server/tika-server-classic/src/test/java/org/apache/tika/server/classic/FetcherTest.java
index 3e5a15f..2343d4a 100644
--- a/tika-server/tika-server-classic/src/test/java/org/apache/tika/server/classic/FetcherTest.java
+++ b/tika-server/tika-server-classic/src/test/java/org/apache/tika/server/classic/FetcherTest.java
@@ -75,7 +75,7 @@ public class FetcherTest extends CXFTestBase {
@Override
protected InputStreamFactory getInputStreamFactory(TikaConfig tikaConfig) {
- return new FetcherStreamFactory(tikaConfig.getFetcher());
+ return new FetcherStreamFactory(tikaConfig.getFetcherManager());
}
@Test
@@ -84,7 +84,8 @@ public class FetcherTest extends CXFTestBase {
.create(endPoint + META_PATH)
.accept("application/json")
.acceptEncoding("gzip")
- .header("fetcherString", "url:https://tika.apache.org").put("");
+ .header("fetcherName", "url")
+ .header("fetchKey", "https://tika.apache.org").put("");
Reader reader = new InputStreamReader(new GzipCompressorInputStream((InputStream) response.getEntity()), UTF_8);
List<Metadata> metadataList = JsonMetadataList.fromJson(reader);
diff --git a/tika-server/tika-server-classic/src/test/resources/config/tika-config-url-fetcher.xml b/tika-server/tika-server-classic/src/test/resources/config/tika-config-url-fetcher.xml
index ea4079c..d3aaff0 100644
--- a/tika-server/tika-server-classic/src/test/resources/config/tika-config-url-fetcher.xml
+++ b/tika-server/tika-server-classic/src/test/resources/config/tika-config-url-fetcher.xml
@@ -22,6 +22,10 @@
<parser class="org.apache.tika.parser.DefaultParser"/>
</parsers>
<fetchers>
- <fetcher class="org.apache.tika.fetcher.SimpleUrlFetcher"/>
+ <fetcher class="org.apache.tika.pipes.fetcher.SimpleUrlFetcher">
+ <params>
+ <param name="name" type="string">url</param>
+ </params>
+ </fetcher>
</fetchers>
</properties>
\ No newline at end of file
diff --git a/tika-server/tika-server-client/src/main/java/org/apache/tika/server/client/TikaClient.java b/tika-server/tika-server-client/src/main/java/org/apache/tika/server/client/TikaClient.java
index f535f52..d7ebcf0 100644
--- a/tika-server/tika-server-client/src/main/java/org/apache/tika/server/client/TikaClient.java
+++ b/tika-server/tika-server-client/src/main/java/org/apache/tika/server/client/TikaClient.java
@@ -16,17 +16,15 @@
*/
package org.apache.tika.server.client;
-
-
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.google.gson.JsonArray;
import com.google.gson.JsonObject;
import com.google.gson.JsonPrimitive;
import org.apache.tika.config.TikaConfig;
-import org.apache.tika.exception.TikaConfigException;
import org.apache.tika.exception.TikaException;
import org.apache.tika.metadata.Metadata;
+import org.apache.tika.pipes.fetcher.FetchId;
import java.io.IOException;
import java.util.ArrayList;
@@ -59,17 +57,18 @@ public class TikaClient {
}*/
- public TikaEmitterResult parse(String fetcherString, Metadata metadata, String emitter)
+ public TikaEmitterResult parse(FetchId fetchId, Metadata metadata, String emitter)
throws IOException, TikaException {
TikaHttpClient client = getHttpClient();
- String jsonRequest = jsonifyRequest(fetcherString, metadata, emitter);
+ String jsonRequest = jsonifyRequest(fetchId, metadata, emitter);
return client.postJson(jsonRequest);
}
- private String jsonifyRequest(String fetcherString, Metadata metadata, String emitter) {
+ private String jsonifyRequest(FetchId fetchId, Metadata metadata, String emitter) {
JsonObject root = new JsonObject();
- root.add("fetcherString", new JsonPrimitive(fetcherString));
+ root.add("fetcherName", new JsonPrimitive(fetchId.getFetcherName()));
+ root.add("fetchKey", new JsonPrimitive(fetchId.getFetchKey()));
root.add("emitter", new JsonPrimitive(emitter));
if (metadata.size() > 0) {
JsonObject m = new JsonObject();
diff --git a/tika-server/tika-server-client/src/main/java/org/apache/tika/server/client/TikaClientCLI.java b/tika-server/tika-server-client/src/main/java/org/apache/tika/server/client/TikaClientCLI.java
index 08489c2..2ca9f97 100644
--- a/tika-server/tika-server-client/src/main/java/org/apache/tika/server/client/TikaClientCLI.java
+++ b/tika-server/tika-server-client/src/main/java/org/apache/tika/server/client/TikaClientCLI.java
@@ -18,9 +18,9 @@ package org.apache.tika.server.client;
import org.apache.tika.config.TikaConfig;
import org.apache.tika.exception.TikaException;
-import org.apache.tika.fetcher.FetchIterator;
-import org.apache.tika.fetcher.FetchMetadataPair;
-import org.apache.tika.fetcher.FileSystemFetchIterator;
+import org.apache.tika.pipes.fetchiterator.FetchIterator;
+import org.apache.tika.pipes.fetcher.FetchIdMetadataPair;
+import org.apache.tika.pipes.fetchiterator.FileSystemFetchIterator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.xml.sax.SAXException;
@@ -43,9 +43,6 @@ import java.util.concurrent.TimeoutException;
public class TikaClientCLI {
- private static final FetchMetadataPair POISON =
- new FetchMetadataPair(null, null);
-
private static final Logger LOGGER = LoggerFactory.getLogger(TikaClientCLI.class);
//make these configurable
@@ -66,14 +63,13 @@ public class TikaClientCLI {
throws TikaException, IOException, SAXException {
TikaConfig config = new TikaConfig(tikaConfigPath);
- ArrayBlockingQueue<FetchMetadataPair> queue = new ArrayBlockingQueue<>(1000);
- ExecutorService executorService = Executors.newFixedThreadPool(numThreads+2);
+ ExecutorService executorService = Executors.newFixedThreadPool(numThreads+1);
ExecutorCompletionService<Integer> completionService = new ExecutorCompletionService<>(executorService);
//TODO: fix this!
- final FetchIterator fetchIterator = new FileSystemFetchIterator(
- "fs", Paths.get("."));
+ final FetchIterator fetchIterator = config.getFetchIterator();
+ ArrayBlockingQueue<FetchIdMetadataPair> queue = fetchIterator.init(numThreads);
+
completionService.submit(fetchIterator);
- completionService.submit(new Enqueuer(queue, fetchIterator, numThreads));
if (tikaServerUrls.size() == numThreads) {
logDiffSizes(tikaServerUrls.size(), numThreads);
for (int i = 0; i < numThreads; i++) {
@@ -82,12 +78,14 @@ public class TikaClientCLI {
completionService.submit(new FetchWorker(queue, client, fetcherString));
}
} else {
- TikaClient client = TikaClient.get(config,tikaServerUrls);
- completionService.submit(new FetchWorker(queue, client, fetcherString));
+ for (int i = 0; i < numThreads; i++) {
+ TikaClient client = TikaClient.get(config, tikaServerUrls);
+ completionService.submit(new FetchWorker(queue, client, fetcherString));
+ }
}
int finished = 0;
- while (finished < numThreads+2) {
+ while (finished < numThreads+1) {
Future<Integer> future = null;
try {
future = completionService.poll(maxWaitMs, TimeUnit.MILLISECONDS);
@@ -107,7 +105,6 @@ public class TikaClientCLI {
}
}
}
-
}
private void logDiffSizes(int servers, int numThreads) {
@@ -116,44 +113,11 @@ public class TikaClientCLI {
servers, numThreads);
}
- private class Enqueuer implements Callable<Integer> {
- //simple class that pulls fetchmetadata pairs from the fetchiterator
- //and enqueues them for the worker threads.
- private final ArrayBlockingQueue<FetchMetadataPair> queue;
- private final FetchIterator fetchIterator;
- private final int numThreads;
-
- public Enqueuer(ArrayBlockingQueue<FetchMetadataPair> queue, FetchIterator fetchIterator, int numThreads) {
- this.queue = queue;
- this.fetchIterator = fetchIterator;
- this.numThreads = numThreads;
- }
-
- @Override
- public Integer call() throws Exception {
- System.out.println("enqueing");
- for (FetchMetadataPair p : fetchIterator) {
- System.out.println("offering "+p);
- boolean offered = queue.offer(p, maxWaitMs, TimeUnit.MILLISECONDS);
- if (! offered) {
- throw new TimeoutException("exceeded max wait");
- }
- }
- for (int i = 0; i < numThreads; i++) {
- boolean offered = queue.offer(POISON, maxWaitMs, TimeUnit.MILLISECONDS);
- if (! offered) {
- throw new TimeoutException("exceeded max wait");
- }
- }
- return 1;
- }
- }
-
private class FetchWorker implements Callable<Integer> {
- private final ArrayBlockingQueue<FetchMetadataPair> queue;
+ private final ArrayBlockingQueue<FetchIdMetadataPair> queue;
private final TikaClient client;
private final String emitterString;
- public FetchWorker(ArrayBlockingQueue<FetchMetadataPair> queue, TikaClient client,
+ public FetchWorker(ArrayBlockingQueue<FetchIdMetadataPair> queue, TikaClient client,
String emitterString) {
this.queue = queue;
this.client = client;
@@ -164,21 +128,21 @@ public class TikaClientCLI {
public Integer call() throws Exception {
while (true) {
- System.out.println("about to work");
- FetchMetadataPair p = queue.poll(maxWaitMs, TimeUnit.MILLISECONDS);
+
+ FetchIdMetadataPair p = queue.poll(maxWaitMs, TimeUnit.MILLISECONDS);
if (p == null) {
throw new TimeoutException("exceeded maxWaitMs");
}
- if (p == POISON) {
+ if (p == FetchIterator.COMPLETED_SEMAPHORE) {
return 1;
}
try {
- System.out.println("parsing; "+p.getFetcherString());
- System.out.println(client.parse(p.getFetcherString(), p.getMetadata(), emitterString));
+ LOGGER.debug("about to parse: {}", p.getFetchId());
+ client.parse(p.getFetchId(), p.getMetadata(), emitterString);
} catch (IOException e) {
- LOGGER.warn(p.getFetcherString(), e);
+ LOGGER.warn(p.getFetchId().toString(), e);
} catch (TikaException e) {
- LOGGER.warn(p.getFetcherString(), e);
+ LOGGER.warn(p.getFetchId().toString(), e);
}
}
}
diff --git a/tika-server/tika-server-client/src/test/resources/tika-config-simple-solr-emitter.xml b/tika-server/tika-server-client/src/test/resources/tika-config-simple-solr-emitter.xml
index ae5a9b2..56be391 100644
--- a/tika-server/tika-server-client/src/test/resources/tika-config-simple-solr-emitter.xml
+++ b/tika-server/tika-server-client/src/test/resources/tika-config-simple-solr-emitter.xml
@@ -20,7 +20,7 @@
<properties>
<service-loader initializableProblemHandler="throw"/>
<fetchIterators>
- <fetchIterator class="org.apache.tika.fetcher.FileSystemFetchIterator">
+ <fetchIterator class="org.apache.tika.pipes.fetchiterator.FileSystemFetchIterator">
<params>
<param name="fetchPrefix" type="string">fs</param>
<param name="basePath" type="string">fix</param>
@@ -28,7 +28,7 @@
</fetchIterator>
</fetchIterators>
<fetchers>
- <fetcher class="org.apache.tika.fetcher.FileSystemFetcher">
+ <fetcher class="org.apache.tika.pipes.fetcher.FileSystemFetcher">
<params>
<param name="name" type="string">fs</param>
<param name="basePath" type="string">fix</param>
diff --git a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/FetcherStreamFactory.java b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/FetcherStreamFactory.java
index 0841004..949ab57 100644
--- a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/FetcherStreamFactory.java
+++ b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/FetcherStreamFactory.java
@@ -19,12 +19,12 @@ package org.apache.tika.server.core;
import javax.ws.rs.core.HttpHeaders;
import java.io.IOException;
import java.io.InputStream;
-import java.net.URL;
+import org.apache.commons.lang3.StringUtils;
import org.apache.tika.exception.TikaException;
-import org.apache.tika.fetcher.Fetcher;
-import org.apache.tika.io.TikaInputStream;
+import org.apache.tika.pipes.fetcher.Fetcher;
import org.apache.tika.metadata.Metadata;
+import org.apache.tika.pipes.fetcher.FetcherManager;
/**
* This class looks for "fileUrl" in the http header. If it is not null
@@ -44,19 +44,25 @@ import org.apache.tika.metadata.Metadata;
*/
public class FetcherStreamFactory implements InputStreamFactory {
- private final Fetcher fetcher;
+ private final FetcherManager fetcherManager;
- public FetcherStreamFactory(Fetcher fetcher) {
- this.fetcher = fetcher;
+ public FetcherStreamFactory(FetcherManager fetcherManager) {
+ this.fetcherManager = fetcherManager;
}
+
@Override
public InputStream getInputSteam(InputStream is, Metadata metadata,
HttpHeaders httpHeaders) throws IOException {
- String fetcherString = httpHeaders.getHeaderString("fetcherString");
+ String fetcherName= httpHeaders.getHeaderString("fetcherName");
+ String fetchKey = httpHeaders.getHeaderString("fetchKey");
+ if (StringUtils.isBlank(fetcherName) != StringUtils.isBlank(fetchKey)) {
+ throw new IOException("Must specify both a 'fetcherName' and a 'fetchKey'. I see: "+
+ " fetcherName:"+fetcherName+" and fetchKey:"+fetchKey);
+ }
- if(fetcherString != null && !"".equals(fetcherString)){
+ if (!StringUtils.isBlank(fetcherName)){
try {
- return fetcher.fetch(fetcherString, metadata);
+ return fetcherManager.getFetcher(fetcherName).fetch(fetchKey, metadata);
} catch (TikaException e) {
throw new IOException(e);
}
diff --git a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/TikaServerCli.java b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/TikaServerCli.java
index 0035efb..6680751 100644
--- a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/TikaServerCli.java
+++ b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/TikaServerCli.java
@@ -45,7 +45,6 @@ import org.apache.cxf.transport.common.gzip.GZIPOutInterceptor;
import org.apache.tika.Tika;
import org.apache.tika.config.ServiceLoader;
import org.apache.tika.config.TikaConfig;
-import org.apache.tika.emitter.Emitter;
import org.apache.tika.parser.DigestingParser;
import org.apache.tika.parser.digestutils.BouncyCastleDigester;
import org.apache.tika.parser.digestutils.CommonsDigester;
@@ -278,7 +277,7 @@ public class TikaServerCli {
InputStreamFactory inputStreamFactory = null;
if (line.hasOption("enableUnsecureFeatures")) {
- inputStreamFactory = new FetcherStreamFactory(tika.getFetcher());
+ inputStreamFactory = new FetcherStreamFactory(tika.getFetcherManager());
LOG.info(UNSECURE_WARNING);
} else {
inputStreamFactory = new DefaultInputStreamFactory();
@@ -376,7 +375,7 @@ public class TikaServerCli {
private static void logFetchersAndEmitters(boolean enableUnsecureFeatures, TikaConfig tika) {
if (enableUnsecureFeatures) {
StringBuilder sb = new StringBuilder();
- Set<String> supportedFetchers = tika.getFetcher().getSupportedPrefixes();
+ Set<String> supportedFetchers = tika.getFetcherManager().getSupported();
sb.append("enableSecureFeatures has been selected.\n");
if (supportedFetchers.size() == 0) {
sb.append("There are no fetchers specified in the TikaConfig");
@@ -386,7 +385,7 @@ public class TikaServerCli {
sb.append(p).append("\n");
}
}
- Set<String> emitters = tika.getEmitter().getSupported();
+ Set<String> emitters = tika.getEmitterManager().getSupported();
if (supportedFetchers.size() == 0) {
sb.append("There are no emitters specified in the TikaConfig");
} else {
@@ -397,16 +396,16 @@ public class TikaServerCli {
}
LOG.info(sb.toString());
} else {
- if (tika.getEmitter().getSupported().size() > 0) {
+ if (tika.getEmitterManager().getSupported().size() > 0) {
String warn = "-enableUnsecureFeatures has not been specified on the commandline.\n"+
- "The "+tika.getEmitter().getSupported().size() + " emitter(s) that you've\n"+
+ "The "+tika.getEmitterManager().getSupported().size() + " emitter(s) that you've\n"+
"specified in TikaConfig will not be available on the /emit endpoint\n"+
"To enable your emitters, start tika-server with the -enableUnsecureFeatures flag\n\n";
LOG.warn(warn);
}
- if (tika.getFetcher().getSupportedPrefixes().size() > 0) {
+ if (tika.getFetcherManager().getSupported().size() > 0) {
String warn = "-enableUnsecureFeatures has not been specified on the commandline.\n"+
- "The "+tika.getFetcher().getSupportedPrefixes().size() + " fetcher(s) that you've\n"+
+ "The "+tika.getFetcherManager().getSupported().size() + " fetcher(s) that you've\n"+
"specified in TikaConfig will not be available\n"+
"To enable your fetchers, start tika-server with the -enableUnsecureFeatures flag\n\n";
LOG.warn(warn);
diff --git a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/EmitterResource.java b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/EmitterResource.java
index 008feed..7f116fa 100644
--- a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/EmitterResource.java
+++ b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/EmitterResource.java
@@ -20,10 +20,11 @@ package org.apache.tika.server.core.resource;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;
-import org.apache.tika.emitter.Emitter;
-import org.apache.tika.emitter.TikaEmitterException;
+import org.apache.tika.pipes.emitter.Emitter;
+import org.apache.tika.pipes.emitter.TikaEmitterException;
import org.apache.tika.exception.TikaException;
-import org.apache.tika.fetcher.Fetcher;
+import org.apache.tika.pipes.fetcher.FetchId;
+import org.apache.tika.pipes.fetcher.Fetcher;
import org.apache.tika.metadata.Metadata;
import org.apache.tika.metadata.TikaCoreProperties;
import org.apache.tika.sax.AbstractRecursiveParserWrapperHandler;
@@ -54,7 +55,8 @@ import java.util.Map;
public class EmitterResource {
private static final String EMITTER_PARAM = "emitter";
- private static final String FETCH_STRING_ABBREV = "f";
+ private static final String FETCHER_NAME_ABBREV = "fn";
+ private static final String FETCH_KEY_ABBREV = "fk";
/**
* key that is safe to pass through http header.
@@ -69,24 +71,23 @@ public class EmitterResource {
* @param httpHeaders
* @param info
* @param emitterName
- * @param fetcherString specify the fetch string in the url's query section
+ * @param fetcherName specify the fetcherName in the url's query section
+ * @param fetchKey specify the fetch key in the url's query section
* @return
* @throws Exception
*/
@GET
@Produces("application/json")
@Path("{" + EMITTER_PARAM + " : (\\w+)?}")
- public Map<String, String> getMetadataAbbrev(InputStream is, @Context HttpHeaders httpHeaders,
+ public Map<String, String> getRmeta(InputStream is, @Context HttpHeaders httpHeaders,
@Context UriInfo info,
@PathParam(EMITTER_PARAM) String emitterName,
- @QueryParam(FETCH_STRING_ABBREV) String fetcherString) throws Exception {
+ @QueryParam(FETCHER_NAME_ABBREV) String fetcherName,
+ @QueryParam(FETCH_KEY_ABBREV) String fetchKey) throws Exception {
Metadata metadata = new Metadata();
- Fetcher fetcher = TikaResource.getConfig().getFetcher();
- System.out.println("FETCHER: " + fetcher.getClass() + " : " + fetcher.getSupportedPrefixes());
- System.out.println("emitter: " + TikaResource.getConfig().getEmitter().getClass()
- + " : " + TikaResource.getConfig().getEmitter().getSupported());
+ Fetcher fetcher = TikaResource.getConfig().getFetcherManager().getFetcher(fetcherName);
List<Metadata> metadataList;
- try (InputStream fetchedIs = fetcher.fetch(fetcherString, metadata)) {
+ try (InputStream fetchedIs = fetcher.fetch(fetchKey, metadata)) {
for (String n : metadata.names()) {
System.out.println(n + " ; "+metadata.get(n));
}
@@ -116,7 +117,7 @@ public class EmitterResource {
@PUT
@Produces("application/json")
@Path("{" + EMITTER_PARAM + " : (\\w+)?}")
- public Map<String, String> getMetadataFromInputStream(InputStream is,
+ public Map<String, String> putRmeta(InputStream is,
@Context HttpHeaders httpHeaders,
@Context UriInfo info,
@PathParam(EMITTER_PARAM) String emitterName
@@ -140,7 +141,7 @@ public class EmitterResource {
* through of metadata from the client.
* <p>
* The extracted text content is stored with the key
- * {@link org.apache.tika.sax.AbstractRecursiveParserWrapperHandler#TIKA_CONTENT}
+ * {@link AbstractRecursiveParserWrapperHandler#TIKA_CONTENT}
* <p>
* Must specify a fetcherString and an emitter in the posted json.
* @param info uri info
@@ -149,7 +150,7 @@ public class EmitterResource {
*/
@POST
@Produces("application/json")
- public Map<String, String> getMetadataFromPost(InputStream is,
+ public Map<String, String> postRmeta(InputStream is,
@Context HttpHeaders httpHeaders,
@Context UriInfo info
) throws Exception {
@@ -157,9 +158,35 @@ public class EmitterResource {
try (Reader reader = new InputStreamReader(is, StandardCharsets.UTF_8)) {
root = JsonParser.parseReader(reader);
}
- String fetcherString = root.getAsJsonObject().get("fetcherString").getAsString();
+ String fetcherName = root.getAsJsonObject().get("fetcherName").getAsString();
+ String fetchKey = root.getAsJsonObject().get("fetchKey").getAsString();
String emitterName = root.getAsJsonObject().get("emitter").getAsString();
Metadata metadata = new Metadata();
+
+
+ List<Metadata> metadataList = null;
+ try (InputStream stream =
+ TikaResource.getConfig().getFetcherManager()
+ .getFetcher(fetcherName).fetch(fetchKey, metadata)) {
+
+ metadataList = RecursiveMetadataResource.parseMetadata(
+ stream,
+ metadata,
+ httpHeaders.getRequestHeaders(), info, "text");
+ } catch (Error error) {
+ return returnError(emitterName, error);
+ }
+
+ injectUserMetadata(metadataList.get(0), root.getAsJsonObject());
+
+ for (String n : metadataList.get(0).names()) {
+ LOG.debug("post parse/pre emit metadata {}: {}",
+ n, metadataList.get(0).get(n));
+ }
+ return emit(emitterName, metadataList);
+ }
+
+ private void injectUserMetadata(Metadata metadata, JsonObject root) {
if (root.getAsJsonObject().has("metadata")) {
JsonObject meta = root.getAsJsonObject().getAsJsonObject("metadata");
for (String k : meta.keySet()) {
@@ -173,25 +200,6 @@ public class EmitterResource {
}
}
}
- for (String n : metadata.names()) {
- System.out.println(n + " ; "+metadata.get(n));
- }
- List<Metadata> metadataList = null;
- try (InputStream stream = TikaResource.getConfig().getFetcher().fetch(fetcherString, metadata)) {
- for (String n : metadata.names()) {
- System.out.println("2: " + n + " ; "+metadata.get(n));
- }
- metadataList = RecursiveMetadataResource.parseMetadata(
- stream,
- metadata,
- httpHeaders.getRequestHeaders(), info, "text");
- } catch (Error error) {
- return returnError(emitterName, error);
- }
- for (String n : metadataList.get(0).names()) {
- System.out.println("3: " + n + " ; "+metadataList.get(0).get(n));
- }
- return emit(emitterName, metadataList);
}
private Map<String, String> returnError(String emitterName, Error error) {
@@ -201,15 +209,14 @@ public class EmitterResource {
String msg = ExceptionUtils.getStackTrace(error);
statusMap.put("parse_error", msg);
return statusMap;
-
}
private Map<String, String> emit(String emitterName, List<Metadata> metadataList) throws TikaException {
- Emitter emitter = TikaResource.getConfig().getEmitter();
+ Emitter emitter = TikaResource.getConfig().getEmitterManager().getEmitter(emitterName);
String status = "ok";
String exceptionMsg = "";
try {
- emitter.emit(emitterName, metadataList);
+ emitter.emit(metadataList);
} catch (IOException|TikaEmitterException e) {
LOG.warn("problem with emitting", e);
status = "emitter_exception";
diff --git a/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaEmitterTest.java b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaEmitterTest.java
index c02f078..0631eef 100644
--- a/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaEmitterTest.java
+++ b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaEmitterTest.java
@@ -64,7 +64,7 @@ import static org.junit.Assert.assertNotNull;
public class TikaEmitterTest extends CXFTestBase {
private static final String EMITTER_PATH = "/emit";
- private static final String EMITTER_PATH_AND_FS = "/emit/fs";
+ private static final String EMITTER_PATH_AND_FS = "/emit/fse";
private static Path TMP_DIR;
private static Path TMP_OUTPUT_DIR;
private static Path TMP_OUTPUT_FILE;
@@ -91,8 +91,9 @@ public class TikaEmitterTest extends CXFTestBase {
TIKA_CONFIG_XML = "<?xml version=\"1.0\" encoding=\"UTF-8\"?>"+
"<properties>"+
"<fetchers>"+
- "<fetcher class=\"org.apache.tika.fetcher.FileSystemFetcher\">"+
+ "<fetcher class=\"org.apache.tika.pipes.fetcher.FileSystemFetcher\">"+
"<params>"+
+ "<param name=\"name\" type=\"string\">fsf</param>"+
"<param name=\"basePath\" type=\"string\">"+inputDir.toAbsolutePath()+"</param>"+
"</params>"+
"</fetcher>"+
@@ -100,6 +101,7 @@ public class TikaEmitterTest extends CXFTestBase {
"<emitters>"+
"<emitter class=\"org.apache.tika.emitter.fs.FileSystemEmitter\">"+
"<params>"+
+ "<param name=\"name\" type=\"string\">fse</param>"+
"<param name=\"basePath\" type=\"string\">"+ TMP_OUTPUT_DIR.toAbsolutePath()+"</param>"+
"</params>"+
"</emitter>"+
@@ -141,14 +143,13 @@ public class TikaEmitterTest extends CXFTestBase {
@Override
protected InputStreamFactory getInputStreamFactory(TikaConfig tikaConfig) {
- return new FetcherStreamFactory(tikaConfig.getFetcher());
+ return new FetcherStreamFactory(tikaConfig.getFetcherManager());
}
@Test
public void testGet() throws Exception {
- String q = "?f="+
- URLEncoder.encode("fs:hello_world.xml", StandardCharsets.UTF_8.name());
+ String q = "?fn=fsf&fk=hello_world.xml";
String getUrl = endPoint+EMITTER_PATH_AND_FS+q;
Response response = WebClient
.create(getUrl)
@@ -171,8 +172,9 @@ public class TikaEmitterTest extends CXFTestBase {
public void testPost() throws Exception {
JsonObject root = new JsonObject();
- root.add("fetcherString", new JsonPrimitive("fs:hello_world.xml"));
- root.add("emitter", new JsonPrimitive("fs"));
+ root.add("fetcherName", new JsonPrimitive("fsf"));
+ root.add("fetchKey", new JsonPrimitive("hello_world.xml"));
+ root.add("emitter", new JsonPrimitive("fse"));
JsonObject userMetadata = new JsonObject();
String[] valueArray = new String[] {"my-value-1", "my-value-2", "my-value-3"};
JsonArray arr = new JsonArray();
@@ -239,8 +241,9 @@ public class TikaEmitterTest extends CXFTestBase {
public void testPostNPE() throws Exception {
JsonObject root = new JsonObject();
- root.add("fetcherString", new JsonPrimitive("fs:null_pointer.xml"));
- root.add("emitter", new JsonPrimitive("fs"));
+ root.add("fetcherName", new JsonPrimitive("fsf"));
+ root.add("fetchKey", new JsonPrimitive("null_pointer.xml"));
+ root.add("emitter", new JsonPrimitive("fse"));
JsonObject userMetadata = new JsonObject();
String[] valueArray = new String[] {"my-value-1", "my-value-2", "my-value-3"};
JsonArray arr = new JsonArray();
diff --git a/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerEmitterIntegrationTest.java b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerEmitterIntegrationTest.java
index acb4657..3fa10f5 100644
--- a/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerEmitterIntegrationTest.java
+++ b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerEmitterIntegrationTest.java
@@ -60,6 +60,9 @@ public class TikaServerEmitterIntegrationTest extends IntegrationTestBase {
private static Path TIKA_CONFIG;
private static Gson GSON = new GsonBuilder().create();
+ private static final String EMITTER_NAME = "fse";
+ private static final String FETCHER_NAME = "fsf";
+
private static String[] FILES = new String[] {
"hello_world.xml",
"heavy_hang_30000.xml", "real_oom.xml", "system_exit.xml"
@@ -83,8 +86,9 @@ public class TikaServerEmitterIntegrationTest extends IntegrationTestBase {
TIKA_CONFIG_XML = "<?xml version=\"1.0\" encoding=\"UTF-8\"?>"+
"<properties>"+
"<fetchers>"+
- "<fetcher class=\"org.apache.tika.fetcher.FileSystemFetcher\">"+
+ "<fetcher class=\"org.apache.tika.pipes.fetcher.FileSystemFetcher\">"+
"<params>"+
+ "<param name=\"name\" type=\"string\">"+FETCHER_NAME+"</param>"+
"<param name=\"basePath\" type=\"string\">"+inputDir.toAbsolutePath()+"</param>"+
"</params>"+
"</fetcher>"+
@@ -92,6 +96,8 @@ public class TikaServerEmitterIntegrationTest extends IntegrationTestBase {
"<emitters>"+
"<emitter class=\"org.apache.tika.emitter.fs.FileSystemEmitter\">"+
"<params>"+
+ "<param name=\"name\" type=\"string\">"+EMITTER_NAME+"</param>"+
+
"<param name=\"basePath\" type=\"string\">"+ TMP_OUTPUT_DIR.toAbsolutePath()+"</param>"+
"</params>"+
"</emitter>"+
@@ -261,8 +267,9 @@ public class TikaServerEmitterIntegrationTest extends IntegrationTestBase {
private String getJsonString(String fileName) {
JsonObject root = new JsonObject();
- root.add("fetcherString", new JsonPrimitive("fs:"+fileName));
- root.add("emitter", new JsonPrimitive("fs"));
+ root.add("fetcherName", new JsonPrimitive(FETCHER_NAME));
+ root.add("fetchKey", new JsonPrimitive(fileName));
+ root.add("emitter", new JsonPrimitive(EMITTER_NAME));
return GSON.toJson(root);
}
}
diff --git a/tika-server/tika-server-core/src/test/resources/test-documents/mock/hello_world.xml b/tika-server/tika-server-core/src/test/resources/test-documents/mock/hello_world.xml
index e1cb32b..27cd62a 100644
--- a/tika-server/tika-server-core/src/test/resources/test-documents/mock/hello_world.xml
+++ b/tika-server/tika-server-core/src/test/resources/test-documents/mock/hello_world.xml
@@ -21,5 +21,6 @@
<mock>
<metadata action="add" name="author">Nikolai Lobachevsky</metadata>
<metadata action="add" name="title">你好，世界</metadata>
+ <metadata action="add" name="my-key">parsers-value</metadata>
<write element="p">hello world</write>
</mock>
\ No newline at end of file