docs: private static <T> Flowable<Notification<T>> createWithParameters(Connection con, String sql,
              Flowable<List<Object>> parameterGroups, List<ParameterPlaceholder> parameterPlaceholders,
              BiFunction<NamedCallableStatement, List<Object>, Single<T>> single) {
          Callable<NamedCallableStatement> resourceFactory = () -> Util.prepareCall(con, sql, parameterPlaceholders);
          final Function<NamedCallableStatement, Flowable<Notification<T>>> flowableFactory = //
                  stmt -> parameterGroups //
                          .flatMap(parameters -> single.apply(stmt, parameters).toFlowable()) //
                          .materialize() //
                          .doOnComplete(() -> Util.commit(stmt.stmt)) //
                          .doOnError(e -> Util.rollback(stmt.stmt));
          Consumer<NamedCallableStatement> disposer = Util::closeCallableStatementAndConnection;
          return Flowable.using(resourceFactory, flowableFactory, disposer, true);
      }