响应式 PostgreSQL 客户端

<dependency>
  <groupId>io.vertx</groupId>
  <artifactId>vertx-pg-client</artifactId>
  <version>5.0.1</version>
</dependency>

dependencies {
  compile 'io.vertx:vertx-pg-client:5.0.1'
}

PgConnectOptions connectOptions = new PgConnectOptions()
  .setPort(5432)
  .setHost("the-host")
  .setDatabase("the-db")
  .setUser("user")
  .setPassword("secret");

// Pool options
PoolOptions poolOptions = new PoolOptions()
  .setMaxSize(5);

// Create the client pool
SqlClient client = PgBuilder
  .client()
  .with(poolOptions)
  .connectingTo(connectOptions)
  .build();

// A simple query
client
  .query("SELECT * FROM users WHERE id='julien'")
  .execute()
  .onComplete(ar -> {
    if (ar.succeeded()) {
      RowSet<Row> result = ar.result();
      System.out.println("Got " + result.size() + " rows ");
    } else {
      System.out.println("Failure: " + ar.cause().getMessage());
    }

    // Now close the pool
    client.close();
  });

PgConnectOptions connectOptions = new PgConnectOptions()
  .setPort(5432)
  .setHost("the-host")
  .setDatabase("the-db")
  .setUser("user")
  .setPassword("secret");

// Pool options
PoolOptions poolOptions = new PoolOptions()
  .setMaxSize(5);

// Create the pooled client
SqlClient client = PgBuilder
  .client()
  .with(poolOptions)
  .connectingTo(connectOptions)
  .build();

PgConnectOptions connectOptions = new PgConnectOptions()
  .setPort(5432)
  .setHost("the-host")
  .setDatabase("the-db")
  .setUser("user")
  .setPassword("secret");

// Pool options
PoolOptions poolOptions = new PoolOptions()
  .setMaxSize(5);

// Create the pooled client
SqlClient client = PgBuilder
  .client()
  .with(poolOptions)
  .connectingTo(connectOptions)
  .using(vertx)
  .build();

client.close();

PgConnectOptions connectOptions = new PgConnectOptions()
  .setPort(5432)
  .setHost("the-host")
  .setDatabase("the-db")
  .setUser("user")
  .setPassword("secret");

// Pool options
PoolOptions poolOptions = new PoolOptions()
  .setMaxSize(5);

// Create the pooled client
Pool pool = PgBuilder
  .pool()
  .with(poolOptions)
  .connectingTo(connectOptions)
  .using(vertx)
  .build();

// Get a connection from the pool
pool.getConnection().compose(conn -> {
  System.out.println("Got a connection from the pool");

  // All operations execute on the same connection
  return conn
    .query("SELECT * FROM users WHERE id='julien'")
    .execute()
    .compose(res -> conn
      .query("SELECT * FROM users WHERE id='emad'")
      .execute())
    .onComplete(ar -> {
      // Release the connection to the pool
      conn.close();
    });
}).onComplete(ar -> {
  if (ar.succeeded()) {
    System.out.println("Done");
  } else {
    System.out.println("Something went wrong " + ar.cause().getMessage());
  }
});

Pool pool = PgBuilder
  .pool()
  .connectingTo(connectOptions.setPipeliningLimit(16))
  .with(poolOptions)
  .using(vertx)
  .build();

SqlClient client = PgBuilder
  .client()
  .with(poolOptions)
  .connectingTo(connectOptions)
  .using(vertx)
  .build();

// Pipelined
Future<RowSet<Row>> res1 = client.query(sql).execute();

// Connection pool
Pool pool = PgBuilder
  .pool()
  .connectingTo(connectOptions)
  .with(poolOptions)
  .using(vertx)
  .build();

// Not pipelined
Future<RowSet<Row>> res2 = pool.query(sql).execute();

Pool pool = Pool.pool(database, new PoolOptions().setMaxSize(maxSize));
vertx.deployVerticle(() -> new VerticleBase() {
  @Override
  public Future<?> start() throws Exception {
    // Use the pool
    return super.start();
  }
}, new DeploymentOptions().setInstances(4));

vertx.deployVerticle(() -> new VerticleBase() {
  Pool pool;
  @Override
  public Future<?> start() throws Exception {
    // Get or create a shared pool
    // this actually creates a lease to the pool
    // when the verticle is undeployed, the lease will be released automaticaly
    pool = PgBuilder.pool()
      .with(new PoolOptions()
        .setMaxSize(maxSize)
        .setShared(true)
        .setName("my-pool"))
      .connectingTo(database)
      .using(vertx)
      .build();
    return super.start();
  }
}, new DeploymentOptions().setInstances(4));

