/*

* Created on 18.06.2015

*/

package com.github.dockerjava.core.async;

import java.io.IOException;

import java.io.InputStream;

import com.fasterxml.jackson.core.JsonFactory;

import com.fasterxml.jackson.core.JsonParser;

import com.fasterxml.jackson.core.JsonToken;

import com.fasterxml.jackson.databind.ObjectMapper;

import com.fasterxml.jackson.databind.node.ObjectNode;

import com.github.dockerjava.api.async.ResultCallback;

/**

*

* @author Marcus Linke

*

*/

public class JsonStreamProcessor<T> implements ResponseStreamProcessor<T> {

private static final JsonFactory JSON_FACTORY = new JsonFactory();

private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();

private final Class<T> clazz;

public JsonStreamProcessor(Class<T> clazz) {

this.clazz = clazz;

}

@Override

public void processResponseStream(InputStream response, ResultCallback<T> resultCallback) {

resultCallback.onStart(response);

OBJECT_MAPPER.configure(JsonParser.Feature.AUTO_CLOSE_SOURCE, true);

try {

JsonParser jp = JSON_FACTORY.createParser(response);

Boolean closed = jp.isClosed();

JsonToken nextToken = jp.nextToken();

while (!closed && nextToken != null && nextToken != JsonToken.END_OBJECT) {

try {

ObjectNode objectNode = OBJECT_MAPPER.readTree(jp);

// exclude empty item serialization into class #461

if (!objectNode.isEmpty(null)) {

T next = OBJECT_MAPPER.treeToValue(objectNode, clazz);

resultCallback.onNext(next);

}

} catch (Exception e) {

resultCallback.onError(e);

}

closed = jp.isClosed();

nextToken = jp.nextToken();

}

} catch (Throwable t) {

resultCallback.onError(t);

} finally {

try {

response.close();

} catch (IOException e) {

resultCallback.onError(e);

} finally {

resultCallback.onComplete();

}

}

}

}