Skip to content

Commit ccce053

Browse files
tballisontballison
committed
TIKA-4498 -- add a PassbackFilter to enable passing back some of the parsed output to the client parser.
1 parent a3ffce5 commit ccce053

6 files changed

Lines changed: 251 additions & 11 deletions

File tree

‎tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PassbackFilter.java‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,28 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
package org.apache.tika.pipes.core;
18+
19+
import java.io.Serializable;
20+
21+
import org.apache.tika.metadata.listfilter.MetadataListFilter;
22+
23+
/**
24+
* Filter/Select some of the emitted output and pass it back to the client parser.
25+
*/
26+
public abstract class PassbackFilter extends MetadataListFilter implements Serializable {
27+
28+
}

‎tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesClient.java‎

Lines changed: 9 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -334,7 +334,11 @@ private PipesResult readResults(FetchEmitTuple t, long start) throws IOException
334334
//there may have been a parse exception, but the parse didn't crash
335335
LOG.debug("pipesClientId={} parse success: {} in {} ms", pipesClientId, t.getId(),
336336
millis);
337-
return deserializeEmitData();
337+
return deserializeEmitData(PipesResult.STATUS.PARSE_SUCCESS);
338+
case EMIT_SUCCESS_PASS_BACK:
339+
LOG.debug("pipesClientId={} emit success with passback: {} in {} ms", pipesClientId, t.getId(),
340+
millis);
341+
return deserializeEmitData(PipesResult.STATUS.EMIT_SUCCESS_PASSBACK);
338342
case PARSE_EXCEPTION_NO_EMIT:
339343
return readMessage(PipesResult.STATUS.PARSE_EXCEPTION_NO_EMIT);
340344
case EMIT_SUCCESS:
@@ -366,7 +370,7 @@ private PipesResult readMessage(PipesResult.STATUS status) throws IOException {
366370
return new PipesResult(status, msg);
367371
}
368372