Pool pool = PgBuilder.pool()
  .with(new PoolOptions()
    .setMaxSize(maxSize)
    .setShared(true)
    .setName("my-pool")
    .setEventLoopSize(4))
  .connectingTo(database)
  .using(vertx)
  .build();

PgConnectOptions connectOptions = new PgConnectOptions()
  .setHost("/var/run/postgresql")
  .setPort(5432)
  .setDatabase("the-db");

// Pool options
PoolOptions poolOptions = new PoolOptions()
  .setMaxSize(5);

// Create the pooled client
Pool client = PgBuilder
  .pool()
  .connectingTo(connectOptions)
  .with(poolOptions)
  .build();

// Create the pooled client with a vertx instance
// Make sure the vertx instance has enabled native transports
Pool client2 = PgBuilder
  .pool()
  .connectingTo(connectOptions)
  .with(poolOptions)
  .using(vertx)
  .build();

options
  .setReconnectAttempts(2)
  .setReconnectInterval(1000);

client
  .query("SELECT * FROM users WHERE id='julien'")
  .execute()
  .onComplete(ar -> {
  if (ar.succeeded()) {
    RowSet<Row> result = ar.result();
    System.out.println("Got " + result.size() + " rows ");
  } else {
    System.out.println("Failure: " + ar.cause().getMessage());
  }
});

client
  .preparedQuery("INSERT INTO color (color_name) VALUES ($1), ($2), ($3) RETURNING color_id")
  .execute(Tuple.of("white", "red", "blue"))
  .onSuccess(rows -> {
    for (Row row : rows) {
      System.out.println("generated key: " + row.getInteger("color_id"));
    }
});

client
  .query("DELETE FROM color RETURNING color_name")
  .execute()
  .onSuccess(rows -> {
    for (Row row : rows) {
      System.out.println("deleted color: " + row.getString("color_name"));
    }
  });

client
  .preparedQuery("INSERT INTO color (color_name) VALUES ($1) RETURNING color_id")
  .executeBatch(Arrays.asList(Tuple.of("white"), Tuple.of("red"), Tuple.of("blue")))
  .onSuccess(res -> {
    for (RowSet<Row> rows = res; rows != null; rows = rows.next()) {
      Integer colorId = rows.iterator().next().getInteger("color_id");
      System.out.println("generated key: " + colorId);
    }
  });

connection
  .prepare("SELECT * FROM users WHERE first_name LIKE $1")
  .onComplete(ar0 -> {
  if (ar0.succeeded()) {
    PreparedStatement pq = ar0.result();

    // Cursors require to run within a transaction
    connection
      .begin()
      .onComplete(ar1 -> {
      if (ar1.succeeded()) {
        Transaction tx = ar1.result();

        // Create a cursor
        Cursor cursor = pq.cursor(Tuple.of("julien"));

        // Read 50 rows
        cursor.read(50).onComplete(ar2 -> {
          if (ar2.succeeded()) {
            RowSet<Row> rows = ar2.result();

            // Check for more ?
            if (cursor.hasMore()) {
              // Repeat the process...
            } else {
              // No more rows - commit the transaction
              tx.commit();
            }
          }
        });
      }
    });
  }
});

cursor
  .read(50)
  .onComplete(ar2 -> {
  if (ar2.succeeded()) {
    // Close the cursor
    cursor.close();
  }
});

connection
  .prepare("SELECT * FROM users WHERE first_name LIKE $1")
  .onComplete(ar0 -> {
  if (ar0.succeeded()) {
    PreparedStatement pq = ar0.result();

    // Streams require to run within a transaction
    connection
      .begin()
      .onComplete(ar1 -> {
      if (ar1.succeeded()) {
        Transaction tx = ar1.result();

        // Fetch 50 rows at a time
        RowStream<Row> stream = pq.createStream(50, Tuple.of("julien"));

        // Use the stream
        stream.exceptionHandler(err -> {
          System.out.println("Error: " + err.getMessage());
        });
        stream.endHandler(v -> {
          // Close the stream to release the resources in the database
          stream
            .close()
            .onComplete(closed -> {
            tx.commit()
              .onComplete(committed -> {
              System.out.println("End of stream");
            });
          });
        });
        stream.handler(row -> {
          System.out.println("User: " + row.getString("last_name"));
        });
      }
    });
  }
});

options.setTracingPolicy(TracingPolicy.ALWAYS);

