Add a streaming Json item reader

This commit adds a new Json item reader with two implementations based
on Jackson and Gson.

Resolves BATCH-2691

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/json/GsonJsonObjectReader.java

@@ -0,0 +1,97 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed 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.springframework.batch.item.json;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+
+import com.google.gson.Gson;
+import com.google.gson.JsonIOException;
+import com.google.gson.JsonSyntaxException;
+import com.google.gson.stream.JsonReader;
+import com.google.gson.stream.JsonToken;
+
+import org.springframework.batch.item.ParseException;
+import org.springframework.core.io.Resource;
+import org.springframework.util.Assert;
+
+/**
+ * Implementation of {@link JsonObjectReader} based on
+ * <a href="https://github.com/google/gson">Google Gson</a>.
+ *
+ * @param <T> type of the target object
+ *
+ * @author Mahmoud Ben Hassine
+ * @since 4.1
+ */
+public class GsonJsonObjectReader<T> implements JsonObjectReader<T> {
+
+	private Class<? extends T> itemType;
+
+	private JsonReader jsonReader;
+
+	private Gson mapper = new Gson();
+
+	private InputStream inputStream;
+
+	/**
+	 * Create a new {@link GsonJsonObjectReader} instance.
+	 * @param itemType the target item type
+	 */
+	public GsonJsonObjectReader(Class<? extends T> itemType) {
+		this.itemType = itemType;
+	}
+
+	/**
+	 * Set the object mapper to use to map Json objects to domain objects.
+	 * @param mapper the object mapper to use
+	 */
+	public void setMapper(Gson mapper) {
+		Assert.notNull(mapper, "The mapper must not be null");
+		this.mapper = mapper;
+	}
+
+	@Override
+	public void open(Resource resource) throws Exception {
+		Assert.notNull(resource, "The resource must not be null");
+		this.inputStream = resource.getInputStream();
+		this.jsonReader = this.mapper.newJsonReader(new InputStreamReader(this.inputStream));
+		Assert.state(this.jsonReader.peek() == JsonToken.BEGIN_ARRAY,
+				"The Json input stream must start with an array of Json objects");
+		this.jsonReader.beginArray();
+	}
+
+	@Override
+	public T read() throws Exception {
+		try {
+			if (this.jsonReader.hasNext()) {
+				return this.mapper.fromJson(this.jsonReader, this.itemType);
+			}
+		} catch (IOException |JsonIOException | JsonSyntaxException e) {
+			throw new ParseException("Unable to read next JSON object", e);
+		}
+		return null;
+	}
+
+	@Override
+	public void close() throws Exception {
+		this.inputStream.close();
+		this.jsonReader.close();
+	}
+
+}

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/json/JacksonJsonObjectReader.java

@@ -0,0 +1,93 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed 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.springframework.batch.item.json;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.core.JsonToken;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import org.springframework.batch.item.ParseException;
+import org.springframework.core.io.Resource;
+import org.springframework.util.Assert;
+
+/**
+ * Implementation of {@link JsonObjectReader} based on
+ * <a href="https://github.com/FasterXML/jackson">Jackson</a>.
+ *
+ * @param <T> type of the target object
+ *
+ * @author Mahmoud Ben Hassine
+ * @since 4.1
+ */
+public class JacksonJsonObjectReader<T> implements JsonObjectReader<T> {
+
+	private Class<? extends T> itemType;
+
+	private JsonParser jsonParser;
+
+	private ObjectMapper mapper = new ObjectMapper();
+
+	private InputStream inputStream;
+
+	/**
+	 * Create a new {@link JacksonJsonObjectReader} instance.
+	 * @param itemType the target item type
+	 */
+	public JacksonJsonObjectReader(Class<? extends T> itemType) {
+		this.itemType = itemType;
+	}
+
+	/**
+	 * Set the object mapper to use to map Json objects to domain objects.
+	 * @param mapper the object mapper to use
+	 */
+	public void setMapper(ObjectMapper mapper) {
+		Assert.notNull(mapper, "The mapper must not be null");
+		this.mapper = mapper;
+	}
+
+	@Override
+	public void open(Resource resource) throws Exception {
+		Assert.notNull(resource, "The resource must not be null");
+		this.inputStream = resource.getInputStream();
+		this.jsonParser = this.mapper.getFactory().createParser(this.inputStream);
+		Assert.state(this.jsonParser.nextToken() == JsonToken.START_ARRAY,
+				"The Json input stream must start with an array of Json objects");
+	}
+
+	@Override
+	public T read() throws Exception {
+		try {
+			if (this.jsonParser.nextToken() == JsonToken.START_OBJECT) {
+				return this.mapper.readValue(this.jsonParser, this.itemType);
+			}
+		} catch (IOException e) {
+			throw new ParseException("Unable to read next JSON object", e);
+		}
+		return null;
+	}
+
+	@Override
+	public void close() throws Exception {
+		this.inputStream.close();
+		this.jsonParser.close();
+	}
+
+}

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/json/JsonItemReader.java

