danog/postgres
1
0
Fork 0
mirror of https://github.com/danog/postgres.git synced 2024-11-26 20:15:02 +01:00
Code Issues Projects Releases Wiki Activity

Drop Operation trait

Browse Source 
Replaced with CompletionQueue.
This commit is contained in:
Aaron Piotrowski 2017-08-03 00:42:53 -05:00
parent f363b05b10
commit 32369079e6
No known key found for this signature in database
GPG Key ID: ADD1EF783EDE9EEB
3 changed files with 27 additions and 51 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
lib/Internal/Operation.php
View File

@ -1,40 +0,0 @@
<?php
namespace Amp\Postgres\Internal;
use Amp\Loop;
trait Operation {
/** @var bool */
private $complete = false;
/** @var callable[] */
private $onComplete = [];
public function onComplete(callable $onComplete) {
if ($this->complete) {
$onComplete();
return;
}
$this->onComplete[] = $onComplete;
}
private function complete() {
if ($this->complete) {
return;
}
$this->complete = true;
foreach ($this->onComplete as $callback) {
try {
$callback();
} catch (\Throwable $exception) {
Loop::defer(function () use ($exception) {
throw $exception; // Rethrow to event loop error handler.
});
}
}
$this->onComplete = null;
}
}

18
lib/Listener.php
View File

@ -2,13 +2,10 @@
namespace Amp\Postgres;
use Amp\CallableMaker;
use Amp\Iterator;
use Amp\Promise;
class Listener implements Iterator, Operation {
use CallableMaker, Internal\Operation;
/** @var \Amp\Iterator */
private $iterator;
@ -18,6 +15,9 @@ class Listener implements Iterator, Operation {
/** @var callable */
private $unlisten;
/** @var \Amp\Postgres\Internal\CompletionQueue */
private $queue;
/**
* @param \Amp\Iterator $iterator Iterator emitting notificatons on the channel.
* @param string $channel Channel name.
@ -27,14 +27,22 @@ class Listener implements Iterator, Operation {
$this->iterator = $iterator;
$this->channel = $channel;
$this->unlisten = $unlisten;
$this->queue = new Internal\CompletionQueue;
}
public function __destruct() {
if ($this->unlisten) {
$this->unlisten(); // Invokes $this->complete().
$this->unlisten(); // Invokes $this->queue->complete().
}
}
/**
* {@inheritdoc}
*/
public function onComplete(callable $onComplete) {
$this->queue->onComplete($onComplete);
}
/**
* {@inheritdoc}
*/
@ -73,7 +81,7 @@ class Listener implements Iterator, Operation {
/** @var \Amp\Promise $promise */
$promise = ($this->unlisten)($this->channel);
$this->unlisten = null;
$promise->onResolve($this->callableFromInstanceMethod("complete"));
$promise->onResolve([$this->queue, "complete"]);
return $promise;
}
}

20
lib/Transaction.php
View File

@ -2,12 +2,9 @@
namespace Amp\Postgres;
use Amp\CallableMaker;
use Amp\Promise;
class Transaction implements Handle, Operation {
use Internal\Operation, CallableMaker;
const UNCOMMITTED = 0;
const COMMITTED = 1;
const REPEATABLE = 2;
@ -19,6 +16,9 @@ class Transaction implements Handle, Operation {
/** @var int */
private $isolation;
/** @var \Amp\Postgres\Internal\CompletionQueue */
private $queue;
/**
* @param \Amp\Postgres\Handle $handle
* @param int $isolation
@ -39,14 +39,22 @@ class Transaction implements Handle, Operation {
}
$this->handle = $handle;
$this->queue = new Internal\CompletionQueue;
}
public function __destruct() {
if ($this->handle) {
$this->rollback(); // Invokes $this->complete().
$this->rollback(); // Invokes $this->queue->complete().
}
}
/**
* {@inheritdoc}
*/
public function onComplete(callable $onComplete) {
$this->queue->onComplete($onComplete);
}
/**
* @return bool True if the transaction is active, false if it has been committed or rolled back.
*/
@ -128,7 +136,7 @@ class Transaction implements Handle, Operation {
$promise = $this->handle->query("COMMIT");
$this->handle = null;
$promise->onResolve($this->callableFromInstanceMethod("complete"));
$promise->onResolve([$this->queue, "complete"]);
return $promise;
}
@ -147,7 +155,7 @@ class Transaction implements Handle, Operation {
$promise = $this->handle->query("ROLLBACK");
$this->handle = null;
$promise->onResolve($this->callableFromInstanceMethod("complete"));
$promise->onResolve([$this->queue, "complete"]);
return $promise;
}