plugins/TwitterBridge/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_log(LOG_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             throw new ServerException('Invalid URL scheme for HTTP stream reader');
  67         }
  68
  69         $host = parse_url($url, PHP_URL_HOST);
  70         $port = parse_url($url, PHP_URL_PORT);
  71         if (!$port) {
  72             if ($scheme == 'https') {
  73                 $port = 443;
  74             } else {
  75                 $port = 80;
  76             }
  77         }
  78
  79         $path = parse_url($url, PHP_URL_PATH);
  80         $query = parse_url($url, PHP_URL_QUERY);
  81         if ($query) {
  82             $path .= '?' . $query;
  83         }
  84
  85         $errno = $errstr = null;
  86         $timeout = 5;
  87         //$flags = STREAM_CLIENT_CONNECT | STREAM_CLIENT_ASYNC_CONNECT;
  88         $flags = STREAM_CLIENT_CONNECT;
  89         // @fixme add SSL params
  90         $this->socket = stream_socket_client("$rawScheme://$host:$port", $errno, $errstr, $timeout, $flags);
  91
  92         $this->send($this->httpOpen($host, $path));
  93
  94         stream_set_blocking($this->socket, false);
  95         $this->state = 'waiting';
  96     }
  97
  98     /**
  99      * Send some fun data off to the server.
 100      *
 101      * @param string $buffer
 102      */
 103     function send($buffer)
 104     {
 105         fwrite($this->socket, $buffer);
 106     }
 107
 108     /**
 109      * Read next packet of data from the socket.
 110      *
 111      * @return string
 112      */
 113     function read()
 114     {
 115         $buffer = fread($this->socket, 65536);
 116         return $buffer;
 117     }
 118
 119     /**
 120      * Build HTTP request headers.
 121      *
 122      * @param string $host
 123      * @param string $path
 124      * @return string
 125      */
 126     protected function httpOpen($host, $path)
 127     {
 128         $lines = array(
 129             "GET $path HTTP/1.1",
 130             "Host: $host",
 131             "User-Agent: StatusNet/" . STATUSNET_VERSION . " (TwitterBridgePlugin)",
 132             "Connection: close",
 133             "",
 134             ""
 135         );
 136         return implode(self::CRLF, $lines);
 137     }
 138
 139     /**
 140      * Close the current connection, if open.
 141      */
 142     public function close()
 143     {
 144         if ($this->isConnected()) {
 145             common_log(LOG_DEBUG, "$this->id closing connection.");
 146             fclose($this->socket);
 147             $this->socket = null;
 148         }
 149     }
 150
 151     /**
 152      * Are we currently connected?
 153      *
 154      * @return boolean
 155      */
 156     public function isConnected()
 157     {
 158         return $this->socket !== null;
 159     }
 160
 161     /**
 162      * Send any sockets we're listening on to the IO manager
 163      * to wait for input.
 164      *
 165      * @return array of resources
 166      */
 167     public function getSockets()
 168     {
 169         if ($this->isConnected()) {
 170             return array($this->socket);
 171         }
 172         return array();
 173     }
 174
 175     /**
 176      * Take a chunk of input over the horn and go go go! :D
 177      *
 178      * @param string $buffer
 179      */
 180     public function handleInput($socket)
 181     {
 182         if ($this->socket !== $socket) {
 183             throw new Exception('Got input from unexpected socket!');
 184         }
 185
 186         try {
 187             $buffer = $this->read();
 188             $lines = explode(self::CRLF, $buffer);
 189             foreach ($lines as $line) {
 190                 $this->handleLine($line);
 191             }
 192         } catch (Exception $e) {
 193             common_log(LOG_ERR, "$this->id aborting connection due to error: " . $e->getMessage());
 194             fclose($this->socket);
 195             throw $e;
 196         }
 197     }
 198
 199     protected function handleLine($line)
 200     {
 201         switch ($this->state)
 202         {
 203             case 'waiting':
 204                 $this->handleLineWaiting($line);
 205                 break;
 206             case 'headers':
 207                 $this->handleLineHeaders($line);
 208                 break;
 209             case 'active':
 210                 $this->handleLineActive($line);
 211                 break;
 212             default:
 213                 throw new Exception('Invalid state in handleLine: ' . $this->state);
 214         }
 215     }
 216
 217     /**
 218      *
 219      * @param <type> $line
 220      */
 221     protected function handleLineWaiting($line)
 222     {
 223         $bits = explode(' ', $line, 3);
 224         if (count($bits) != 3) {
 225             throw new Exception("Invalid HTTP response line: $line");
 226         }
 227
 228         list($http, $status, $text) = $bits;
 229         if (substr($http, 0, 5) != 'HTTP/') {
 230             throw new Exception("Invalid HTTP response line chunk '$http': $line");
 231         }
 232         if ($status != '200') {
 233             throw new Exception("Bad HTTP response code $status: $line");
 234         }
 235         common_log(LOG_DEBUG, "$this->id $line");
 236         $this->state = 'headers';
 237     }
 238
 239     protected function handleLineHeaders($line)
 240     {
 241         if ($line == '') {
 242             $this->state = 'active';
 243             common_log(LOG_DEBUG, "$this->id connection is active!");
 244         } else {
 245             common_log(LOG_DEBUG, "$this->id read HTTP header: $line");
 246             $this->responseHeaders[] = $line;
 247         }
 248     }
 249
 250     protected function handleLineActive($line)
 251     {
 252         if ($line == "") {
 253             // Server sends empty lines as keepalive.
 254             return;
 255         }
 256         $data = json_decode($line);
 257         if ($data) {
 258             $this->handleJson($data);
 259         } else {
 260             common_log(LOG_ERR, "$this->id received bogus JSON data: " . var_export($line, true));
 261         }
 262     }
 263
 264     abstract protected function handleJson(stdClass $data);
 265 }