. * * @category Plugin * @package StatusNet * @author Brion Vibber * @copyright 2010 StatusNet, Inc. * @license http://www.fsf.org/licensing/licenses/agpl-3.0.html GNU Affero General Public License version 3.0 * @link http://status.net/ */ class OAuthData { public $consumer_key, $consumer_secret, $token, $token_secret; } /** * */ abstract class JsonStreamReader { const CRLF = "\r\n"; public $id; protected $socket = null; protected $state = 'init'; // 'init', 'connecting', 'waiting', 'headers', 'active' public function __construct() { $this->id = get_class($this) . '.' . substr(md5(mt_rand()), 0, 8); } /** * Starts asynchronous connect operation... * * @param $url */ public function connect($url) { common_log(LOG_DEBUG, "$this->id opening connection to $url"); $scheme = parse_url($url, PHP_URL_SCHEME); if ($scheme == 'http') { $rawScheme = 'tcp'; } else if ($scheme == 'https') { $rawScheme = 'ssl'; } else { throw new ServerException('Invalid URL scheme for HTTP stream reader'); } $host = parse_url($url, PHP_URL_HOST); $port = parse_url($url, PHP_URL_PORT); if (!$port) { if ($scheme == 'https') { $port = 443; } else { $port = 80; } } $path = parse_url($url, PHP_URL_PATH); $query = parse_url($url, PHP_URL_QUERY); if ($query) { $path .= '?' . $query; } $errno = $errstr = null; $timeout = 5; //$flags = STREAM_CLIENT_CONNECT | STREAM_CLIENT_ASYNC_CONNECT; $flags = STREAM_CLIENT_CONNECT; // @fixme add SSL params $this->socket = stream_socket_client("$rawScheme://$host:$port", $errno, $errstr, $timeout, $flags); $this->send($this->httpOpen($host, $path)); stream_set_blocking($this->socket, false); $this->state = 'waiting'; } function send($buffer) { echo "Writing...\n"; var_dump($buffer); fwrite($this->socket, $buffer); } function read() { echo "Reading...\n"; $buffer = fread($this->socket, 65536); var_dump($buffer); return $buffer; } protected function httpOpen($host, $path) { $lines = array( "GET $path HTTP/1.1", "Host: $host", "User-Agent: StatusNet/" . STATUSNET_VERSION . " (TwitterBridgePlugin)", "Connection: close", "", "" ); return implode(self::CRLF, $lines); } /** * Close the current connection, if open. */ public function close() { if ($this->isConnected()) { common_log(LOG_DEBUG, "$this->id closing connection."); fclose($this->socket); $this->socket = null; } } /** * Are we currently connected? * * @return boolean */ public function isConnected() { return $this->socket !== null; } /** * Send any sockets we're listening on to the IO manager * to wait for input. * * @return array of resources */ public function getSockets() { if ($this->isConnected()) { return array($this->socket); } return array(); } /** * Take a chunk of input over the horn and go go go! :D * @param string $buffer */ function handleInput($socket) { if ($this->socket !== $socket) { throw new Exception('Got input from unexpected socket!'); } $buffer = $this->read(); switch ($this->state) { case 'waiting': $this->handleInputWaiting($buffer); break; case 'headers': $this->handleInputHeaders($buffer); break; case 'active': $this->handleInputActive($buffer); break; default: throw new Exception('Invalid state in handleInput: ' . $this->state); } } function handleInputWaiting($buffer) { common_log(LOG_DEBUG, "$this->id Does this happen? " . $buffer); $this->state = 'headers'; $this->handleInputHeaders($buffer); } function handleInputHeaders($buffer) { $lines = explode(self::CRLF, $buffer); foreach ($lines as $line) { if ($line == '') { $this->state = 'active'; common_log(LOG_DEBUG, "$this->id connection is active!"); } else { common_log(LOG_DEBUG, "$this->id read HTTP header: $line"); $this->responseHeaders[] = $line; } } } function handleInputActive($buffer) { // One JSON object on each line... // Will we always deliver on packet boundaries? $lines = explode("\n", $buffer); foreach ($lines as $line) { $data = json_decode($line, true); if ($data) { $this->handleJson($data); } else { common_log(LOG_ERR, "$this->id received bogus JSON data: " . $line); } } } abstract function handleJson(array $data); }