3 * StatusNet, the distributed open-source microblogging tool
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.
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.
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/>.
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/
30 public $consumer_key, $consumer_secret, $token, $token_secret;
36 abstract class JsonStreamReader
41 protected $socket = null;
42 protected $state = 'init'; // 'init', 'connecting', 'waiting', 'headers', 'active'
44 public function __construct()
46 $this->id = get_class($this) . '.' . substr(md5(mt_rand()), 0, 8);
50 * Starts asynchronous connect operation...
52 * @fixme Can we do the open-socket fully async to? (need write select infrastructure)
56 public function connect($url)
58 common_log(LOG_DEBUG, "$this->id opening connection to $url");
60 $scheme = parse_url($url, PHP_URL_SCHEME);
61 if ($scheme == 'http') {
63 } else if ($scheme == 'https') {
66 throw new ServerException('Invalid URL scheme for HTTP stream reader');
69 $host = parse_url($url, PHP_URL_HOST);
70 $port = parse_url($url, PHP_URL_PORT);
72 if ($scheme == 'https') {
79 $path = parse_url($url, PHP_URL_PATH);
80 $query = parse_url($url, PHP_URL_QUERY);
82 $path .= '?' . $query;
85 $errno = $errstr = null;
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);
92 $this->send($this->httpOpen($host, $path));
94 stream_set_blocking($this->socket, false);
95 $this->state = 'waiting';
99 * Send some fun data off to the server.
101 * @param string $buffer
103 function send($buffer)
105 fwrite($this->socket, $buffer);
109 * Read next packet of data from the socket.
115 $buffer = fread($this->socket, 65536);
120 * Build HTTP request headers.
122 * @param string $host
123 * @param string $path
126 protected function httpOpen($host, $path)
129 "GET $path HTTP/1.1",
131 "User-Agent: StatusNet/" . STATUSNET_VERSION . " (TwitterBridgePlugin)",
136 return implode(self::CRLF, $lines);
140 * Close the current connection, if open.
142 public function close()
144 if ($this->isConnected()) {
145 common_log(LOG_DEBUG, "$this->id closing connection.");
146 fclose($this->socket);
147 $this->socket = null;
152 * Are we currently connected?
156 public function isConnected()
158 return $this->socket !== null;
162 * Send any sockets we're listening on to the IO manager
165 * @return array of resources
167 public function getSockets()
169 if ($this->isConnected()) {
170 return array($this->socket);
176 * Take a chunk of input over the horn and go go go! :D
178 * @param string $buffer
180 public function handleInput($socket)
182 if ($this->socket !== $socket) {
183 throw new Exception('Got input from unexpected socket!');
187 $buffer = $this->read();
188 $lines = explode(self::CRLF, $buffer);
189 foreach ($lines as $line) {
190 $this->handleLine($line);
192 } catch (Exception $e) {
193 common_log(LOG_ERR, "$this->id aborting connection due to error: " . $e->getMessage());
194 fclose($this->socket);
199 protected function handleLine($line)
201 switch ($this->state)
204 $this->handleLineWaiting($line);
207 $this->handleLineHeaders($line);
210 $this->handleLineActive($line);
213 throw new Exception('Invalid state in handleLine: ' . $this->state);
219 * @param <type> $line
221 protected function handleLineWaiting($line)
223 $bits = explode(' ', $line, 3);
224 if (count($bits) != 3) {
225 throw new Exception("Invalid HTTP response line: $line");
228 list($http, $status, $text) = $bits;
229 if (substr($http, 0, 5) != 'HTTP/') {
230 throw new Exception("Invalid HTTP response line chunk '$http': $line");
232 if ($status != '200') {
233 throw new Exception("Bad HTTP response code $status: $line");
235 common_log(LOG_DEBUG, "$this->id $line");
236 $this->state = 'headers';
239 protected function handleLineHeaders($line)
242 $this->state = 'active';
243 common_log(LOG_DEBUG, "$this->id connection is active!");
245 common_log(LOG_DEBUG, "$this->id read HTTP header: $line");
246 $this->responseHeaders[] = $line;
250 protected function handleLineActive($line)
253 // Server sends empty lines as keepalive.
256 $data = json_decode($line);
258 $this->handleJson($data);
260 common_log(LOG_ERR, "$this->id received bogus JSON data: " . var_export($line, true));
264 abstract protected function handleJson(stdClass $data);