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...
54 public function connect($url)
56 common_log(LOG_DEBUG, "$this->id opening connection to $url");
58 $scheme = parse_url($url, PHP_URL_SCHEME);
59 if ($scheme == 'http') {
61 } else if ($scheme == 'https') {
64 throw new ServerException('Invalid URL scheme for HTTP stream reader');
67 $host = parse_url($url, PHP_URL_HOST);
68 $port = parse_url($url, PHP_URL_PORT);
70 if ($scheme == 'https') {
77 $path = parse_url($url, PHP_URL_PATH);
78 $query = parse_url($url, PHP_URL_QUERY);
80 $path .= '?' . $query;
83 $errno = $errstr = null;
85 //$flags = STREAM_CLIENT_CONNECT | STREAM_CLIENT_ASYNC_CONNECT;
86 $flags = STREAM_CLIENT_CONNECT;
87 // @fixme add SSL params
88 $this->socket = stream_socket_client("$rawScheme://$host:$port", $errno, $errstr, $timeout, $flags);
90 $this->send($this->httpOpen($host, $path));
92 stream_set_blocking($this->socket, false);
93 $this->state = 'waiting';
96 function send($buffer)
100 fwrite($this->socket, $buffer);
106 $buffer = fread($this->socket, 65536);
111 protected function httpOpen($host, $path)
114 "GET $path HTTP/1.1",
116 "User-Agent: StatusNet/" . STATUSNET_VERSION . " (TwitterBridgePlugin)",
121 return implode(self::CRLF, $lines);
125 * Close the current connection, if open.
127 public function close()
129 if ($this->isConnected()) {
130 common_log(LOG_DEBUG, "$this->id closing connection.");
131 fclose($this->socket);
132 $this->socket = null;
137 * Are we currently connected?
141 public function isConnected()
143 return $this->socket !== null;
147 * Send any sockets we're listening on to the IO manager
150 * @return array of resources
152 public function getSockets()
154 if ($this->isConnected()) {
155 return array($this->socket);
161 * Take a chunk of input over the horn and go go go! :D
162 * @param string $buffer
164 function handleInput($socket)
166 if ($this->socket !== $socket) {
167 throw new Exception('Got input from unexpected socket!');
170 $buffer = $this->read();
171 switch ($this->state)
174 $this->handleInputWaiting($buffer);
177 $this->handleInputHeaders($buffer);
180 $this->handleInputActive($buffer);
183 throw new Exception('Invalid state in handleInput: ' . $this->state);
187 function handleInputWaiting($buffer)
189 common_log(LOG_DEBUG, "$this->id Does this happen? " . $buffer);
190 $this->state = 'headers';
191 $this->handleInputHeaders($buffer);
194 function handleInputHeaders($buffer)
196 $lines = explode(self::CRLF, $buffer);
197 foreach ($lines as $line) {
199 $this->state = 'active';
200 common_log(LOG_DEBUG, "$this->id connection is active!");
202 common_log(LOG_DEBUG, "$this->id read HTTP header: $line");
203 $this->responseHeaders[] = $line;
208 function handleInputActive($buffer)
210 // One JSON object on each line...
211 // Will we always deliver on packet boundaries?
212 $lines = explode("\n", $buffer);
213 foreach ($lines as $line) {
214 $data = json_decode($line, true);
216 $this->handleJson($data);
218 common_log(LOG_ERR, "$this->id received bogus JSON data: " . $line);
223 abstract function handleJson(array $data);