369-
private PipesResult deserializeEmitData() throws IOException {
373+
private PipesResult deserializeEmitData(PipesResult.STATUS status) throws IOException {
370374
int length = input.readInt();
371375
byte[] bytes = new byte[length];
372376
input.readFully(bytes);
@@ -376,9 +380,9 @@ private PipesResult deserializeEmitData() throws IOException {
376380

377381
String stack = emitData.getContainerStackTrace();
378382
if (StringUtils.isBlank(stack)) {
379-
return new PipesResult(emitData);
383+
return new PipesResult(status, emitData, false);
380384
} else {
381-
return new PipesResult(emitData, stack);
385+
return new PipesResult(status, emitData, stack, false);
382386
}
383387
} catch (ClassNotFoundException e) {
384388
LOG.error("class not found exception deserializing data", e);
@@ -387,6 +391,7 @@ private PipesResult deserializeEmitData() throws IOException {
387391
}
388392
}
389393

394+
390395
private PipesResult deserializeIntermediateResult(EmitKey emitKey, ParseContext parseContext) throws IOException {
391396

392397
int length = input.readInt();

‎tika-pipes/tika-pipes-core/src/main/java/org/apache/tika/pipes/core/PipesResult.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,7 @@ public enum STATUS {
3434
OOM, TIMEOUT, UNSPECIFIED_CRASH,
3535
NO_EMITTER_FOUND,
3636
EMIT_SUCCESS, EMIT_SUCCESS_PARSE_EXCEPTION, EMIT_EXCEPTION,
37+
EMIT_SUCCESS_PASSBACK,//emit happened and some data is returned
3738
INTERRUPTED_EXCEPTION, NO_FETCHER_FOUND,
3839
INTERMEDIATE_RESULT;
3940
}
@@ -51,7 +52,7 @@ public enum STATUS {
5152
private final EmitData emitData;
5253
private final String message;
5354

54-
private PipesResult(STATUS status, EmitData emitData, String message, boolean intermediate) {
55+
public PipesResult(STATUS status, EmitData emitData, String message, boolean intermediate) {
5556
this.status = status;
5657
this.emitData = emitData;
5758
this.message = message;

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

Lines changed: 38 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -105,7 +105,7 @@ public class PipesServer implements Runnable {
105105
public enum STATUS {
106106
READY, CALL, PING, FAILED_TO_START, FETCHER_NOT_FOUND, EMITTER_NOT_FOUND,
107107
FETCHER_INITIALIZATION_EXCEPTION, FETCH_EXCEPTION, PARSE_SUCCESS, PARSE_EXCEPTION_NO_EMIT,
108-
EMIT_SUCCESS, EMIT_SUCCESS_PARSE_EXCEPTION, EMIT_EXCEPTION, OOM, TIMEOUT, EMPTY_OUTPUT,
108+
EMIT_SUCCESS, EMIT_SUCCESS_PASS_BACK, EMIT_SUCCESS_PARSE_EXCEPTION, EMIT_EXCEPTION, OOM, TIMEOUT, EMPTY_OUTPUT,
109109
INTERMEDIATE_RESULT;
110110

111111
byte getByte() {
@@ -312,12 +312,44 @@ private void emit(String taskId, EmitKey emitKey,
312312
write(STATUS.EMIT_EXCEPTION, bytes);
313313
return;
314314
}
315-
if (StringUtils.isBlank(parseExceptionStack)) {
316-
write(STATUS.EMIT_SUCCESS);
317-
} else {
318-
write(STATUS.EMIT_SUCCESS_PARSE_EXCEPTION,
319-
parseExceptionStack.getBytes(StandardCharsets.UTF_8));
315+
writeEmitResponse(emitKey, parseData.metadataList, parseExceptionStack, parseContext);
316+
317+
}
318+
319+
private void writeEmitResponse(EmitKey emitKey, List<Metadata> metadataList, String parseExceptionStack, ParseContext parseContext) {
320+
321+
STATUS status = (StringUtils.isBlank(parseExceptionStack)) ? STATUS.EMIT_SUCCESS :
322+
STATUS.EMIT_SUCCESS_PARSE_EXCEPTION;
323+
PassbackFilter filter = parseContext.get(PassbackFilter.class);
324+
if (filter == null) {
325+
if (status == STATUS.EMIT_SUCCESS) {
326+
write(status);
327+
} else {
328+
write(status, parseExceptionStack.getBytes(StandardCharsets.UTF_8));
329+
}
330+
return;
320331
}
332+
List<Metadata> filtered = null;
333+
try {
334+
filtered = filter.filter(metadataList);
335+
} catch (TikaException e) {
336+
LOG.error("problem filtering data for passback", e);
337+
exit(1);
338+
}
339+
340+
EmitData filteredEmitData = new EmitData(emitKey, filtered, parseExceptionStack);
341+
342+
try {
343+
UnsynchronizedByteArrayOutputStream bos = UnsynchronizedByteArrayOutputStream.builder().get();
344+
try (ObjectOutputStream objectOutputStream = new ObjectOutputStream(bos)) {
345+
objectOutputStream.writeObject(filteredEmitData);
346+
}
347+
write(STATUS.EMIT_SUCCESS_PASS_BACK, bos.toByteArray());
348+
} catch (IOException e) {
349+
LOG.error("problem writing response data (forking process shutdown?)", e);
350+
exit(1);
351+
}
352+
321353
}
322354

323355
private void emitContentsAndBytes(Emitter emitter, EmitKey emitKey,

‎tika-pipes/tika-pipes-core/src/test/java/org/apache/tika/pipes/core/PassbackFilterTest.java‎

Lines changed: 127 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,127 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
package org.apache.tika.pipes.core;
18+
19+
import static org.junit.jupiter.api.Assertions.assertEquals;
20+
import static org.junit.jupiter.api.Assertions.assertNull;
21+
22+
import java.io.IOException;
23+
import java.nio.charset.StandardCharsets;
24+
import java.nio.file.Files;
25+
import java.nio.file.Path;
26+
import java.nio.file.Paths;
27+
import java.util.ArrayList;
28+
import java.util.List;
29+
import java.util.Locale;
30+
import javax.xml.parsers.ParserConfigurationException;
31+
32+
import org.apache.commons.io.FileUtils;
33+
import org.junit.jupiter.api.AfterEach;
34+
import org.junit.jupiter.api.Assertions;
35+
import org.junit.jupiter.api.BeforeEach;
36+
import org.junit.jupiter.api.Test;
37+
import org.xml.sax.SAXException;
38+
39+
import org.apache.tika.exception.TikaConfigException;
40+
import org.apache.tika.exception.TikaException;
41+
import org.apache.tika.metadata.Metadata;
42+
import org.apache.tika.metadata.TikaCoreProperties;
43+
import org.apache.tika.parser.ParseContext;
44+
import org.apache.tika.pipes.core.emitter.EmitKey;
45+
import org.apache.tika.pipes.core.fetcher.FetchKey;
46+
import org.apache.tika.serialization.JsonMetadataList;
47+
import org.apache.tika.utils.StringUtils;
48+
49+
public class PassbackFilterTest {
50+
51+
private Path tmpDir;
52+
String fetcherName = "fs";
53+
String testPdfFile = "testOverlappingText.pdf";
54+
55+
private PipesClient pipesClient;
56+
57+
@BeforeEach
58+
public void init() throws TikaConfigException, IOException, ParserConfigurationException, SAXException {
59+
Path tikaConfigTemplate = Paths.get("src", "test", "resources", "org", "apache", "tika", "pipes", "core", "tika-emit-config.xml");
60+
tmpDir = Files.createTempDirectory("tika-pipes");
61+
Path tikaConfigPath = Files.createTempFile(tmpDir, "tika-pipes-", ".xml");
62+
String template = Files.readString(tikaConfigTemplate, StandardCharsets.UTF_8);
63+
template = template.replace("EMITTER_BASE_PATH", tmpDir
64+
.toAbsolutePath()
65+
.toString());
66+
Files.writeString(tikaConfigPath, template);
67+
PipesConfig pipesConfig = PipesConfig.load(tikaConfigPath);
68+
pipesClient = new PipesClient(pipesConfig);
69+
}
70+
71+
@AfterEach
72+
public void tearDown() throws IOException {
73+
FileUtils.deleteDirectory(tmpDir.toFile());
74+
}
75+
76+
@Test
77+
public void testPassbackFilter() throws Exception {
78+
String emitFileBase = "blah";
79+
ParseContext parseContext = new ParseContext();
80+
parseContext.set(PassbackFilter.class, new MyPassbackFilter());
81+
PipesResult pipesResult = pipesClient.process(
82+
new FetchEmitTuple(testPdfFile, new FetchKey(fetcherName, testPdfFile), new EmitKey("fs", emitFileBase), new Metadata(), parseContext,
83+
FetchEmitTuple.ON_PARSE_EXCEPTION.SKIP));
84+
assertEquals(PipesResult.STATUS.EMIT_SUCCESS_PASSBACK, pipesResult.getStatus());
85+
Assertions.assertNotNull(pipesResult
86+
.getEmitData()
87+
.getMetadataList());
88+
assertEquals(1, pipesResult
89+
.getEmitData()
90+
.getMetadataList()
91+
.size());
92+
Metadata metadata = pipesResult
93+
.getEmitData()
94+
.getMetadataList()
95+
.get(0);
96+
assertEquals("TESTOVERLAPPINGTEXT.PDF", metadata.get(TikaCoreProperties.RESOURCE_NAME_KEY));
97+
assertNull(metadata.get(Metadata.CONTENT_TYPE));
98+
assertNull(metadata.get(Metadata.CONTENT_LENGTH));
99+
assertEquals(1, metadata.names().length);
100+
101+
List<Metadata> metadataList = JsonMetadataList.fromJson(Files.newBufferedReader(tmpDir.resolve(emitFileBase + ".json"), StandardCharsets.UTF_8));
102+
assertEquals(1, metadataList.size());
103+
assertEquals("application/pdf", metadataList
104+
.get(0)
105+
.get(Metadata.CONTENT_TYPE));
106+
assertEquals("899", metadataList
107+
.get(0)
108+
.get(Metadata.CONTENT_LENGTH));
109+
}
110+
111+
private static class MyPassbackFilter extends PassbackFilter {
112+
@Override
113+
public List<Metadata> filter(List<Metadata> metadataList) throws TikaException {
114+
List<Metadata> ret = new ArrayList<>();
115+
for (Metadata m : metadataList) {
116+
String val = m.get(TikaCoreProperties.RESOURCE_NAME_KEY);
117+
if (!StringUtils.isBlank(val)) {
118+
Metadata retM = new Metadata();
119+
retM.add(TikaCoreProperties.RESOURCE_NAME_KEY, val.toUpperCase(Locale.ROOT));
120+
ret.add(retM);
121+
}
122+
}
123+
return ret;
124+
}
125+
}
126+
127+
}

‎tika-pipes/tika-pipes-core/src/test/resources/org/apache/tika/pipes/core/tika-emit-config.xml‎

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,47 @@
1+
<?xml version="1.0" encoding="UTF-8"?>
2+
<!--
3+
Licensed to the Apache Software Foundation (ASF) under one or more
4+
contributor license agreements. See the NOTICE file distributed with
5+
this work for additional information regarding copyright ownership.
6+
The ASF licenses this file to You under the Apache License, Version 2.0
7+
(the "License"); you may not use this file except in compliance with
8+
the License. You may obtain a copy of the License at
9+
10+
http://www.apache.org/licenses/LICENSE-2.0
11+
12+
Unless required by applicable law or agreed to in writing, software
13+
distributed under the License is distributed on an "AS IS" BASIS,
14+
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
See the License for the specific language governing permissions and
16+
limitations under the License.
17+
-->
18+
<properties>
19+
<pipes>
20+
<params>
21+
<numClients>2</numClients>
22+
<forkedJvmArgs>
23+
<arg>-Xmx1g</arg>
24+
<arg>-XX:ParallelGCThreads=2</arg>
25+
</forkedJvmArgs>
26+
<timeoutMillis>60000</timeoutMillis>
27+
<maxForEmitBatchBytes>0</maxForEmitBatchBytes> <!-- always emit -->
28+
</params>
29+
</pipes>
30+
<autoDetectParserConfig>
31+
<digesterFactory class="org.apache.tika.pipes.core.async.MockDigesterFactory">
32+
<skipContainerDocument>false</skipContainerDocument>
33+
</digesterFactory>
34+
</autoDetectParserConfig>
35+
<fetchers>
36+
<fetcher class="org.apache.tika.pipes.fetcher.fs.FileSystemFetcher">
37+
<name>fs</name>
38+
<basePath>src/test/resources/test-documents</basePath>
39+
</fetcher>
40+
</fetchers>
41+
<emitters>
42+
<emitter class="org.apache.tika.pipes.emitter.fs.FileSystemEmitter">
43+
<name>fs</name>
44+
<basePath>EMITTER_BASE_PATH</basePath>
45+
</emitter>
46+
</emitters>
47+
</properties>

0 commit comments

Comments
 (0)