@Test public void correctItems() { Subscription subscription = mock(Subscription.class); Answer<Void> onNext = new Answer<Void>() { public Void answer(InvocationOnMock invocation) throws Throwable { for (JsonToken token: SAMPLE_TOKENS) { processor.onNext(token); } return null; } }; doAnswer(onNext).when(subscription).request(Long.MAX_VALUE); processor.onSubscribe(subscription); Subscriber<String> subscriber = mock(Subscriber.class); VoidAnswer requestAll = invocation -> invocation.<Subscription>getArgument(0).request(Long.MAX_VALUE); doAnswer(requestAll).when(subscriber).onSubscribe(any()); processor.subscribe(subscriber); ArgumentCaptor<String> values = ArgumentCaptor.forClass(String.class); verify(subscriber).onSubscribe(any()); verify(subscriber, times(13)).onNext(values.capture()); assertThat(values.getAllValues()).containsExactly( SAMPLE_TRIMMED_JSON_OBJECT, "{}", "[1,2,3]", "[]", "\"Json String\"", "0", "42", "-11", "3.141", "3e141", "true", "false", "null"); }
@Override public void onNext(JsonToken token) { switch (state) { case EXPECT_ARRAY: if (JsonToken.START_ARRAY.equals(token)) { tokenStack.push(token); state = State.EXPECT_VALUE; request(1); } else { emitError(JsonToken.START_ARRAY, token); } break; case EXPECT_VALUE: if (JsonToken.START_ARRAY.equals(token) || JsonToken.START_OBJECT.equals(token)) { tokenStack.push(token); } else if (JsonToken.END_ARRAY.equals(token) || JsonToken.END_OBJECT.equals(token)) { JsonToken opening = tokenStack.pop(); if (!match(opening, token)) { emitError(JsonToken.END_ARRAY.equals(token) ? JsonToken.END_ARRAY : JsonToken.END_OBJECT, token); } } if (token.toChar() == '"') { buffer.append(token.toChar()); } buffer.append(token.getValue()); if (token.toChar() == '"') { buffer.append(token.toChar()); } if (tokenStack.size() == 1) { String value = buffer.toString(); buffer.setLength(0); state = State.EXPECT_COMMA; publish(value); } else { request(1); } break; case EXPECT_COMMA: if (JsonToken.COMMA.equals(token)) { state = State.EXPECT_VALUE; request(1); } else if (JsonToken.END_ARRAY.equals(token)) { tokenStack.pop(); onComplete(); } else { emitError(JsonToken.COMMA, token); } break; default: onError(new JsonParsingException("Unexpected token " + token.getValue(), null)); } }
JsonArrayProcessor extends AbstractSimpleProcessor<JsonToken, String> { @Override public void onNext(JsonToken token) { switch (state) { case EXPECT_ARRAY: if (JsonToken.START_ARRAY.equals(token)) { tokenStack.push(token); state = State.EXPECT_VALUE; request(1); } else { emitError(JsonToken.START_ARRAY, token); } break; case EXPECT_VALUE: if (JsonToken.START_ARRAY.equals(token) || JsonToken.START_OBJECT.equals(token)) { tokenStack.push(token); } else if (JsonToken.END_ARRAY.equals(token) || JsonToken.END_OBJECT.equals(token)) { JsonToken opening = tokenStack.pop(); if (!match(opening, token)) { emitError(JsonToken.END_ARRAY.equals(token) ? JsonToken.END_ARRAY : JsonToken.END_OBJECT, token); } } if (token.toChar() == '"') { buffer.append(token.toChar()); } buffer.append(token.getValue()); if (token.toChar() == '"') { buffer.append(token.toChar()); } if (tokenStack.size() == 1) { String value = buffer.toString(); buffer.setLength(0); state = State.EXPECT_COMMA; publish(value); } else { request(1); } break; case EXPECT_COMMA: if (JsonToken.COMMA.equals(token)) { state = State.EXPECT_VALUE; request(1); } else if (JsonToken.END_ARRAY.equals(token)) { tokenStack.pop(); onComplete(); } else { emitError(JsonToken.COMMA, token); } break; default: onError(new JsonParsingException("Unexpected token " + token.getValue(), null)); } } }
JsonArrayProcessor extends AbstractSimpleProcessor<JsonToken, String> { @Override public void onNext(JsonToken token) { switch (state) { case EXPECT_ARRAY: if (JsonToken.START_ARRAY.equals(token)) { tokenStack.push(token); state = State.EXPECT_VALUE; request(1); } else { emitError(JsonToken.START_ARRAY, token); } break; case EXPECT_VALUE: if (JsonToken.START_ARRAY.equals(token) || JsonToken.START_OBJECT.equals(token)) { tokenStack.push(token); } else if (JsonToken.END_ARRAY.equals(token) || JsonToken.END_OBJECT.equals(token)) { JsonToken opening = tokenStack.pop(); if (!match(opening, token)) { emitError(JsonToken.END_ARRAY.equals(token) ? JsonToken.END_ARRAY : JsonToken.END_OBJECT, token); } } if (token.toChar() == '"') { buffer.append(token.toChar()); } buffer.append(token.getValue()); if (token.toChar() == '"') { buffer.append(token.toChar()); } if (tokenStack.size() == 1) { String value = buffer.toString(); buffer.setLength(0); state = State.EXPECT_COMMA; publish(value); } else { request(1); } break; case EXPECT_COMMA: if (JsonToken.COMMA.equals(token)) { state = State.EXPECT_VALUE; request(1); } else if (JsonToken.END_ARRAY.equals(token)) { tokenStack.pop(); onComplete(); } else { emitError(JsonToken.COMMA, token); } break; default: onError(new JsonParsingException("Unexpected token " + token.getValue(), null)); } }  }
JsonArrayProcessor extends AbstractSimpleProcessor<JsonToken, String> { @Override public void onNext(JsonToken token) { switch (state) { case EXPECT_ARRAY: if (JsonToken.START_ARRAY.equals(token)) { tokenStack.push(token); state = State.EXPECT_VALUE; request(1); } else { emitError(JsonToken.START_ARRAY, token); } break; case EXPECT_VALUE: if (JsonToken.START_ARRAY.equals(token) || JsonToken.START_OBJECT.equals(token)) { tokenStack.push(token); } else if (JsonToken.END_ARRAY.equals(token) || JsonToken.END_OBJECT.equals(token)) { JsonToken opening = tokenStack.pop(); if (!match(opening, token)) { emitError(JsonToken.END_ARRAY.equals(token) ? JsonToken.END_ARRAY : JsonToken.END_OBJECT, token); } } if (token.toChar() == '"') { buffer.append(token.toChar()); } buffer.append(token.getValue()); if (token.toChar() == '"') { buffer.append(token.toChar()); } if (tokenStack.size() == 1) { String value = buffer.toString(); buffer.setLength(0); state = State.EXPECT_COMMA; publish(value); } else { request(1); } break; case EXPECT_COMMA: if (JsonToken.COMMA.equals(token)) { state = State.EXPECT_VALUE; request(1); } else if (JsonToken.END_ARRAY.equals(token)) { tokenStack.pop(); onComplete(); } else { emitError(JsonToken.COMMA, token); } break; default: onError(new JsonParsingException("Unexpected token " + token.getValue(), null)); } }  @Override void onNext(JsonToken token); }
JsonArrayProcessor extends AbstractSimpleProcessor<JsonToken, String> { @Override public void onNext(JsonToken token) { switch (state) { case EXPECT_ARRAY: if (JsonToken.START_ARRAY.equals(token)) { tokenStack.push(token); state = State.EXPECT_VALUE; request(1); } else { emitError(JsonToken.START_ARRAY, token); } break; case EXPECT_VALUE: if (JsonToken.START_ARRAY.equals(token) || JsonToken.START_OBJECT.equals(token)) { tokenStack.push(token); } else if (JsonToken.END_ARRAY.equals(token) || JsonToken.END_OBJECT.equals(token)) { JsonToken opening = tokenStack.pop(); if (!match(opening, token)) { emitError(JsonToken.END_ARRAY.equals(token) ? JsonToken.END_ARRAY : JsonToken.END_OBJECT, token); } } if (token.toChar() == '"') { buffer.append(token.toChar()); } buffer.append(token.getValue()); if (token.toChar() == '"') { buffer.append(token.toChar()); } if (tokenStack.size() == 1) { String value = buffer.toString(); buffer.setLength(0); state = State.EXPECT_COMMA; publish(value); } else { request(1); } break; case EXPECT_COMMA: if (JsonToken.COMMA.equals(token)) { state = State.EXPECT_VALUE; request(1); } else if (JsonToken.END_ARRAY.equals(token)) { tokenStack.pop(); onComplete(); } else { emitError(JsonToken.COMMA, token); } break; default: onError(new JsonParsingException("Unexpected token " + token.getValue(), null)); } }  @Override void onNext(JsonToken token);  }