@@ -0,0 +1,126 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed 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.springframework.batch.item.json;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.batch.item.ItemStreamReader;
+import org.springframework.batch.item.file.ResourceAwareItemReaderItemStream;
+import org.springframework.batch.item.support.AbstractItemCountingItemStreamItemReader;
+import org.springframework.core.io.Resource;
+import org.springframework.util.Assert;
+
+/**
+ * {@link ItemStreamReader} implementation that reads Json objects from a
+ * {@link Resource} having the following format:
+ * <p>
+ * <code>
+ * [
+ *    {
+ *       // JSON object
+ *    },
+ *    {
+ *       // JSON object
+ *    }
+ * ]
+ * </code>
+ * <p>
+ *
+ * The implementation is <b>not</b> thread-safe.
+ *
+ * @param <T> the type of json objects to read
+ *
+ * @author Mahmoud Ben Hassine
+ * @since 4.1
+ */
+public class JsonItemReader<T> extends AbstractItemCountingItemStreamItemReader<T> implements
+		ResourceAwareItemReaderItemStream<T> {
+
+	private static final Log LOGGER = LogFactory.getLog(JsonItemReader.class);
+
+	private Resource resource;
+
+	private JsonObjectReader<T> jsonObjectReader;
+
+	private boolean strict = true;
+
+	/**
+	 * Create a new {@link JsonItemReader} instance.
+	 * @param resource the input json resource
+	 * @param jsonObjectReader the json object reader to use
+	 */
+	public JsonItemReader(Resource resource, JsonObjectReader<T> jsonObjectReader) {
+		Assert.notNull(resource, "The resource must not be null.");
+		Assert.notNull(jsonObjectReader, "The json object reader must not be null.");
+		this.resource = resource;
+		this.jsonObjectReader = jsonObjectReader;
+	}
+
+	/**
+	 * Set the {@link JsonObjectReader} to use to read and map Json fragments to domain objects.
+	 * @param jsonObjectReader the json object reader to use
+	 */
+	public void setJsonObjectReader(JsonObjectReader<T> jsonObjectReader) {
+		this.jsonObjectReader = jsonObjectReader;
+	}
+
+	/**
+	 * In strict mode the reader will throw an exception on
+	 * {@link #open(org.springframework.batch.item.ExecutionContext)} if the
+	 * input resource does not exist.
+	 * @param strict true by default
+	 */
+	public void setStrict(boolean strict) {
+		this.strict = strict;
+	}
+
+	@Override
+	public void setResource(Resource resource) {
+		this.resource = resource;
+	}
+
+	@Override
+	protected T doRead() throws Exception {
+		return jsonObjectReader.read();
+	}
+
+	@Override
+	protected void doOpen() throws Exception {
+		if (!this.resource.exists()) {
+			if (this.strict) {
+				throw new IllegalStateException("Input resource must exist (reader is in 'strict' mode)");
+			}
+			LOGGER.warn("Input resource does not exist " + this.resource.getDescription());
+			return;
+		}
+		if (!this.resource.isReadable()) {
+			if (this.strict) {
+				throw new IllegalStateException("Input resource must be readable (reader is in 'strict' mode)");
+			}
+			LOGGER.warn("Input resource is not readable " + this.resource.getDescription());
+			return;
+		}
+		this.jsonObjectReader.open(this.resource);
+	}
+
+	@Override
+	protected void doClose() throws Exception {
+		this.jsonObjectReader.close();
+	}
+
+}

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/json/JsonObjectReader.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed 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.springframework.batch.item.json;
+
+import org.springframework.core.io.Resource;
+
+/**
+ * Strategy interface for Json readers. Implementations are expected to use
+ * a streaming API in order to read Json objects one at a time.
+ *
+ * @param <T> type of the target object
+ *
+ * @author Mahmoud Ben Hassine
+ * @since 4.1
+ */
+public interface JsonObjectReader<T> {
+
+	/**
+	 * Open the Json resource for reading.
+	 * @param resource the input resource
+	 * @throws Exception if unable to open the resource
+	 */
+	default void open(Resource resource) throws Exception {
+
+	}
+
+	/**
+	 * Read the next object in the Json resource if any.
+	 * @return the next object or null if the resource is exhausted
+	 * @throws Exception if unable to read the next object
+	 */
+	T read() throws Exception;
+
+	/**
+	 * Close the input resource.
+	 * @throws Exception if unable to close the input resource
+	 */
+	default void close() throws Exception {
+
+	}
+}

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/json/builder/JsonItemReaderBuilder.java

