Streaming data ... the client side

Here comes the counterpart to the “Jdbi and Streaming” post. This post will cover the client side of things. We will use Jackson’s JsonParser and ObjectMapper to process the streamed data chunk by chunk … or rather JSON object by JSON object :)

This implementation doesn’t rely on any specific web framework as it just operates on an InputStream. And as it uses just an InputStream it doesn’t even necessarily has to be the response of a web server but could also be a FileInputStream for a file on disk.

Code

Most of the code is from the StackOverflow question “How to parse a JSON input stream?”.

public class JsonObjectIterator<T> implements Iterator<T>, Closeable {

    private final InputStream inputStream;
    private JsonParser jsonParser;
    private boolean isInitialized;

    private T nextObject;
    private ObjectMapper objectMapper;
    private Class<T> clazz;

    public JsonObjectIterator(final InputStream inputStream, Class<T> clazz) {
        this(inputStream, clazz, new ObjectMapper());
    }

    public JsonObjectIterator(final InputStream inputStream, Class<T> clazz, ObjectMapper objectMapper) {
        this.inputStream = inputStream;
        this.isInitialized = false;
        this.nextObject = null;
        this.objectMapper = objectMapper;
        this.clazz = clazz;
    }

    private void init() {
        this.initJsonParser();
        this.initFirstElement();
        this.isInitialized = true;
    }

    private void initJsonParser() {
        final JsonFactory jsonFactory = objectMapper.getFactory();

        try {
            this.jsonParser = jsonFactory.createParser(inputStream);
        } catch (final IOException e) {
            throw new RuntimeException("There was a problem setting up the JsonParser: " + e.getMessage(), e);
        }
    }

    private void initFirstElement() {
        try {
            // Check that the first element is the start of an array
            final JsonToken arrayStartToken = this.jsonParser.nextToken();
            if (arrayStartToken != JsonToken.START_ARRAY) {
                throw new IllegalStateException(
                        "The first element of the Json structure was expected to be a start array token, but it was: " + arrayStartToken);
            }

            // Initialize the first object
            this.initNextObject();
        } catch (final Exception e) {
            throw new RuntimeException(
                    "There was a problem initializing the first element of the Json Structure: " + e.getMessage(), e);
        }

    }

    private void initNextObject() {
        try {
            final JsonToken nextToken = this.jsonParser.nextToken();

            // Check for the end of the array which will mean we're done
            if (nextToken == JsonToken.END_ARRAY) {
                this.nextObject = null;
                return;
            }

            // Make sure the next token is the start of an object
            if (nextToken != JsonToken.START_OBJECT) {
                throw new IllegalStateException(
                        "The next token of Json structure was expected to be a start object token, but it was: "
                                + nextToken);
            }

            // Get the next object and make sure it's not null
            this.nextObject = this.jsonParser.readValueAs(clazz);
            if (this.nextObject == null) {
                throw new IllegalStateException("The next parsed object of the Json structure was null");
            }
        } catch (final Exception e) {
            throw new RuntimeException("There was a problem initializing the next Object: " + e.getMessage(), e);
        }
    }

    @Override
    public boolean hasNext() {
        if (!this.isInitialized) { this.init(); }

        return this.nextObject != null;
    }

    @Override
    public T next() {
        // This method will return the current object and initialize the next object so
        // hasNext will always have knowledge of the current state

        // Makes sure we're initialized first
        if (!this.isInitialized) { this.init(); }

        // Store the current next object for return
        final T currentNextObject = this.nextObject;

        // Initialize the next object
        this.initNextObject();

        return currentNextObject;
    }

    @Override
    public void close() throws IOException {
        try {
            this.jsonParser.close();
        } catch (Exception e) {
            // close quietly
        }

        try {
            this.inputStream.close();
        } catch (Exception e) {
            // close quietly
        }
    }

}        

So you just need to get the InputStream (from the HTTP client of your choice) and pass it to the JsonObjectIterator. As we want to iterator over a number of JSON objects received from the web service the JsonObjectIterator expects a JSON array as the top level entity in the InputStream.

All in all … no magic to be found here ;-)

Wrap Up

It is pretty easy to process a big JSON array chunk by chunk with the help of the Jackson library. I skipped the part on how to use the JsonObjectIterator as it is just an iterator and I am sure you already know how to code a while loop with calling hasNext and next.

Happy streaming!

Mihael