### Question:
Database { public <T> StreamerBuilder<T> stream(String query, Class<T> type) { return new StreamerBuilder<>(this, query, type); } Database(Map<String, String> settings, List<String> resourcePackages);  Database(Map<String, String> settings, List<String> resourcePackages,
      PoolConfiguration configuration);  Database(SessionFactory sessionFactory, int maxPoolSize); ReactiveExecutionBuilder<T> execute(Function<EntityManager, T> function); StreamerBuilder<T> stream(String query, Class<T> type);  }### Answer:
@Test public void testStreamExecution() throws SQLException { setupStreamTransactionMocks(IsolationLevel.READ_COMMITTED); Database database = new Database(sessionFactory, 1); Flux<Person> stream = database .stream(DefaultStreamerTest.QUERY, Person.class) .isolationLevel(IsolationLevel.READ_COMMITTED) .addParameter("key", "value") .addParameters(Collections.emptyMap()) .fetchSize(5) .flux(); StepVerifier.withVirtualTime(() -> stream) .expectNext(DefaultStreamerTest.PERSON) .verifyComplete(); }
@Test public void testStreamWithParameterListExecution() throws SQLException { setupStreamTransactionMocks(IsolationLevel.READ_COMMITTED); Database database = new Database(sessionFactory, 1); Flux<Person> stream = database .stream(DefaultStreamerTest.QUERY, Person.class) .isolationLevel(IsolationLevel.READ_COMMITTED) .parameterList(Collections.emptyList()) .fetchSize(5) .flux(); StepVerifier.withVirtualTime(() -> stream) .expectNext(DefaultStreamerTest.PERSON) .verifyComplete(); }
@Test public void testStreamWithLimitAndOffset() throws SQLException { setupStreamTransactionMocks(IsolationLevel.READ_COMMITTED); Database database = new Database(sessionFactory, 1); Flux<Person> stream = database .stream(DefaultStreamerTest.QUERY, Person.class) .isolationLevel(IsolationLevel.READ_COMMITTED) .parameterList(Collections.emptyList()) .firstResult(0) .maxResults(1) .fetchSize(5) .flux(); StepVerifier.withVirtualTime(() -> stream) .expectNext(DefaultStreamerTest.PERSON) .verifyComplete(); }