@@ -0,0 +1,157 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed 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.springframework.batch.item.json.builder;
+
+import org.springframework.batch.item.json.JsonItemReader;
+import org.springframework.batch.item.json.JsonObjectReader;
+import org.springframework.core.io.Resource;
+import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
+
+/**
+ * A builder for {@link JsonItemReader}.
+ *
+ * @param <T> type of the target item
+ *
+ * @author Mahmoud Ben Hassine
+ * @since 4.1
+ */
+public class JsonItemReaderBuilder<T> {
+
+	private JsonObjectReader<T> jsonObjectReader;
+
+	private Resource resource;
+
+	private String name;
+
+	private boolean strict = true;
+
+	private boolean saveState = true;
+
+	private int maxItemCount = Integer.MAX_VALUE;
+
+	private int currentItemCount;
+
+	/**
+	 * Set the {@link JsonObjectReader} to use to read and map Json objects to domain objects.
+	 * @param jsonObjectReader to use
+	 * @return The current instance of the builder.
+	 * @see JsonItemReader#setJsonObjectReader(JsonObjectReader)
+	 */
+	public JsonItemReaderBuilder<T> jsonObjectReader(JsonObjectReader<T> jsonObjectReader) {
+		this.jsonObjectReader = jsonObjectReader;
+
+		return this;
+	}
+
+	/**
+	 * The {@link Resource} to be used as input.
+	 * @param resource the input to the reader.
+	 * @return The current instance of the builder.
+	 * @see JsonItemReader#setResource(Resource)
+	 */
+	public JsonItemReaderBuilder<T> resource(Resource resource) {
+		this.resource = resource;
+
+		return this;
+	}
+
+	/**
+	 * The name used to calculate the key within the
+	 * {@link org.springframework.batch.item.ExecutionContext}. Required if
+	 * {@link #saveState(boolean)} is set to true.
+	 * @param name name of the reader instance
+	 * @return The current instance of the builder.
+	 * @see org.springframework.batch.item.ItemStreamSupport#setName(String)
+	 */
+	public JsonItemReaderBuilder<T> name(String name) {
+		this.name = name;
+
+		return this;
+	}
+
+	/**
+	 * Setting this value to true indicates that it is an error if the input
+	 * does not exist and an exception will be thrown. Defaults to true.
+	 * @param strict indicates the input resource must exist
+	 * @return The current instance of the builder.
+	 * @see JsonItemReader#setStrict(boolean)
+	 */
+	public JsonItemReaderBuilder<T> strict(boolean strict) {
+		this.strict = strict;
+
+		return this;
+	}
+
+	/**
+	 * Configure if the state of the {@link org.springframework.batch.item.ItemStreamSupport}
+	 * should be persisted within the {@link org.springframework.batch.item.ExecutionContext}
+	 * for restart purposes.
+	 * @param saveState defaults to true
+	 * @return The current instance of the builder.
+	 */
+	public JsonItemReaderBuilder<T> saveState(boolean saveState) {
+		this.saveState = saveState;
+
+		return this;
+	}
+
+	/**
+	 * Configure the max number of items to be read.
+	 * @param maxItemCount the max items to be read
+	 * @return The current instance of the builder.
+	 * @see org.springframework.batch.item.support.AbstractItemCountingItemStreamItemReader#setMaxItemCount(int)
+	 */
+	public JsonItemReaderBuilder<T> maxItemCount(int maxItemCount) {
+		this.maxItemCount = maxItemCount;
+
+		return this;
+	}
+
+	/**
+	 * Index for the current item. Used on restarts to indicate where to start from.
+	 * @param currentItemCount current index
+	 * @return The current instance of the builder.
+	 * @see org.springframework.batch.item.support.AbstractItemCountingItemStreamItemReader#setCurrentItemCount(int)
+	 */
+	public JsonItemReaderBuilder<T> currentItemCount(int currentItemCount) {
+		this.currentItemCount = currentItemCount;
+
+		return this;
+	}
+
+	/**
+	 * Validate the configuration and build a new {@link JsonItemReader}.
+	 * @return a new instance of the {@link JsonItemReader}
+	 */
+	public JsonItemReader<T> build() {
+		Assert.notNull(this.jsonObjectReader, "A json object reader is required.");
+		Assert.notNull(this.resource, "A resource is required.");
+		if (this.saveState) {
+			Assert.state(StringUtils.hasText(this.name), "A name is required when saveState is set to true.");
+		}
+
+		JsonItemReader<T> reader = new JsonItemReader<>(this.resource, this.jsonObjectReader);
+		reader.setName(this.name);
+		reader.setStrict(this.strict);
+		reader.setSaveState(this.saveState);
+		reader.setMaxItemCount(this.maxItemCount);
+		reader.setCurrentItemCount(this.currentItemCount);
+
+		return reader;
+	}
+}