@Test public void testUnsuccessfulExecution() { when(factory.openStatelessSession()).thenReturn(session); when(session.createQuery(QUERY, Person.class)).thenReturn(query); when(query.setFetchSize(DefaultStreamer.DEFAULT_FETCH_SIZE)).thenReturn(query); when(query.setReadOnly(true)).thenReturn(query); when(query.scroll(ScrollMode.FORWARD_ONLY)).thenReturn(results); when(results.next()).thenThrow(new RuntimeException()); DefaultStreamer<Person> defaultStreamer = DefaultStreamer.<Person>builder() .query(QUERY) .type(Person.class) .sessionFactory(factory) .fetchSize(DefaultStreamer.DEFAULT_FETCH_SIZE) .build(); Flux<Person> flux = Flux.create(defaultStreamer::stream); StepVerifier.create(flux) .expectError(RuntimeException.class) .verify(); }
private Query<T> createQuery(StatelessSession session) { Query<T> compiledQuery = session.createQuery(query, type); setParameters(compiledQuery); if (maxResults > 0) { compiledQuery.setMaxResults(maxResults); } if (firstResult >= 0) { compiledQuery.setFirstResult(firstResult); } return compiledQuery; }
DefaultStreamer implements Streamer<T> { private Query<T> createQuery(StatelessSession session) { Query<T> compiledQuery = session.createQuery(query, type); setParameters(compiledQuery); if (maxResults > 0) { compiledQuery.setMaxResults(maxResults); } if (firstResult >= 0) { compiledQuery.setFirstResult(firstResult); } return compiledQuery; } }
DefaultStreamer implements Streamer<T> { private Query<T> createQuery(StatelessSession session) { Query<T> compiledQuery = session.createQuery(query, type); setParameters(compiledQuery); if (maxResults > 0) { compiledQuery.setMaxResults(maxResults); } if (firstResult >= 0) { compiledQuery.setFirstResult(firstResult); } return compiledQuery; }  }
DefaultStreamer implements Streamer<T> { private Query<T> createQuery(StatelessSession session) { Query<T> compiledQuery = session.createQuery(query, type); setParameters(compiledQuery); if (maxResults > 0) { compiledQuery.setMaxResults(maxResults); } if (firstResult >= 0) { compiledQuery.setFirstResult(firstResult); } return compiledQuery; }  void stream(FluxSink<T> sink); }
DefaultStreamer implements Streamer<T> { private Query<T> createQuery(StatelessSession session) { Query<T> compiledQuery = session.createQuery(query, type); setParameters(compiledQuery); if (maxResults > 0) { compiledQuery.setMaxResults(maxResults); } if (firstResult >= 0) { compiledQuery.setFirstResult(firstResult); } return compiledQuery; }  void stream(FluxSink<T> sink); static final int DEFAULT_FETCH_SIZE; }