pool
  .query("SELECT 1::BIGINT \"VAL\"")
  .execute()
  .onComplete(ar -> {
    RowSet<Row> rowSet = ar.result();
    Row row = rowSet.iterator().next();

    // Stored as java.lang.Long
    Object value = row.getValue(0);

    // Convert to java.lang.Integer
    Integer intValue = row.getInteger(0);
  });

pool
  .query("SELECT 1::BIGINT \"VAL\"")
  .execute()
  .onComplete(ar -> {
    RowSet<Row> rowSet = ar.result();
    Row row = rowSet.iterator().next();

    // Stored as java.lang.Long
    Object value = row.getValue(0);

    // Convert to java.lang.Integer
    Integer intValue = row.getInteger(0);
  });

Tuple tuple = Tuple.of(new String[]{ "a", "tuple", "with", "arrays" });

// Add a string array to the tuple
tuple.addArrayOfString(new String[]{"another", "array"});

// Get the first array of string
String[] array = tuple.getArrayOfStrings(0);

client
  .query("SELECT 'infinity'::DATE \"LocalDate\"")
  .execute()
  .onComplete(ar -> {
    if (ar.succeeded()) {
      Row row = ar.result().iterator().next();
      System.out.println(row.getLocalDate("LocalDate").equals(LocalDate.MAX));
    } else {
      System.out.println("Failure: " + ar.cause().getMessage());
    }
  });

client
  .preparedQuery("SELECT address, (address).city FROM address_book WHERE id=$1")
  .execute(Tuple.of(3))
  .onComplete(ar -> {
  if (ar.succeeded()) {
    RowSet<Row> rows = ar.result();
    for (Row row : rows) {
      System.out.println("Full Address " + row.getString(0) + ", City " + row.getString(1));
    }
  } else {
    System.out.println("Failure: " + ar.cause().getMessage());
  }
});

client
  .preparedQuery("INSERT INTO address_book (id, address) VALUES ($1, $2)")
  .execute(Tuple.of(3, "('Anytown', 'Second Ave', false)"))
  .onComplete(ar -> {
  if (ar.succeeded()) {
    RowSet<Row> rows = ar.result();
    System.out.println(rows.rowCount());
  } else {
    System.out.println("Failure: " + ar.cause().getMessage());
  }
});

client
  .preparedQuery("SELECT to_tsvector( $1 ) @@ to_tsquery( $2 )")
  .execute(Tuple.of("fat cats ate fat rats", "fat & rat"))
  .onComplete(ar -> {
  if (ar.succeeded()) {
    RowSet<Row> rows = ar.result();
    for (Row row : rows) {
      System.out.println("Match : " + row.getBoolean(0));
    }
  } else {
    System.out.println("Failure: " + ar.cause().getMessage());
  }
});

client
  .preparedQuery("SELECT to_tsvector( $1 ), to_tsquery( $2 )")
  .execute(Tuple.of("fat cats ate fat rats", "fat & rat"))
  .onComplete(ar -> {
  if (ar.succeeded()) {
    RowSet<Row> rows = ar.result();
    for (Row row : rows) {
      System.out.println("Vector : " + row.getString(0) + ", query : "+row.getString(1));
    }
  } else {
    System.out.println("Failure: " + ar.cause().getMessage());
  }
});

client
  .preparedQuery("INSERT INTO colors VALUES ($2)")
  .execute(Tuple.of("red"))
  .onComplete(res -> {
    // ...
  });

client
  .preparedQuery("INSERT INTO colors VALUES ($1)")
  .execute(Tuple.of(Color.red))
  .flatMap(res ->
    client
      .preparedQuery("SELECT color FROM colors")
      .execute()
  ).onComplete(res -> {
    if (res.succeeded()) {
      RowSet<Row> rows = res.result();
      for (Row row : rows) {
        System.out.println(row.get(Color.class, "color"));
      }
    }
});

Collector<Row, ?, Map<Long, String>> collector = Collectors.toMap(
  row -> row.getLong("id"),
  row -> row.getString("last_name"));

// Run the query with the collector
client.query("SELECT * FROM users")
  .collecting(collector)
  .execute()
  .onComplete(ar -> {
  if (ar.succeeded()) {
    SqlResult<Map<Long, String>> result = ar.result();

    // Get the map created by the collector
    Map<Long, String> map = result.value();
    System.out.println("Got " + map);
  } else {
    System.out.println("Failure: " + ar.cause().getMessage());
  }
});

Collector<Row, ?, String> collector = Collectors.mapping(
  row -> row.getString("last_name"),
  Collectors.joining(",", "(", ")")
);

