Skip to content

Commit baee4e3

Browse files
tballisontballison
authored
TIKA-4503 -- refactor serialization (#2351)
* TIKA-4503 -- refactor serialization
1 parent 2d88518 commit baee4e3

14 files changed

Lines changed: 510 additions & 487 deletions

File tree

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

Lines changed: 126 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,126 @@
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.serialization;
18+
19+
import static org.apache.tika.pipes.core.serialization.FetchEmitTupleSerializer.EMIT_KEY;
20+
import static org.apache.tika.pipes.core.serialization.FetchEmitTupleSerializer.EMITTER;
21+
import static org.apache.tika.pipes.core.serialization.FetchEmitTupleSerializer.FETCHER;
22+
import static org.apache.tika.pipes.core.serialization.FetchEmitTupleSerializer.FETCH_KEY;
23+
import static org.apache.tika.pipes.core.serialization.FetchEmitTupleSerializer.FETCH_RANGE_END;
24+
import static org.apache.tika.pipes.core.serialization.FetchEmitTupleSerializer.FETCH_RANGE_START;
25+
import static org.apache.tika.pipes.core.serialization.FetchEmitTupleSerializer.ID;
26+
import static org.apache.tika.pipes.core.serialization.FetchEmitTupleSerializer.METADATA_KEY;
27+
import static org.apache.tika.pipes.core.serialization.FetchEmitTupleSerializer.ON_PARSE_EXCEPTION;
28+
import static org.apache.tika.serialization.ParseContextSerializer.PARSE_CONTEXT;
29+
30+
import java.io.IOException;
31+
import java.util.Map;
32+
33+
import com.fasterxml.jackson.core.JacksonException;
34+
import com.fasterxml.jackson.core.JsonParser;
35+
import com.fasterxml.jackson.databind.DeserializationContext;
36+
import com.fasterxml.jackson.databind.JsonDeserializer;
37+
import com.fasterxml.jackson.databind.JsonNode;
38+
39+
import org.apache.tika.metadata.Metadata;
40+
import org.apache.tika.parser.ParseContext;
41+
import org.apache.tika.pipes.core.FetchEmitTuple;
42+
import org.apache.tika.pipes.core.emitter.EmitKey;
43+
import org.apache.tika.pipes.core.fetcher.FetchKey;
44+
import org.apache.tika.serialization.ParseContextDeserializer;
45+
46+
public class FetchEmitTupleDeserializer extends JsonDeserializer<FetchEmitTuple> {
47+
48+
@Override
49+
public FetchEmitTuple deserialize(JsonParser jsonParser, DeserializationContext deserializationContext) throws IOException, JacksonException {
50+
JsonNode root = jsonParser.readValueAsTree();
51+
52+
String id = readVal(ID, root, null, true);
53+
String fetcherName = readVal(FETCHER, root, null, true);
54+
String fetchKey = readVal(FETCH_KEY, root, null, true);
55+
String emitterName = readVal(EMITTER, root, "", false);
56+
String emitKey = readVal(EMIT_KEY, root, "", false);
57+
long fetchRangeStart = readLong(FETCH_RANGE_START, root, -1l, false);
58+
long fetchRangeEnd = readLong(FETCH_RANGE_END, root, -1l, false);
59+
Metadata metadata = readMetadata(root);
60+
JsonNode parseContextNode = root.get(PARSE_CONTEXT);
61+
ParseContext parseContext = parseContextNode == null ? new ParseContext() : ParseContextDeserializer.readParseContext(parseContextNode);
62+
FetchEmitTuple.ON_PARSE_EXCEPTION onParseException = readOnParseException(root);
63+
64+
return new FetchEmitTuple(id, new FetchKey(fetcherName, fetchKey, fetchRangeStart, fetchRangeEnd),
65+
new EmitKey(emitterName, emitKey), metadata, parseContext,
66+
onParseException);
67+
}
68+
69+
private static FetchEmitTuple.ON_PARSE_EXCEPTION readOnParseException(JsonNode root) throws IOException {
70+
JsonNode onParseExNode = root.get(ON_PARSE_EXCEPTION);
71+
if (onParseExNode == null) {
72+
return FetchEmitTuple.ON_PARSE_EXCEPTION.EMIT;
73+
}
74+
String txt = onParseExNode.asText();
75+
if ("skip".equalsIgnoreCase(txt)) {
76+
return FetchEmitTuple.ON_PARSE_EXCEPTION.SKIP;
77+
} else if ("emit".equalsIgnoreCase(txt)) {
78+
return FetchEmitTuple.ON_PARSE_EXCEPTION.EMIT;
79+
} else {
80+
throw new IOException(ON_PARSE_EXCEPTION + " must be either 'skip' or 'emit'");
81+
}
82+
}
83+
84+
private static Metadata readMetadata(JsonNode root) {
85+
JsonNode metadataNode = root.get(METADATA_KEY);
86+
if (metadataNode == null) {
87+
return new Metadata();
88+
}
89+
Metadata metadata = new Metadata();
90+
for (Map.Entry<String, JsonNode> e : metadataNode.properties()) {
91+
JsonNode vals = e.getValue();
92+
String k = e.getKey();
93+
if (vals.isArray()) {
94+
for (JsonNode arrVal : vals) {
95+
metadata.add(k, arrVal.textValue());
96+
}
97+
} else {
98+
metadata.set(k, vals.asText());
99+
}
100+
}
101+
return metadata;
102+
}
103+
104+
private static String readVal(String key, JsonNode jsonObj, String defaultRet, boolean isRequired) throws IOException {
105+
JsonNode valNode = jsonObj.get(key);
106+
if (valNode == null) {
107+
if (isRequired) {
108+
throw new IOException("required value string, but see: " + key);
109+
}
110+
return defaultRet;
111+
}
112+
return valNode.asText();
113+
}
114+
115+
private static long readLong(String key, JsonNode jsonObj, long defaultVal, boolean isRequired) throws IOException {
116+
JsonNode val = jsonObj.get(key);
117+
if (val == null) {
118+
if (isRequired) {
119+
throw new IOException("required value long, but see: " + key);
120+
}
121+
return defaultVal;
122+
}
123+
return val.longValue();
124+
}
125+
126+
}

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

Lines changed: 65 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,65 @@
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.serialization;
18+
19+
import static org.apache.tika.serialization.ParseContextSerializer.PARSE_CONTEXT;
20+
21+
import java.io.IOException;
22+
import java.util.Locale;
23+
24+
import com.fasterxml.jackson.core.JsonGenerator;
25+
import com.fasterxml.jackson.databind.JsonSerializer;
26+
import com.fasterxml.jackson.databind.SerializerProvider;
27+
28+
import org.apache.tika.pipes.core.FetchEmitTuple;
29+
import org.apache.tika.utils.StringUtils;
30+
31+
public class FetchEmitTupleSerializer extends JsonSerializer<FetchEmitTuple> {
32+
public static final String ID = "id";
33+
public static final String FETCHER = "fetcher";
34+
public static final String FETCH_KEY = "fetchKey";
35+
public static final String FETCH_RANGE_START = "fetchRangeStart";
36+
public static final String FETCH_RANGE_END = "fetchRangeEnd";
37+
public static final String EMITTER = "emitter";
38+
public static final String EMIT_KEY = "emitKey";
39+
public static final String METADATA_KEY = "metadata";
40+
public static final String ON_PARSE_EXCEPTION = "onParseException";
41+
42+
public void serialize(FetchEmitTuple t, JsonGenerator jsonGenerator, SerializerProvider serializerProvider) throws IOException {
43+
44+
jsonGenerator.writeStartObject();
45+
jsonGenerator.writeStringField(ID, t.getId());
46+
jsonGenerator.writeStringField(FETCHER, t.getFetchKey().getFetcherName());
47+
jsonGenerator.writeStringField(FETCH_KEY, t.getFetchKey().getFetchKey());
48+
if (t.getFetchKey().hasRange()) {
49+
jsonGenerator.writeNumberField(FETCH_RANGE_START, t.getFetchKey().getRangeStart());
50+
jsonGenerator.writeNumberField(FETCH_RANGE_END, t.getFetchKey().getRangeEnd());
51+
}
52+
jsonGenerator.writeStringField(EMITTER, t.getEmitKey().getEmitterName());
53+
if (!StringUtils.isBlank(t.getEmitKey().getEmitKey())) {
54+
jsonGenerator.writeStringField(EMIT_KEY, t.getEmitKey().getEmitKey());
55+
}
56+
if (t.getMetadata().size() > 0) {
57+
jsonGenerator.writeObjectField(METADATA_KEY, t.getMetadata());
58+
}
59+
jsonGenerator.writeStringField(ON_PARSE_EXCEPTION, t.getOnParseException().name().toLowerCase(Locale.US));
60+
if (!t.getParseContext().isEmpty()) {
61+
jsonGenerator.writeObjectField(PARSE_CONTEXT, t.getParseContext());
62+
}
63+
jsonGenerator.writeEndObject();
64+
}
65+
}

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

Lines changed: 16 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -19,41 +19,28 @@
1919
import java.io.IOException;
2020
import java.io.Writer;
2121

22-
import com.fasterxml.jackson.core.JsonFactory;
23-
import com.fasterxml.jackson.core.JsonGenerator;
24-
import com.fasterxml.jackson.core.StreamReadConstraints;
22+
import com.fasterxml.jackson.databind.ObjectMapper;
23+
import com.fasterxml.jackson.databind.module.SimpleModule;
2524

26-
import org.apache.tika.config.TikaConfig;
2725
import org.apache.tika.metadata.Metadata;
26+
import org.apache.tika.parser.ParseContext;
27+
import org.apache.tika.pipes.core.FetchEmitTuple;
2828
import org.apache.tika.pipes.core.emitter.EmitData;
29-
import org.apache.tika.pipes.core.emitter.EmitKey;
30-
import org.apache.tika.serialization.JsonMetadata;
29+
import org.apache.tika.serialization.MetadataSerializer;
30+
import org.apache.tika.serialization.ParseContextSerializer;
3131

3232
public class JsonEmitData {
33+
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
34+
35+
static {
36+
SimpleModule module = new SimpleModule();
37+
module.addSerializer(FetchEmitTuple.class, new FetchEmitTupleSerializer());
38+
module.addSerializer(ParseContext.class, new ParseContextSerializer());
39+
module.addSerializer(Metadata.class, new MetadataSerializer());
40+
OBJECT_MAPPER.registerModule(module);
41+
}
3342

3443
public static void toJson(EmitData emitData, Writer writer) throws IOException {
35-
try (JsonGenerator jsonGenerator = new JsonFactory()
36-
.setStreamReadConstraints(StreamReadConstraints
37-
.builder()
38-
.maxStringLength(TikaConfig.getMaxJsonStringFieldLength())
39-
.build())
40-
.createGenerator(writer)) {
41-
jsonGenerator.writeStartObject();
42-
EmitKey key = emitData.getEmitKey();
43-
jsonGenerator.writeStringField(JsonFetchEmitTuple.EMITTER, key.getEmitterName());
44-
jsonGenerator.writeStringField(JsonFetchEmitTuple.EMITKEY, key.getEmitKey());
45-
if (!emitData
46-
.getParseContext()
47-
.isEmpty()) {
48-
jsonGenerator.writeObject(emitData.getParseContext());
49-
}
50-
jsonGenerator.writeFieldName("data");
51-
jsonGenerator.writeStartArray();
52-
for (Metadata m : emitData.getMetadataList()) {
53-
JsonMetadata.writeMetadataObject(m, jsonGenerator, false);
54-
}
55-
jsonGenerator.writeEndArray();
56-
jsonGenerator.writeEndObject();
57-
}
44+
OBJECT_MAPPER.writeValue(writer, emitData);
5845
}
5946
}

0 commit comments

Comments
 (0)