git.m6w6.name Git - m6w6/atick/blob - lib/atick/Ticker.php

   1 <?php
   2
   3 namespace atick;
   4
   5 /**
   6  * Asynchronnous resource handling, optionally (ab)using ticks
   7  *
   8  * Example with ticks:
   9  * <code>
  10  * <?php
  11  * declare(ticks=1);
  12  *
  13  * $conn = new \pq\Connection;
  14  * $conn->execAsync("SELECT * FROM foo", function ($rs) {
  15  *     var_dump($rs);
  16  * });
  17  *
  18  * $ticker = new \atick\Ticker;
  19  * $ticker->register();
  20  * $ticker->read($conn->socket, function($fd) use ($conn) {
  21  *     $conn->poll();
  22  *     if ($conn->busy) {
  23  *         return false;
  24  *     }
  25  *     $conn->getResult();
  26  *     return true;
  27  * });
  28  *
  29  * while (count($ticker));
  30  * ?>
  31  * </code>
  32  *
  33  * And an example without ticks:
  34  * <code>
  35  * <?php
  36  * $conn = new \pq\Connection;
  37  * $conn->execAsync("SELECT * FROM foo", function ($r) {
  38  *     var_dump($r);
  39  * });
  40  *
  41  * $ticker = new \atick\Ticker;
  42  * $ticker->read($conn->socket, function($fd) use ($conn) {
  43  *     $conn->poll();
  44  *     if ($conn->busy) {
  45  *         return false;
  46  *     }
  47  *     $conn->getResult();
  48  *     return true;
  49  * });
  50  *
  51  * while($ticker());
  52  * ?>
  53  * </code>
  54  */
  55 class Ticker implements \Countable
  56 {
  57         /**
  58          * @var array
  59          */
  60         protected $read = array();
  61
  62         /**
  63          * @var array
  64          */
  65         protected $write = array();
  66
  67         /**
  68          * Register the ticker as tick function
  69          * @return \atick\Ticker
  70          */
  71         function register() {
  72                 register_tick_function(array($this, "__invoke"));
  73                 return $this;
  74         }
  75
  76         /**
  77          * Unregister the ticker as tick function
  78          * @return \atick\Ticker
  79          */
  80         function unregister() {
  81                 unregister_tick_function(array($this, "__invoke"));
  82                 return $this;
  83         }
  84
  85         /**
  86          * The tick handler; calls atick\Ticker::wait(0)
  87          * @return int
  88          */
  89         function __invoke($timeout = 0) {
  90                 return $this->wait($timeout);
  91         }
  92
  93         /**
  94          * Wait for read/write readiness on the watched fds
  95          * @param float $timeout
  96          * @return int count of wached fds
  97          */
  98         function wait($timeout = 1) {
  99                 $r = $w = $e = array();
 100
 101                 foreach ($this->read as $s) {
 102                         is_resource($s[0]) and $r[] = $s[0];
 103                 }
 104
 105                 foreach ($this->write as $s) {
 106                         is_resource($s[0]) and $w[] = $s[0];
 107                 }
 108
 109                 $t = (int) $timeout;
 110                 $u = (int) (($timeout - $t) * 1000000);
 111
 112                 if (($r || $w) && stream_select($r, $w, $e, $t, $u)) {
 113                         foreach ($r as $s) {
 114                                 $this->read[(int)$s][1]($s);
 115                         }
 116                         foreach ($w as $s) {
 117                                 $this->write[(int)$s][1]($s);
 118                         }
 119                 }
 120
 121                 return $this->count();
 122         }
 123
 124         /**
 125          * Returns the count of watched fds
 126          * @implements \Countable
 127          * @return int
 128          */
 129         function count() {
 130                 foreach ($this->read as $i => $s) {
 131                         list($fd,,$verify) = $s;
 132                         if (!$verify($fd)) {
 133                                 unset($this->read[$i]);
 134                         }
 135                 }
 136
 137                 foreach ($this->write as $i => $s) {
 138                         list($fd,,$verify) = $s;
 139                         if (!$verify($fd)) {
 140                                 unset($this->write[$i]);
 141                         }
 142                 }
 143
 144                 return count($this->read) + count($this->write);
 145         }
 146
 147         /**
 148          * Attach a read handler
 149          * @param resource $fd
 150          * @param callable $onread void($fd) the descriptor is readable, read data, now!
 151          * @param callable $verify bool($fd) wheter the fd is still valid and should be watched
 152          * @return \atick\Ticker
 153          */
 154         function read($fd, callable $onread, callable $verify = null) {
 155                 $this->read[(int)$fd] = array($fd, $onread, $verify ?: function($fd) {
 156                         return is_resource($fd) && !feof($fd);
 157                 });
 158                 return $this;
 159         }
 160
 161         /**
 162          * Attach a write handler
 163          * @param resource $fd
 164          * @param callable $onwrite void($fd) the descriptor is writable, write data.
 165          * @param callable $verify bool($fd) wheter the fd is still valid and should be watched
 166          * @return \atick\Ticker
 167          */
 168         function write($fd, callable $onwrite, callable $verify = null) {
 169                 $this->write[(int)$fd] = array($fd, $onwrite, $verify ?: function($fd) {
 170                         return is_resource($fd) && !feof($fd);
 171                 });
 172                 return $this;
 173         }
 174 }