RxJava2 Observable/Flowable Example with Mutiny

[Luke]

Hi,

Is there any equivalent piece of code for RxJava2 Flowable example on vert.x Mutiny?

Example Code from Reactive Postgresql page:

Flowable<Row> flowable = pool.rxGetConnection().flatMapPublisher(conn -> conn

  .rxBegin()

  .flatMapPublisher(tx ->

    conn

      .rxPrepare("SELECT * FROM users WHERE first_name LIKE $1")

      .flatMapPublisher(preparedQuery -> {

        // Fetch 50 rows at a time

        RowStream<Row> stream = preparedQuery.createStream(50, Tuple.of("julien"));

        return stream.toFlowable();

      })

      .doAfterTerminate(tx::commit)));

// Then subscribe

flowable.subscribe(new Subscriber<Row>() {

  private Subscription sub;

  @Override

  public void onSubscribe(Subscription subscription) {

    sub = subscription;

    subscription.request(1);

  }

  @Override

  public void onNext(Row row) {

    sub.request(1);

    System.out.println("User: " + row.getString("last_name"));

  }

  @Override

  public void onError(Throwable err) {

    System.out.println("Error: " + err.getMessage());

  }

  @Override

  public void onComplete() {

    System.out.println("End of stream");

  }

});

Thank you

[Julien Ponge]

Hi,

It's not in the docs, but I would start with pool.getConnection() and see where you go :-)

Cheers

[Thomas SEGISMONT]

Check out the Quarkus docs: https://quarkus.io/guides/reactive-sql-clients#transactions

In the example there's nothing Quarkus-specific: it's merely the PgPool with Mutiny API

--
You received this message because you are subscribed to the Google Groups "vert.x" group.
To unsubscribe from this group and stop receiving emails from it, send an email to vertx+un...@googlegroups.com.
To view this discussion on the web, visit https://groups.google.com/d/msgid/vertx/90683d69-f476-41d4-8b05-0c75ea59f068n%40googlegroups.com.