plugins/TwitterBridge/lib/jsonstreamreader.php

   1 <?php
   2 /**
   3  * StatusNet, the distributed open-source microblogging tool
   4  *
   5  * PHP version 5
   6  *
   7  * LICENCE: This program is free software: you can redistribute it and/or modify
   8  * it under the terms of the GNU Affero General Public License as published by
   9  * the Free Software Foundation, either version 3 of the License, or
  10  * (at your option) any later version.
  11  *
  12  * This program is distributed in the hope that it will be useful,
  13  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15  * GNU Affero General Public License for more details.
  16  *
  17  * You should have received a copy of the GNU Affero General Public License
  18  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  19  *
  20  * @category  Plugin
  21  * @package   StatusNet
  22  * @author    Brion Vibber <brion@status.net>
  23  * @copyright 2010 StatusNet, Inc.
  24  * @license   http://www.fsf.org/licensing/licenses/agpl-3.0.html GNU Affero General Public License version 3.0
  25  * @link      http://status.net/
  26  */
  27
  28 class OAuthData
  29 {
  30     public $consumer_key, $consumer_secret, $token, $token_secret;
  31 }
  32
  33 /**
  34  *
  35  */
  36 abstract class JsonStreamReader
  37 {
  38     const CRLF = "\r\n";
  39
  40     public $id;
  41     protected $socket = null;
  42     protected $state = 'init'; // 'init', 'connecting', 'waiting', 'headers', 'active'
  43
  44     public function __construct()
  45     {
  46         $this->id = get_class($this) . '.' . substr(md5(mt_rand()), 0, 8);
  47     }
  48
  49     /**
  50      * Starts asynchronous connect operation...
  51      *
  52      * @fixme Can we do the open-socket fully async to? (need write select infrastructure)
  53      *
  54      * @param string $url
  55      */
  56     public function connect($url)
  57     {
  58         common_debug("$this->id opening connection to $url");
  59
  60         $scheme = parse_url($url, PHP_URL_SCHEME);
  61         if ($scheme == 'http') {
  62             $rawScheme = 'tcp';
  63         } else if ($scheme == 'https') {
  64             $rawScheme = 'ssl';
  65         } else {
  66             // TRANS: Server exception thrown when an invalid URL scheme is detected.
  67             throw new ServerException(_m('Invalid URL scheme for HTTP stream reader.'));
  68         }
  69
  70         $host = parse_url($url, PHP_URL_HOST);
  71         $port = parse_url($url, PHP_URL_PORT);
  72         if (!$port) {
  73             if ($scheme == 'https') {
  74                 $port = 443;
  75             } else {
  76                 $port = 80;
  77             }
  78         }
  79
  80         $path = parse_url($url, PHP_URL_PATH);
  81         $query = parse_url($url, PHP_URL_QUERY);
  82         if ($query) {
  83             $path .= '?' . $query;
  84         }
  85
  86         $errno = $errstr = null;
  87         $timeout = 5;
  88         //$flags = STREAM_CLIENT_CONNECT | STREAM_CLIENT_ASYNC_CONNECT;
  89         $flags = STREAM_CLIENT_CONNECT;
  90         // @fixme add SSL params
  91         $this->socket = stream_socket_client("$rawScheme://$host:$port", $errno, $errstr, $timeout, $flags);
  92
  93         $this->send($this->httpOpen($host, $path));
  94
  95         stream_set_blocking($this->socket, false);
  96         $this->state = 'waiting';
  97     }
  98
  99     /**
 100      * Send some fun data off to the server.
 101      *
 102      * @param string $buffer
 103      */
 104     function send($buffer)
 105     {
 106         fwrite($this->socket, $buffer);
 107     }
 108
 109     /**
 110      * Read next packet of data from the socket.
 111      *
 112      * @return string
 113      */
 114     function read()
 115     {
 116         $buffer = fread($this->socket, 65536);
 117         return $buffer;
 118     }
 119
 120     /**
 121      * Build HTTP request headers.
 122      *
 123      * @param string $host
 124      * @param string $path
 125      * @return string
 126      */
 127     protected function httpOpen($host, $path)
 128     {
 129         $lines = array(
 130             "GET $path HTTP/1.1",
 131             "Host: $host",
 132             'User-Agent: ' . HTTPClient::userAgent() . ' (TwitterBridgePlugin)',
 133             "Connection: close",
 134             "",
 135             ""
 136         );
 137         return implode(self::CRLF, $lines);
 138     }
 139
 140     /**
 141      * Close the current connection, if open.
 142      */
 143     public function close()
 144     {
 145         if ($this->isConnected()) {
 146             common_debug("$this->id closing connection.");
 147             fclose($this->socket);
 148             $this->socket = null;
 149         }
 150     }
 151
 152     /**
 153      * Are we currently connected?
 154      *
 155      * @return boolean
 156      */
 157     public function isConnected()
 158     {
 159         return $this->socket !== null;
 160     }
 161
 162     /**
 163      * Send any sockets we're listening on to the IO manager
 164      * to wait for input.
 165      *
 166      * @return array of resources
 167      */
 168     public function getSockets()
 169     {
 170         if ($this->isConnected()) {
 171             return array($this->socket);
 172         }
 173         return array();
 174     }
 175
 176     /**
 177      * Take a chunk of input over the horn and go go go! :D
 178      *
 179      * @param string $buffer
 180      */
 181     public function handleInput($socket)
 182     {
 183         if ($this->socket !== $socket) {
 184             // TRANS: Exception thrown when input from an inexpected socket is encountered.
 185             throw new Exception(_m('Got input from unexpected socket!'));
 186         }
 187
 188         try {
 189             $buffer = $this->read();
 190             $lines = explode(self::CRLF, $buffer);
 191             foreach ($lines as $line) {
 192                 $this->handleLine($line);
 193             }
 194         } catch (Exception $e) {
 195             common_log(LOG_ERR, "$this->id aborting connection due to error: " . $e->getMessage());
 196             fclose($this->socket);
 197             throw $e;
 198         }
 199     }
 200
 201     protected function handleLine($line)
 202     {
 203         switch ($this->state)
 204         {
 205             case 'waiting':
 206                 $this->handleLineWaiting($line);
 207                 break;
 208             case 'headers':
 209                 $this->handleLineHeaders($line);
 210                 break;
 211             case 'active':
 212                 $this->handleLineActive($line);
 213                 break;
 214             default:
 215                 // TRANS: Exception thrown when an invalid state is encountered in handleLine.
 216                 // TRANS: %s is the invalid state.
 217                 throw new Exception(sprintf(_m('Invalid state in handleLine: %s.'),$this->state));
 218         }
 219     }
 220
 221     /**
 222      *
 223      * @param <type> $line
 224      */
 225     protected function handleLineWaiting($line)
 226     {
 227         $bits = explode(' ', $line, 3);
 228         if (count($bits) != 3) {
 229             // TRANS: Exception thrown when an invalid response line is encountered.
 230             // TRANS: %s is the invalid line.
 231             throw new Exception(sprintf(_m('Invalid HTTP response line: %s.'),$line));
 232         }
 233
 234         list($http, $status, $text) = $bits;
 235         if (substr($http, 0, 5) != 'HTTP/') {
 236             // TRANS: Exception thrown when an invalid response line part is encountered.
 237             // TRANS: %1$s is the chunk, %2$s is the line.
 238             throw new Exception(sprintf(_m('Invalid HTTP response line chunk "%1$s": %2$s.'),$http, $line));
 239         }
 240         if ($status != '200') {
 241             // TRANS: Exception thrown when an invalid response code is encountered.
 242             // TRANS: %1$s is the response code, %2$s is the line.
 243             throw new Exception(sprintf(_m('Bad HTTP response code %1$s: %2$s.'),$status,$line));
 244         }
 245         common_debug("$this->id $line");
 246         $this->state = 'headers';
 247     }
 248
 249     protected function handleLineHeaders($line)
 250     {
 251         if ($line == '') {
 252             $this->state = 'active';
 253             common_debug("$this->id connection is active!");
 254         } else {
 255             common_debug("$this->id read HTTP header: $line");
 256             $this->responseHeaders[] = $line;
 257         }
 258     }
 259
 260     protected function handleLineActive($line)
 261     {
 262         if ($line == "") {
 263             // Server sends empty lines as keepalive.
 264             return;
 265         }
 266         $data = json_decode($line);
 267         if ($data) {
 268             $this->handleJson($data);
 269         } else {
 270             common_log(LOG_ERR, "$this->id received bogus JSON data: " . var_export($line, true));
 271         }
 272     }
 273
 274     abstract protected function handleJson(stdClass $data);
 275 }