private static JsonNode convertToJsonNode(RelDataType rowType, List<? extends List<RexLiteral>> tuples) throws IOException {
  TokenBuffer out = new TokenBuffer(MAPPER.getFactory().getCodec(), false);
  JsonOutput json = new ExtendedJsonOutput(out);
  json.writeStartArray();
  String[] fields = rowType.getFieldNames().toArray(new String[rowType.getFieldCount()]);

  for (List<RexLiteral> row : tuples) {
    json.writeStartObject();
    int i = 0;
    for (RexLiteral field : row) {
      json.writeFieldName(fields[i]);
      writeLiteral(field, json);
      i++;
    }
    json.writeEndObject();
  }
  json.writeEndArray();
  json.flush();
  return out.asParser().readValueAsTree();
}
 

@Override
public Flux<Object> decode(Publisher<DataBuffer> input, ResolvableType elementType,
		@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {

	Flux<TokenBuffer> tokens = Jackson2Tokenizer.tokenize(
			Flux.from(input), this.jsonFactory, getObjectMapper(), true);

	ObjectReader reader = getObjectReader(elementType, hints);

	return tokens.handle((tokenBuffer, sink) -> {
		try {
			Object value = reader.readValue(tokenBuffer.asParser(getObjectMapper()));
			logValue(value, hints);
			if (value != null) {
				sink.next(value);
			}
		}
		catch (IOException ex) {
			sink.error(processException(ex));
		}
	});
}
 

private List<TokenBuffer> parseTokenBufferFlux() throws IOException {
	List<TokenBuffer> result = new ArrayList<>();

	while (true) {
		JsonToken token = this.parser.nextToken();
		// SPR-16151: Smile data format uses null to separate documents
		if (token == JsonToken.NOT_AVAILABLE ||
				(token == null && (token = this.parser.nextToken()) == null)) {
			break;
		}
		updateDepth(token);
		if (!this.tokenizeArrayElements) {
			processTokenNormal(token, result);
		}
		else {
			processTokenArray(token, result);
		}
	}
	return result;
}
 

/**
 * Tokenize the given {@code Flux<DataBuffer>} into {@code Flux<TokenBuffer>}.
 * @param dataBuffers the source data buffers
 * @param jsonFactory the factory to use
 * @param objectMapper the current mapper instance
 * @param tokenizeArrayElements if {@code true} and the "top level" JSON object is
 * an array, each element is returned individually immediately after it is received
 * @return the resulting token buffers
 */
public static Flux<TokenBuffer> tokenize(Flux<DataBuffer> dataBuffers, JsonFactory jsonFactory,
		ObjectMapper objectMapper, boolean tokenizeArrayElements) {

	try {
		JsonParser parser = jsonFactory.createNonBlockingByteArrayParser();
		DeserializationContext context = objectMapper.getDeserializationContext();
		if (context instanceof DefaultDeserializationContext) {
			context = ((DefaultDeserializationContext) context).createInstance(
					objectMapper.getDeserializationConfig(), parser, objectMapper.getInjectableValues());
		}
		Jackson2Tokenizer tokenizer = new Jackson2Tokenizer(parser, context, tokenizeArrayElements);
		return dataBuffers.concatMapIterable(tokenizer::tokenize).concatWith(tokenizer.endOfInput());
	}
	catch (IOException ex) {
		return Flux.error(ex);
	}
}
 

private void testTokenize(List<String> source, List<String> expected, boolean tokenizeArrayElements) {
	Flux<TokenBuffer> tokens = Jackson2Tokenizer.tokenize(
			Flux.fromIterable(source).map(this::stringBuffer),
			this.jsonFactory, this.objectMapper, tokenizeArrayElements);

	Flux<String> result = tokens
			.map(tokenBuffer -> {
				try {
					TreeNode root = this.objectMapper.readTree(tokenBuffer.asParser());
					return this.objectMapper.writeValueAsString(root);
				}
				catch (IOException ex) {
					throw new UncheckedIOException(ex);
				}
			});

	StepVerifier.FirstStep<String> builder = StepVerifier.create(result);
	expected.forEach(s -> builder.assertNext(new JSONAssertConsumer(s)));
	builder.verifyComplete();
}
 

private Flux<TokenBuffer> parseTokenBufferFlux() throws IOException {
	List<TokenBuffer> result = new ArrayList<>();

	while (true) {
		JsonToken token = this.parser.nextToken();
		// SPR-16151: Smile data format uses null to separate documents
		if ((token == JsonToken.NOT_AVAILABLE) ||
				(token == null && (token = this.parser.nextToken()) == null)) {
			break;
		}
		updateDepth(token);

		if (!this.tokenizeArrayElements) {
			processTokenNormal(token, result);
		}
		else {
			processTokenArray(token, result);
		}
	}
	return Flux.fromIterable(result);
}
 

private void testTokenize(List<String> source, List<String> expected, boolean tokenizeArrayElements) {

		Flux<TokenBuffer> tokenBufferFlux = Jackson2Tokenizer.tokenize(
				Flux.fromIterable(source).map(this::stringBuffer),
				this.jsonFactory,
				tokenizeArrayElements);

		Flux<String> result = tokenBufferFlux
				.map(tokenBuffer -> {
					try {
						TreeNode root = this.objectMapper.readTree(tokenBuffer.asParser());
						return this.objectMapper.writeValueAsString(root);
					}
					catch (IOException ex) {
						throw new UncheckedIOException(ex);
					}
				});

		StepVerifier.FirstStep<String> builder = StepVerifier.create(result);
		expected.forEach(s -> builder.assertNext(new JSONAssertConsumer(s)));
		builder.verifyComplete();
	}
 

/**
 * Read a string from the current location in parser.
 *
 * <p>This method differs from the original JsonDecoder by serializing all
 * structures captured by the current token into a JSON string. This enables
 * consistent behavior for handling variant types (e.g. field that can be a
 * boolean and a string) and for under-specified schemas.
 *
 * <p>This encoding is lossy because JSON strings are conflated with standard
 * strings. Consider the case where a number is decoded into a string. To
 * convert this Avro file back into the original JSON document, the encoder
 * must parse all strings as JSON and inline them into the tree. Now, if the
 * original JSON represents a JSON object as a string (e.g. `{"payload":
 * "{\"foo\":\"bar\"}"`), then the encoder will generate a new object that is
 * different from the original.
 *
 * <p>There are a few ways to avoid this if it is undesirable. One way is to use
 * a binary encoding for the JSON data such as BSON or base64. A second is to
 * normalize documents to avoid nested JSON encodings and to specify a schema
 * explictly to guide the proper typing.
 */
@Override
public String readString() throws IOException {
  parser.advance(Symbol.STRING);
  if (parser.topSymbol() == Symbol.MAP_KEY_MARKER) {
    parser.advance(Symbol.MAP_KEY_MARKER);
    assertCurrentToken(JsonToken.FIELD_NAME, "map-key");
  }

  String result = null;
  if (in.getCurrentToken() == JsonToken.VALUE_STRING
      || in.getCurrentToken() == JsonToken.FIELD_NAME) {
    result = in.getValueAsString();
  } else {
    // Does this create excessive garbage collection?
    TokenBuffer buffer = new TokenBuffer(in);
    buffer.copyCurrentStructure(in);
    result = mapper.readTree(buffer.asParser()).toString();
    buffer.close();
  }
  in.nextToken();
  return result;
}
 

/**
 * Helper method called for rare case of pointing to {@link JsonToken#VALUE_NULL}
 * token. While this is most often an erroneous condition, there is one specific
 * case with XML handling where polymorphic type with no properties is exposed
 * as such, and should be handled same as empty Object.
 *
 * @since 2.7
 */
protected Object deserializeFromNull(JsonParser p, DeserializationContext ctxt)
    throws IOException
{
    // 17-Dec-2015, tatu: Highly specialized case, mainly to support polymorphic
    //   "empty" POJOs deserialized from XML, where empty XML tag synthesizes a
    //   `VALUE_NULL` token.
    if (p.requiresCustomCodec()) { // not only XML module, but mostly it...
        @SuppressWarnings("resource")
        TokenBuffer tb = new TokenBuffer(p, ctxt);
        tb.writeEndObject();
        JsonParser p2 = tb.asParser(p);
        p2.nextToken(); // to point to END_OBJECT
        // note: don't have ObjectId to consider at this point, so:
        Object ob = _vanillaProcessing ? vanillaDeserialize(p2, ctxt, JsonToken.END_OBJECT)
                : deserializeFromObject(p2, ctxt);
        p2.close();
        return ob;
    }
    return ctxt.handleUnexpectedToken(handledType(), p);
}
 

@SuppressWarnings("resource")
@Override
public final Object deserializeKey(String key, DeserializationContext ctxt)
    throws IOException
{
    if (key == null) { // is this even legal call?
        return null;
    }
    TokenBuffer tb = new TokenBuffer(ctxt.getParser(), ctxt);
    tb.writeString(key);
    try {
        // Ugh... should not have to give parser which may or may not be correct one...
        JsonParser p = tb.asParser();
        p.nextToken();
        Object result = _delegate.deserialize(p, ctxt);
        if (result != null) {
            return result;
        }
        return ctxt.handleWeirdKey(_keyClass, key, "not a valid representation");
    } catch (Exception re) {
        return ctxt.handleWeirdKey(_keyClass, key, "not a valid representation: %s", re.getMessage());
    }
}
 

@SuppressWarnings("resource")
protected final Object _deserialize(JsonParser p, DeserializationContext ctxt,
        int index, String typeId) throws IOException
{
    JsonParser p2 = _tokens[index].asParser(p);
    JsonToken t = p2.nextToken();
    // 29-Sep-2015, tatu: As per [databind#942], nulls need special support
    if (t == JsonToken.VALUE_NULL) {
        return null;
    }
    TokenBuffer merged = new TokenBuffer(p, ctxt);
    merged.writeStartArray();
    merged.writeString(typeId);
    merged.copyCurrentStructure(p2);
    merged.writeEndArray();

    // needs to point to START_OBJECT (or whatever first token is)
    JsonParser mp = merged.asParser(p);
    mp.nextToken();
    return _properties[index].getProperty().deserialize(mp, ctxt);
}
 

@SuppressWarnings("resource")
protected final void _deserializeAndSet(JsonParser p, DeserializationContext ctxt,
        Object bean, int index, String typeId) throws IOException
{
    /* Ok: time to mix type id, value; and we will actually use "wrapper-array"
     * style to ensure we can handle all kinds of JSON constructs.
     */
    JsonParser p2 = _tokens[index].asParser(p);
    JsonToken t = p2.nextToken();
    // 29-Sep-2015, tatu: As per [databind#942], nulls need special support
    if (t == JsonToken.VALUE_NULL) {
        _properties[index].getProperty().set(bean, null);
        return;
    }
    TokenBuffer merged = new TokenBuffer(p, ctxt);
    merged.writeStartArray();
    merged.writeString(typeId);

    merged.copyCurrentStructure(p2);
    merged.writeEndArray();
    // needs to point to START_OBJECT (or whatever first token is)
    JsonParser mp = merged.asParser(p);
    mp.nextToken();
    _properties[index].getProperty().deserializeAndSet(mp, ctxt, bean);
}
 

@SuppressWarnings({ "unchecked", "resource" })
public <T extends JsonNode> T valueToTree(Object fromValue)
    throws IllegalArgumentException
{
    if (fromValue == null) return null;
    TokenBuffer buf = new TokenBuffer(this, false);
    if (isEnabled(DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS)) {
        buf = buf.forceUseOfBigDecimal(true);
    }
    JsonNode result;
    try {
        writeValue(buf, fromValue);
        JsonParser p = buf.asParser();
        result = readTree(p);
        p.close();
    } catch (IOException e) { // should not occur, no real i/o...
        throw new IllegalArgumentException(e.getMessage(), e);
    }
    return (T) result;
}
 

private static JsonNode convertToJsonNode(RelDataType rowType, ImmutableList<ImmutableList<RexLiteral>> tuples) throws IOException{
  TokenBuffer out = new TokenBuffer(MAPPER.getFactory().getCodec(), false);
  JsonOutput json = new ExtendedJsonOutput(out);
  json.writeStartArray();
  String[] fields = rowType.getFieldNames().toArray(new String[rowType.getFieldCount()]);

  for(List<RexLiteral> row : tuples){
    json.writeStartObject();
    int i =0;
    for(RexLiteral field : row){
      json.writeFieldName(fields[i]);
      writeLiteral(field, json);
      i++;
    }
    json.writeEndObject();
  }
  json.writeEndArray();
  json.flush();
  return out.asParser().readValueAsTree();
}
 

private JsonNode getResultsNode(long count) {
  try{
    TokenBuffer out = new TokenBuffer(MAPPER.getFactory().getCodec(), false);
    JsonOutput json = new ExtendedJsonOutput(out);
    json.writeStartArray();
    json.writeStartObject();
    json.writeFieldName("count");
    json.writeBigInt(count);
    json.writeEndObject();
    json.writeEndArray();
    json.flush();
    return out.asParser().readValueAsTree();
  }catch(IOException ex){
    throw Throwables.propagate(ex);
  }
}
 

@Override
public Object deepCopy( Object value ) throws HibernateException
{
    if ( value == null )
    {
        return null;
    }

    final TokenBuffer tb = new TokenBuffer( resultingMapper, false );
    try
    {
        writeValue( tb, value );
        return readValue( tb.asParser() );
    }
    catch ( IOException e )
    {
        throw new HibernateException( "Could not deep copy JSONB object.", e );
    }
}
 

@Override
public <T> T unmarshall(final Object marshalled, final Class<T> type) throws Exception {
  checkNotNull(marshalled);
  checkState(marshalled instanceof Map, "Marshalled data must be a Map; found: %s", marshalled.getClass());

  // FIXME: This allows the top-level object to be created, but if any children objects of this are missing
  // FIXME: ... no-arg CTOR then Jackson will fail to construct them.
  // FIXME: Is there any way to configure the basic instance creation for Jackson?
  Object value = instanceCreator.newInstance(type);

  // performs same basic logic as ObjectMapper.convertValue(Object, Class) helper
  ObjectReader reader = objectMapper.readerForUpdating(value);
  TokenBuffer buff = new TokenBuffer(objectMapper, false);
  objectMapper.writeValue(buff, marshalled);
  reader.readValue(buff.asParser());

  return type.cast(value);
}
 

private Jackson2Tokenizer(
		JsonParser parser, DeserializationContext deserializationContext, boolean tokenizeArrayElements) {

	this.parser = parser;
	this.deserializationContext = deserializationContext;
	this.tokenizeArrayElements = tokenizeArrayElements;
	this.tokenBuffer = new TokenBuffer(parser, deserializationContext);
	this.inputFeeder = (ByteArrayFeeder) this.parser.getNonBlockingInputFeeder();
}
 

private void processTokenNormal(JsonToken token, List<TokenBuffer> result) throws IOException {
	this.tokenBuffer.copyCurrentEvent(this.parser);

	if ((token.isStructEnd() || token.isScalarValue()) && this.objectDepth == 0 && this.arrayDepth == 0) {
		result.add(this.tokenBuffer);
		this.tokenBuffer = new TokenBuffer(this.parser, this.deserializationContext);
	}

}
 

private void processTokenArray(JsonToken token, List<TokenBuffer> result) throws IOException {
	if (!isTopLevelArrayToken(token)) {
		this.tokenBuffer.copyCurrentEvent(this.parser);
	}

	if (this.objectDepth == 0 && (this.arrayDepth == 0 || this.arrayDepth == 1) &&
			(token == JsonToken.END_OBJECT || token.isScalarValue())) {
		result.add(this.tokenBuffer);
		this.tokenBuffer = new TokenBuffer(this.parser, this.deserializationContext);
	}
}
 

@Test
public void errorInStream() {
	DataBuffer buffer = stringBuffer("{\"id\":1,\"name\":");
	Flux<DataBuffer> source = Flux.just(buffer).concatWith(Flux.error(new RuntimeException()));
	Flux<TokenBuffer> result = Jackson2Tokenizer.tokenize(source, this.jsonFactory, this.objectMapper, true);

	StepVerifier.create(result)
			.expectError(RuntimeException.class)
			.verify();
}
 

@Test  // SPR-16521
public void jsonEOFExceptionIsWrappedAsDecodingError() {
	Flux<DataBuffer> source = Flux.just(stringBuffer("{\"status\": \"noClosingQuote}"));
	Flux<TokenBuffer> tokens = Jackson2Tokenizer.tokenize(source, this.jsonFactory, this.objectMapper, false);

	StepVerifier.create(tokens)
			.expectError(DecodingException.class)
			.verify();
}
 

@Override
public Flux<Object> decode(Publisher<DataBuffer> input, ResolvableType elementType,
		@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {

	Flux<TokenBuffer> tokens = Jackson2Tokenizer.tokenize(Flux.from(input), this.jsonFactory, true);
	return decodeInternal(tokens, elementType, mimeType, hints);
}
 

@Override
public Mono<Object> decodeToMono(Publisher<DataBuffer> input, ResolvableType elementType,
		@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {

	Flux<TokenBuffer> tokens = Jackson2Tokenizer.tokenize(Flux.from(input), this.jsonFactory, false);
	return decodeInternal(tokens, elementType, mimeType, hints).singleOrEmpty();
}
 

private Jackson2Tokenizer(JsonParser parser, boolean tokenizeArrayElements) {
	Assert.notNull(parser, "'parser' must not be null");

	this.parser = parser;
	this.tokenizeArrayElements = tokenizeArrayElements;
	this.tokenBuffer = new TokenBuffer(parser);
	this.inputFeeder = (ByteArrayFeeder) this.parser.getNonBlockingInputFeeder();
}
 

/**
 * Tokenize the given {@code Flux<DataBuffer>} into {@code Flux<TokenBuffer>}.
 * @param dataBuffers the source data buffers
 * @param jsonFactory the factory to use
 * @param tokenizeArrayElements if {@code true} and the "top level" JSON
 * object is an array, each element is returned individually, immediately
 * after it is received.
 * @return the result token buffers
 */
public static Flux<TokenBuffer> tokenize(Flux<DataBuffer> dataBuffers, JsonFactory jsonFactory,
		boolean tokenizeArrayElements) {

	try {
		JsonParser parser = jsonFactory.createNonBlockingByteArrayParser();
		Jackson2Tokenizer tokenizer = new Jackson2Tokenizer(parser, tokenizeArrayElements);
		return dataBuffers.flatMap(tokenizer::tokenize, Flux::error, tokenizer::endOfInput);
	}
	catch (IOException ex) {
		return Flux.error(ex);
	}
}
 

private void processTokenNormal(JsonToken token, List<TokenBuffer> result) throws IOException {
	this.tokenBuffer.copyCurrentEvent(this.parser);

	if ((token.isStructEnd() || token.isScalarValue()) &&
			this.objectDepth == 0 && this.arrayDepth == 0) {
		result.add(this.tokenBuffer);
		this.tokenBuffer = new TokenBuffer(this.parser);
	}

}
 

private void processTokenArray(JsonToken token, List<TokenBuffer> result) throws IOException {
	if (!isTopLevelArrayToken(token)) {
		this.tokenBuffer.copyCurrentEvent(this.parser);
	}

	if (this.objectDepth == 0 &&
			(this.arrayDepth == 0 || this.arrayDepth == 1) &&
			(token == JsonToken.END_OBJECT || token.isScalarValue())) {
		result.add(this.tokenBuffer);
		this.tokenBuffer = new TokenBuffer(this.parser);
	}
}
 

@Test
public void errorInStream() {
	DataBuffer buffer = stringBuffer("{\"id\":1,\"name\":");
	Flux<DataBuffer> source = Flux.just(buffer)
			.concatWith(Flux.error(new RuntimeException()));

	Flux<TokenBuffer> result = Jackson2Tokenizer.tokenize(source, this.jsonFactory, true);

	StepVerifier.create(result)
			.expectError(RuntimeException.class)
			.verify();
}
 

@Test // SPR-16521
public void jsonEOFExceptionIsWrappedAsDecodingError() {
	Flux<DataBuffer> source = Flux.just(stringBuffer("{\"status\": \"noClosingQuote}"));
	Flux<TokenBuffer> tokens = Jackson2Tokenizer.tokenize(source, this.jsonFactory, false);

	StepVerifier.create(tokens)
			.expectError(DecodingException.class)
			.verify();
}