// Run the query with the collector
client
  .query("SELECT * FROM users")
  .collecting(collector)
  .execute()
  .onComplete(ar -> {
    if (ar.succeeded()) {
      SqlResult<String> result = ar.result();

      // Get the string created by the collector
      String list = result.value();
      System.out.println("Got " + list);
    } else {
      System.out.println("Failure: " + ar.cause().getMessage());
    }
  });

connection.notificationHandler(notification -> {
  System.out.println("Received " + notification.getPayload() + " on channel " + notification.getChannel());
});

connection
  .query("LISTEN some-channel")
  .execute()
  .onComplete(ar -> {
  System.out.println("Subscribed to channel");
});

PgSubscriber subscriber = PgSubscriber.subscriber(vertx, new PgConnectOptions()
  .setPort(5432)
  .setHost("the-host")
  .setDatabase("the-db")
  .setUser("user")
  .setPassword("secret")
);

// You can set the channel before connect
subscriber.channel("channel1").handler(payload -> {
  System.out.println("Received " + payload);
});

subscriber
  .connect()
  .onComplete(ar -> {
  if (ar.succeeded()) {

    // Or you can set the channel after connect
    subscriber.channel("channel2").handler(payload -> {
      System.out.println("Received " + payload);
    });
  }
});

PgSubscriber subscriber = PgSubscriber.subscriber(vertx, new PgConnectOptions()
  .setPort(5432)
  .setHost("the-host")
  .setDatabase("the-db")
  .setUser("user")
  .setPassword("secret")
);

subscriber
  .connect()
  .onComplete(ar -> {
    if (ar.succeeded()) {
      // Complex channel name - name in PostgreSQL requires a quoted ID
      subscriber.channel("Complex.Channel.Name").handler(payload -> {
        System.out.println("Received " + payload);
      });
      subscriber.channel("Complex.Channel.Name").subscribeHandler(subscribed -> {
        subscriber.actualConnection()
          .query("NOTIFY \"Complex.Channel.Name\", 'msg'")
          .execute()
          .onComplete(notified -> {
            System.out.println("Notified \"Complex.Channel.Name\"");
          });
      });

      // PostgreSQL simple ID's are forced lower-case
      subscriber.channel("simple_channel").handler(payload -> {
          System.out.println("Received " + payload);
      });
      subscriber.channel("simple_channel").subscribeHandler(subscribed -> {
        // The following simple channel identifier is forced to lower case
        subscriber.actualConnection()
          .query("NOTIFY Simple_CHANNEL, 'msg'")
          .execute()
          .onComplete(notified -> {
            System.out.println("Notified simple_channel");
          });
      });

      // The following channel name is longer than the current
      // (NAMEDATALEN = 64) - 1 == 63 character limit and will be truncated
      subscriber.channel("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaabbbbb")
        .handler(payload -> {
        System.out.println("Received " + payload);
      });
    }
  });

PgSubscriber subscriber = PgSubscriber.subscriber(vertx, new PgConnectOptions()
  .setPort(5432)
  .setHost("the-host")
  .setDatabase("the-db")
  .setUser("user")
  .setPassword("secret")
);

// Reconnect at most 10 times after 100 ms each
subscriber.reconnectPolicy(retries -> {
  if (retries < 10) {
    return 100L;
  } else {
    return -1L;
  }
});

connection.noticeHandler(notice -> {
  System.out.println("Received notice " + notice.getSeverity() + "" + notice.getMessage());
});

connection
  .query("SELECT pg_sleep(20)")
  .execute()
  .onComplete(ar -> {
  if (ar.succeeded()) {
    // imagine this is a long query and is still running
    System.out.println("Query success");
  } else {
    // the server will abort the current query after cancelling request
    System.out.println("Failed to query due to " + ar.cause().getMessage());
  }
});
connection
  .cancelRequest()
  .onComplete(ar -> {
  if (ar.succeeded()) {
    System.out.println("Cancelling request has been sent");
  } else {
    System.out.println("Failed to send cancelling request");
  }
});

PgConnectOptions options = new PgConnectOptions()
  .setPort(5432)
  .setHost("the-host")
  .setDatabase("the-db")
  .setUser("user")
  .setPassword("secret")
  .setSslMode(SslMode.VERIFY_CA)
  .setSslOptions(new ClientSSLOptions().setTrustOptions(new PemTrustOptions().addCertPath("/path/to/cert.pem")));

PgConnection
  .connect(vertx, options)
  .onComplete(res -> {
  if (res.succeeded()) {
    // Connected with SSL
  } else {
    System.out.println("Could not connect " + res.cause());
  }
});

connectOptions.setUseLayer7Proxy(true);