StreamConverter implements GenericHttpMessageConverter<Stream<T>> { @Override public Stream<T> read(final Class<? extends Stream<T>> clazz, final HttpInputMessage inputMessage) throws HttpMessageNotReadableException { throw new UnsupportedOperationException(); }  @Override boolean canRead(final Class<?> clazz, final MediaType mediaType); @Override boolean canRead(final Type type, @Nullable final Class<?> contextClass, final MediaType mediaType); @Override List<MediaType> getSupportedMediaTypes(); @Override Stream<T> read(final Class<? extends Stream<T>> clazz, final HttpInputMessage inputMessage); @Override Stream<T> read(final Type type, final Class<?> contextClass, final HttpInputMessage inputMessage); @Override boolean canWrite(final Class<?> clazz, final MediaType mediaType); boolean canWrite(final Type type, final Class<?> clazz, final MediaType mediaType); @Override void write(final Stream<T> t, final MediaType mediaType, final HttpOutputMessage message); void write(final Stream<T> t, final Type type, final MediaType mediaType, final HttpOutputMessage message);  }
@SuppressWarnings("unchecked") @Test void shouldNotSupportReadStream() throws Exception { final StreamConverter unit = new StreamConverter<>(new ObjectMapper().findAndRegisterModules(), singletonList(APPLICATION_X_JSON_STREAM)); final HttpInputMessage input = mockWithContentType(APPLICATION_X_JSON_STREAM); when(input.getBody()).thenReturn(new ClassPathResource("account-stream.json").getInputStream()); assertThrows(UnsupportedOperationException.class, () -> unit.read(Stream.class, input)); }
@Test void shouldSupportGenericReadStream() throws Exception { final Type type = Streams.streamOf(AccountBody.class).getType(); final StreamConverter<AccountBody> unit = new StreamConverter<>(new ObjectMapper().findAndRegisterModules(), singletonList(APPLICATION_X_JSON_STREAM)); final HttpInputMessage input = mockWithContentType(APPLICATION_X_JSON_STREAM); when(input.getBody()).thenReturn(new ClassPathResource("account-stream.json").getInputStream()); final Stream<AccountBody> stream = unit.read(type, null, input); @SuppressWarnings("unchecked") final Consumer<? super AccountBody> verifier = mock(Consumer.class); stream.forEach(verifier); verify(verifier).accept(new AccountBody("1234567890", "Acme Corporation")); verify(verifier).accept(new AccountBody("1234567891", "Acme Company")); verify(verifier).accept(new AccountBody("1234567892", "Acme GmbH")); verify(verifier).accept(new AccountBody("1234567893", "Acme SE")); verify(verifier, times(4)).accept(any(AccountBody.class)); }
@Test void shouldSupportReadSequence() throws Exception { final Type type = Streams.streamOf(AccountBody.class).getType(); final StreamConverter<AccountBody> unit = new StreamConverter<>(new ObjectMapper().findAndRegisterModules(), singletonList(APPLICATION_JSON_SEQ)); final HttpInputMessage input = mockWithContentType(APPLICATION_JSON_SEQ); when(input.getBody()).thenReturn(new ClassPathResource("account-sequence.json").getInputStream()); final Stream<AccountBody> stream = unit.read(type, null, input); @SuppressWarnings("unchecked") final Consumer<? super AccountBody> verifier = mock(Consumer.class); stream.forEach(verifier); verify(verifier).accept(new AccountBody("1234567890", "Acme Corporation")); verify(verifier).accept(new AccountBody("1234567891", "Acme Company")); verify(verifier).accept(new AccountBody("1234567892", "Acme GmbH")); verify(verifier).accept(new AccountBody("1234567893", "Acme SE")); verify(verifier, times(4)).accept(any(AccountBody.class)); }
@Test void shouldThrowIOExceptionOnClose() throws Exception { final ObjectMapper mapper = spy(new ObjectMapper().findAndRegisterModules()); final JsonFactory factory = spy(mapper.getFactory()); final JsonParser parser = mock(JsonParser.class); when(mapper.getFactory()).thenReturn(factory); when(factory.createParser(any(InputStream.class))).thenReturn(parser); doThrow(new IOException()).when(parser).close(); final Type type = Streams.streamOf(AccountBody.class).getType(); final StreamConverter<AccountBody> unit = streamConverter(mapper); final HttpInputMessage input = mockWithContentType(APPLICATION_X_JSON_STREAM); when(input.getBody()).thenReturn(new ClassPathResource("account-stream.json").getInputStream()); assertThrows(IOException.class, () -> unit.read(type, null, input).close()); }
@Test void shouldFailOnReadForIOException() throws Exception { final StreamConverter<Object> unit = streamConverter(); final HttpInputMessage input = mockWithContentType(APPLICATION_X_JSON_STREAM); doThrow(new IOException()).when(input).getBody(); final HttpMessageNotReadableException exception = assertThrows(HttpMessageNotReadableException.class, () -> unit.read(Streams.streamOf(Object.class).getType(), null, input)); assertThat(exception.getCause(), is(instanceOf(IOException.class))); }