danog/postgres
1
0
Fork 0
mirror of https://github.com/danog/postgres.git synced 2024-12-11 08:59:42 +01:00
Code Issues Projects Releases Wiki Activity

WIP

Browse Source
This commit is contained in:
Aaron Piotrowski 2020-05-22 00:00:03 -05:00
parent 98214c247c
commit 0743e60583
No known key found for this signature in database
GPG Key ID: ADD1EF783EDE9EEB
9 changed files with 98 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
examples/basic.php
View File

@ -14,8 +14,7 @@ Amp\Loop::run(function () {
/** @var \Amp\Postgres\ResultSet $result */
$result = yield $connection->query('SHOW ALL');
while (yield $result->advance()) {
$row = $result->getCurrent();
while ($row = yield $result->continue()) {
\printf("%-35s = %s (%s)\n", $row['name'], $row['setting'], $row['description']);
}
});

3
examples/listen.php
View File

@ -31,8 +31,7 @@ Loop::run(function () {
return $pool->notify($channel, "Data 2"); // Send second notification.
});
while (yield $listener->advance()) {
$notification = $listener->getCurrent();
while ($notification = yield $listener->continue()) {
\printf(
"Received notification from PID %d on channel '%s' with payload: %s\n",
$notification->pid,

7
examples/multi-listen.php
View File

@ -3,9 +3,9 @@
require \dirname(__DIR__) . '/vendor/autoload.php';
use Amp\Iterator;
use Amp\Loop;
use Amp\Postgres;
use Amp\Stream;
Loop::run(function () {
$config = Postgres\ConnectionConfig::fromString('host=localhost user=postgres');
@ -51,10 +51,9 @@ Loop::run(function () {
return $pool->notify($channel1, "Data 1.2");
});
$iterator = Iterator\merge([$listener1, $listener2]); // Merge both listeners into single iterator.
$stream = Stream\merge([$listener1, $listener2]); // Merge both listeners into single iterator.
while (yield $iterator->advance()) {
$notification = $iterator->getCurrent();
while ($notification = yield $stream->continue()) {
\printf(
"Received notification from PID %d on channel '%s' with payload: %s\n",
$notification->pid,

3
examples/transaction.php
View File

@ -29,8 +29,7 @@ Amp\Loop::run(function () {
$format = "%-20s | %-10s\n";
\printf($format, 'TLD', 'Domain');
while (yield $result->advance()) {
$row = $result->getCurrent();
while ($row = yield $result->continue()) {
\printf($format, $row['domain'], $row['tld']);
}

11
src/PgSqlResultSet.php
View File

@ -25,12 +25,17 @@ final class PgSqlResultSet implements ResultSet
/** @var Internal\ArrayParser */
private $parser;
/** @var Promise<ResultSet|null> */
private $nextResult;
/**
* @param resource $handle PostgreSQL result resource.
* @param Promise<ResultSet|null> $nextResult
*/
public function __construct($handle)
public function __construct($handle, Promise $nextResult)
{
$this->handle = $handle;
$this->nextResult = $nextResult;
$numFields = \pg_num_fields($this->handle);
for ($i = 0; $i < $numFields; ++$i) {
@ -73,14 +78,14 @@ final class PgSqlResultSet implements ResultSet
return new Success($this->processRow($result));
}
public function dispose()
public function dispose(): void
{
$this->handle = null;
}
public function getNextResultSet(): Promise
{
return new Success;
return $this->nextResult;
}
/**

11
src/PqBufferedResultSet.php
View File

@ -16,13 +16,18 @@ final class PqBufferedResultSet implements ResultSet
/** @var int */
private $position = 0;
/** @var Promise<ResultSet|null> */
private $nextResult;
/**
* @param pq\Result $result PostgreSQL result object.
* @param Promise<ResultSet|null> $nextResult Promise for next result set.
*/
public function __construct(pq\Result $result)
public function __construct(pq\Result $result, Promise $nextResult)
{
$this->result = $result;
$this->result->autoConvert = pq\Result::CONV_SCALAR | pq\Result::CONV_ARRAY;
$this->nextResult = $nextResult;
}
/**
@ -44,14 +49,14 @@ final class PqBufferedResultSet implements ResultSet
/**
* @inheritDoc
*/
public function dispose()
public function dispose(): void
{
$this->result = null;
}
public function getNextResultSet(): Promise
{
return new Success; // Empty stub for now.
return $this->nextResult;
}
public function getFieldCount(): int

65
src/PqHandle.php
View File

@ -169,8 +169,6 @@ final class PqHandle implements Handle
*
* @return \Generator
*
* @resolve \Amp\Sql\CommandResult|\pq\Statement
*
* @throws FailureException
*/
private function send(?string $sql, callable $method, ...$args): \Generator
@ -208,19 +206,64 @@ final class PqHandle implements Handle
throw new FailureException("Unknown query result");
}
$result = $this->makeResult($result, $sql);
if ($handle instanceof pq\Statement) {
return $handle; // Will be wrapped into a PqStatement object.
}
return $result;
// switch ($result->status) {
// case pq\Result::EMPTY_QUERY:
// throw new QueryError("Empty query string");
//
// case pq\Result::COMMAND_OK:
// if ($handle instanceof pq\Statement) {
// return $handle; // Will be wrapped into a PqStatement object.
// }
//
// return new PqCommandResult($result);
//
// case pq\Result::TUPLES_OK:
// return new PqBufferedResultSet($result);
//
// case pq\Result::SINGLE_TUPLE:
// $this->busy = new Deferred;
// $result = new PqUnbufferedResultSet(
// coroutine(\Closure::fromCallable([$this, 'fetch'])),
// $result,
// \Closure::fromCallable([$this, 'release'])
// );
// return $result;
//
// case pq\Result::NONFATAL_ERROR:
// case pq\Result::FATAL_ERROR:
// throw new QueryExecutionError($result->errorMessage, $result->diag, null, $sql ?? '');
//
// case pq\Result::BAD_RESPONSE:
// throw new FailureException($result->errorMessage);
//
// default:
// throw new FailureException("Unknown result status");
// }
}
private function makeResult(pq\Result $result, ?string $sql)
{
switch ($result->status) {
case pq\Result::EMPTY_QUERY:
throw new QueryError("Empty query string");
case pq\Result::COMMAND_OK:
if ($handle instanceof pq\Statement) {
return $handle; // Will be wrapped into a PqStatement object.
}
return new PqCommandResult($result);
case pq\Result::TUPLES_OK:
return new PqBufferedResultSet($result);
if (!$this->handle->busy && ($next = $this->handle->getResult()) instanceof pq\Result) {
$next = new Success($this->makeResult($next, $sql));
}
return new PqBufferedResultSet($result, $next ?? new Success);
case pq\Result::SINGLE_TUPLE:
$this->busy = new Deferred;
@ -268,6 +311,14 @@ final class PqHandle implements Handle
switch ($result->status) {
case pq\Result::TUPLES_OK: // End of result set.
while (!$this->handle->busy && ($next = $this->handle->getResult()) instanceof pq\Result) {
if ($next->status === pq\Result::TUPLES_OK) {
continue;
}
return $this->makeResult($result, null);
}
return null;
case pq\Result::SINGLE_TUPLE:

24
src/PqUnbufferedResultSet.php
View File

@ -3,9 +3,9 @@
namespace Amp\Postgres;
use Amp\AsyncGenerator;
use Amp\Deferred;
use Amp\DisposedException;
use Amp\Promise;
use Amp\Success;
use pq;
final class PqUnbufferedResultSet implements ResultSet
@ -16,8 +16,11 @@ final class PqUnbufferedResultSet implements ResultSet
/** @var AsyncGenerator */
private $generator;
/** @var Deferred */
private $next;
/**
* @param callable():Promise<pq\Result> $fetch Function to fetch next result row.
* @param callable():Promise<pq\Result|ResultSet|null> $fetch Function to fetch next result row.
* @param \pq\Result $result PostgreSQL result object.
* @param callable():void $release Invoked once the result has been fully consumed.
*/
@ -25,8 +28,9 @@ final class PqUnbufferedResultSet implements ResultSet
{
$this->numCols = $result->numCols;
$this->next = $deferred = new Deferred;
$this->generator = new AsyncGenerator(static function (callable $yield) use (
$release, $result, $fetch
$deferred, $release, $result, $fetch
): \Generator {
try {
do {
@ -37,18 +41,26 @@ final class PqUnbufferedResultSet implements ResultSet
} while ($result instanceof pq\Result);
} catch (DisposedException $exception) {
// Discard remaining rows in the result set.
while ((yield $promise) instanceof pq\Result) {
while (($result = yield $promise) instanceof pq\Result) {
$promise = $fetch();
}
} finally {
if ($result instanceof ResultSet) {
$deferred->resolve($result);
return;
}
// Only release if there was no next result set.
$release();
$deferred->resolve(null);
}
});
}
public function getNextResultSet(): Promise
{
return new Success; // Empty stub for now.
return $this->next->promise();
}
/**
@ -62,7 +74,7 @@ final class PqUnbufferedResultSet implements ResultSet
/**
* @inheritDoc
*/
public function dispose()
public function dispose(): void
{
$this->generator->dispose();
}

2
test/AbstractLinkTest.php
View File

@ -73,8 +73,6 @@ abstract class AbstractLinkTest extends AsyncTestCase
public function testMultipleQueryWithTupleResult(): \Generator
{
$this->markTestSkipped('Unimplemented');
/** @var \Amp\Postgres\ResultSet $result */
$result = yield $this->connection->query("SELECT * FROM test; SELECT * FROM test");