return;
}
- // check for an image upload
+ // check for file uploads
$bgimage = $this->saveBackgroundImage();
+ $customTheme = $this->saveCustomTheme();
+
+ $oldtheme = common_config('site', 'theme');
+ if ($customTheme) {
+ // This feels pretty hacky :D
+ $this->args['theme'] = $customTheme;
+ $themeChanged = true;
+ } else {
+ $themeChanged = ($this->trimmed('theme') != $oldtheme);
+ }
static $settings = array('theme', 'logo');
$this->validate($values);
- $oldtheme = common_config('site', 'theme');
-
$config = new Config();
$config->query('BEGIN');
// Only update colors if the theme has not changed.
- if ($oldtheme == $values['theme']) {
+ if (!$themeChanged) {
$bgcolor = new WebColor($this->trimmed('design_background'));
$ccolor = new WebColor($this->trimmed('design_content'));
Config::save('design', 'backgroundimage', $bgimage);
}
+ if (common_config('custom_css', 'enabled')) {
+ $css = $this->arg('css');
+ if ($css != common_config('custom_css', 'css')) {
+ Config::save('custom_css', 'css', $css);
+ }
+ }
+
$config->query('COMMIT');
}
}
}
+ /**
+ * Save the custom theme if the user uploaded one.
+ *
+ * @return mixed custom theme name, if succesful, or null if no theme upload.
+ * @throws ClientException for invalid theme archives
+ * @throws ServerException if trouble saving the theme files
+ */
+
+ function saveCustomTheme()
+ {
+ if (common_config('theme_upload', 'enabled') &&
+ $_FILES['design_upload_theme']['error'] == UPLOAD_ERR_OK) {
+
+ $upload = ThemeUploader::fromUpload('design_upload_theme');
+ $basedir = common_config('local', 'dir');
+ if (empty($basedir)) {
+ $basedir = INSTALLDIR . '/local';
+ }
+ $name = 'custom'; // @todo allow multiples, custom naming?
+ $outdir = $basedir . '/theme/' . $name;
+ $upload->extract($outdir);
+ return $name;
+ } else {
+ return null;
+ }
+ }
+
/**
* Attempt to validate setting values
*
function formData()
{
+ $this->showLogo();
+ $this->showTheme();
+ $this->showBackground();
+ $this->showColors();
+ $this->showAdvanced();
+ }
+ function showLogo()
+ {
$this->out->elementStart('fieldset', array('id' => 'settings_design_logo'));
$this->out->element('legend', null, _('Change logo'));
$this->out->elementEnd('ul');
$this->out->elementEnd('fieldset');
+
+ }
+
+ function showTheme()
+ {
$this->out->elementStart('fieldset', array('id' => 'settings_design_theme'));
$this->out->element('legend', null, _('Change theme'));
false, $this->value('theme'));
$this->unli();
+ if (common_config('theme_upload', 'enabled')) {
+ $this->li();
+ $this->out->element('label', array('for' => 'design_upload_theme'), _('Custom theme'));
+ $this->out->element('input', array('id' => 'design_upload_theme',
+ 'name' => 'design_upload_theme',
+ 'type' => 'file'));
+ $this->out->element('p', 'form_guide', _('You can upload a custom StatusNet theme as a .ZIP archive.'));
+ $this->unli();
+ }
+
$this->out->elementEnd('ul');
$this->out->elementEnd('fieldset');
+ }
+ function showBackground()
+ {
$design = $this->out->design;
$this->out->elementStart('fieldset', array('id' =>
$this->out->elementEnd('ul');
$this->out->elementEnd('fieldset');
+ }
+
+ function showColors()
+ {
+ $design = $this->out->design;
$this->out->elementStart('fieldset', array('id' => 'settings_design_color'));
$this->out->element('legend', null, _('Change colours'));
$this->out->elementStart('ul', 'form_data');
try {
+ // @fixme avoid loop unrolling in non-performance-critical contexts like this
$bgcolor = new WebColor($design->backgroundcolor);
$this->unli();
} catch (WebColorException $e) {
+ // @fixme normalize them individually!
common_log(LOG_ERR, 'Bad color values in site design: ' .
$e->getMessage());
}
$this->out->elementEnd('ul');
}
+ function showAdvanced()
+ {
+ if (common_config('custom_css', 'enabled')) {
+ $this->out->elementStart('fieldset', array('id' => 'settings_design_advanced'));
+ $this->out->element('legend', null, _('Advanced'));
+ $this->out->elementStart('ul', 'form_data');
+
+ $this->li();
+ $this->out->element('label', array('for' => 'css'), _('Custom CSS'));
+ $this->out->element('textarea', array('name' => 'css',
+ 'id' => 'css',
+ 'cols' => '50',
+ 'rows' => '10'),
+ strval(common_config('custom_css', 'css')));
+ $this->unli();
+
+ $this->out->elementEnd('fieldset');
+ $this->out->elementEnd('ul');
+ }
+ }
+
/**
* Action elements
*
return $ns;
}
+ /**
+ * Determine whether the notice was locally created
+ *
+ * @return boolean locality
+ */
+
+ public function isLocal()
+ {
+ return ($this->is_local == Notice::LOCAL_PUBLIC ||
+ $this->is_local == Notice::LOCAL_NONPUBLIC);
+ }
+
}
$this->decache(); # while we still have the values!
return parent::delete();
}
-
+
/**
* @param string $servername hostname
- * @param string $pathname URL base path
* @param string $wildcard hostname suffix to match wildcard config
+ * @return mixed Status_network or null
*/
- static function setupSite($servername, $pathname, $wildcard)
+ static function getFromHostname($servername, $wildcard)
{
- global $config;
-
$sn = null;
-
- // XXX I18N, probably not crucial for hostnames
- // XXX This probably needs a tune up
-
if (0 == strncasecmp(strrev($wildcard), strrev($servername), strlen($wildcard))) {
// special case for exact match
if (0 == strcasecmp($servername, $wildcard)) {
}
}
}
+ return $sn;
+ }
+
+ /**
+ * @param string $servername hostname
+ * @param string $pathname URL base path
+ * @param string $wildcard hostname suffix to match wildcard config
+ */
+ static function setupSite($servername, $pathname, $wildcard)
+ {
+ global $config;
+
+ $sn = null;
+
+ // XXX I18N, probably not crucial for hostnames
+ // XXX This probably needs a tune up
+ $sn = self::getFromHostname($servername, $wildcard);
if (!empty($sn)) {
Event::handle('EndShowDesign', array($this));
}
Event::handle('EndShowStyles', array($this));
+
+ if (common_config('custom_css', 'enabled')) {
+ $css = common_config('custom_css', 'css');
+ if (Event::handle('StartShowCustomCss', array($this, &$css))) {
+ if (trim($css) != '') {
+ $this->style($css);
+ }
+ Event::handle('EndShowCustomCss', array($this));
+ }
+ }
}
}
$this->clientError(_("Unable to delete design setting."));
return null;
}
+ return $result;
}
- return $result;
+ return null;
}
function canAdmin($name)
'dir' => null,
'path'=> null,
'ssl' => null),
+ 'theme_upload' =>
+ array('enabled' => extension_loaded('zip')),
'javascript' =>
array('server' => null,
'path'=> null,
'ssl' => null),
+ 'local' => // To override path/server for themes in 'local' dir (not currently applied to local plugins)
+ array('server' => null,
+ 'dir' => null,
+ 'path' => null,
+ 'ssl' => null),
'throttle' =>
array('enabled' => false, // whether to throttle edits; false by default
'count' => 20, // number of allowed messages in timespan
'linkcolor' => null,
'backgroundimage' => null,
'disposition' => null),
+ 'custom_css' =>
+ array('enabled' => true,
+ 'css' => ''),
'notice' =>
array('contentlimit' => null),
'message' =>
}
return $frame;
}
-}
+
+ /**
+ * Write frame to server
+ *
+ * @param StompFrame $stompFrame
+ */
+ protected function _writeFrame (StompFrame $stompFrame)
+ {
+ if (!is_resource($this->_socket)) {
+ require_once 'Stomp/Exception.php';
+ throw new StompException('Socket connection hasn\'t been established');
+ }
+
+ $data = $stompFrame->__toString();
+
+ // Make sure the socket's in a writable state; if not, wait a bit.
+ stream_set_blocking($this->_socket, 1);
+
+ $r = fwrite($this->_socket, $data, strlen($data));
+ stream_set_blocking($this->_socket, 0);
+ if ($r === false || $r == 0) {
+ $this->_reconnect();
+ $this->_writeFrame($stompFrame);
+ }
+ }
+ }
*
* @param mixed $object
* @param string $queue
+ * @param string $siteNickname optional override to drop into another site's queue
*
* @return boolean true on success
* @throws StompException on connection or send error
*/
- public function enqueue($object, $queue)
+ public function enqueue($object, $queue, $siteNickname=null)
{
$this->_connect();
if (common_config('queue', 'stomp_enqueue_on')) {
} else {
$idx = $this->defaultIdx;
}
- return $this->_doEnqueue($object, $queue, $idx);
+ return $this->_doEnqueue($object, $queue, $idx, $siteNickname);
}
/**
* @return boolean true on success
* @throws StompException on connection or send error
*/
- protected function _doEnqueue($object, $queue, $idx)
+ protected function _doEnqueue($object, $queue, $idx, $siteNickname=null)
{
$rep = $this->logrep($object);
- $envelope = array('site' => common_config('site', 'nickname'),
+ $envelope = array('site' => $siteNickname ? $siteNickname : common_config('site', 'nickname'),
'handler' => $queue,
'payload' => $this->encode($object));
$msg = serialize($envelope);
* Themes are directories with some expected sub-directories and files
* in them. They're found in either local/theme (for locally-installed themes)
* or theme/ subdir of installation dir.
+ *
+ * Note that the 'local' directory can be overridden as $config['local']['path']
+ * and $config['local']['dir'] etc.
*
* This used to be a couple of functions, but for various reasons it's nice
* to have a class instead.
if (file_exists($fulldir) && is_dir($fulldir)) {
$this->dir = $fulldir;
- $this->path = common_path('local/theme/'.$name.'/');
+ $this->path = $this->relativeThemePath('local', 'local', 'theme/' . $name);
return;
}
if (file_exists($fulldir) && is_dir($fulldir)) {
$this->dir = $fulldir;
+ $this->path = $this->relativeThemePath('theme', 'theme', $name);
+ }
+ }
- $path = common_config('theme', 'path');
+ /**
+ * Build a full URL to the given theme's base directory, possibly
+ * using an offsite theme server path.
+ *
+ * @param string $group configuration section name to pull paths from
+ * @param string $fallbackSubdir default subdirectory under INSTALLDIR
+ * @param string $name theme name
+ *
+ * @return string URL
+ *
+ * @todo consolidate code with that for other customizable paths
+ */
- if (empty($path)) {
- $path = common_config('site', 'path') . '/theme/';
- }
+ protected function relativeThemePath($group, $fallbackSubdir, $name)
+ {
+ $path = common_config($group, 'path');
- if ($path[strlen($path)-1] != '/') {
- $path .= '/';
+ if (empty($path)) {
+ $path = common_config('site', 'path') . '/';
+ if ($fallbackSubdir) {
+ $path .= $fallbackSubdir . '/';
}
+ }
- if ($path[0] != '/') {
- $path = '/'.$path;
- }
+ if ($path[strlen($path)-1] != '/') {
+ $path .= '/';
+ }
- $server = common_config('theme', 'server');
+ if ($path[0] != '/') {
+ $path = '/'.$path;
+ }
- if (empty($server)) {
- $server = common_config('site', 'server');
- }
+ $server = common_config($group, 'server');
- $ssl = common_config('theme', 'ssl');
+ if (empty($server)) {
+ $server = common_config('site', 'server');
+ }
- if (is_null($ssl)) { // null -> guess
- if (common_config('site', 'ssl') == 'always' &&
- !common_config('theme', 'server')) {
- $ssl = true;
- } else {
- $ssl = false;
- }
+ $ssl = common_config($group, 'ssl');
+
+ if (is_null($ssl)) { // null -> guess
+ if (common_config('site', 'ssl') == 'always' &&
+ !common_config($group, 'server')) {
+ $ssl = true;
+ } else {
+ $ssl = false;
}
+ }
- $protocol = ($ssl) ? 'https' : 'http';
+ $protocol = ($ssl) ? 'https' : 'http';
- $this->path = $protocol . '://'.$server.$path.$name;
- }
+ $path = $protocol . '://'.$server.$path.$name;
+ return $path;
}
/**
protected static function localRoot()
{
- return INSTALLDIR.'/local/theme';
+ $basedir = common_config('local', 'dir');
+
+ if (empty($basedir)) {
+ $basedir = INSTALLDIR . '/local';
+ }
+
+ return $basedir . '/theme';
}
/**
--- /dev/null
+<?php
+/**
+ * StatusNet, the distributed open-source microblogging tool
+ *
+ * Utilities for theme files and paths
+ *
+ * PHP version 5
+ *
+ * LICENCE: This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ *
+ * @category Paths
+ * @package StatusNet
+ * @author Brion Vibber <brion@status.net>
+ * @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/
+ */
+
+if (!defined('STATUSNET') && !defined('LACONICA')) {
+ exit(1);
+}
+
+/**
+ * Encapsulation of the validation-and-save process when dealing with
+ * a user-uploaded StatusNet theme archive...
+ *
+ * @todo extract theme metadata from css/display.css
+ * @todo allow saving multiple themes
+ */
+class ThemeUploader
+{
+ protected $sourceFile;
+ protected $isUpload;
+ private $prevErrorReporting;
+
+ public function __construct($filename)
+ {
+ if (!class_exists('ZipArchive')) {
+ throw new Exception(_("This server cannot handle theme uploads without ZIP support."));
+ }
+ $this->sourceFile = $filename;
+ }
+
+ public static function fromUpload($name)
+ {
+ if (!isset($_FILES[$name]['error'])) {
+ throw new ServerException(_("Theme upload missing or failed."));
+ }
+ if ($_FILES[$name]['error'] != UPLOAD_ERR_OK) {
+ throw new ServerException(_("Theme upload missing or failed."));
+ }
+ return new ThemeUploader($_FILES[$name]['tmp_name']);
+ }
+
+ /**
+ * @param string $destDir
+ * @throws Exception on bogus files
+ */
+ public function extract($destDir)
+ {
+ $zip = $this->openArchive();
+
+ // First pass: validate but don't save anything to disk.
+ // Any errors will trip an exception.
+ $this->traverseArchive($zip);
+
+ // Second pass: now that we know we're good, actually extract!
+ $tmpDir = $destDir . '.tmp' . getmypid();
+ $this->traverseArchive($zip, $tmpDir);
+
+ $zip->close();
+
+ if (file_exists($destDir)) {
+ $killDir = $tmpDir . '.old';
+ $this->quiet();
+ $ok = rename($destDir, $killDir);
+ $this->loud();
+ if (!$ok) {
+ common_log(LOG_ERR, "Could not move old custom theme from $destDir to $killDir");
+ throw new ServerException(_("Failed saving theme."));
+ }
+ } else {
+ $killDir = false;
+ }
+
+ $this->quiet();
+ $ok = rename($tmpDir, $destDir);
+ $this->loud();
+ if (!$ok) {
+ common_log(LOG_ERR, "Could not move saved theme from $tmpDir to $destDir");
+ throw new ServerException(_("Failed saving theme."));
+ }
+
+ if ($killDir) {
+ $this->recursiveRmdir($killDir);
+ }
+ }
+
+ /**
+ *
+ */
+ protected function traverseArchive($zip, $outdir=false)
+ {
+ $sizeLimit = 2 * 1024 * 1024; // 2 megabyte space limit?
+ $blockSize = 4096; // estimated; any entry probably takes this much space
+
+ $totalSize = 0;
+ $hasMain = false;
+ $commonBaseDir = false;
+
+ for ($i = 0; $i < $zip->numFiles; $i++) {
+ $data = $zip->statIndex($i);
+ $name = str_replace('\\', '/', $data['name']);
+
+ if (substr($name, -1) == '/') {
+ // A raw directory... skip!
+ continue;
+ }
+
+ // Check the directory structure...
+ $path = pathinfo($name);
+ $dirs = explode('/', $path['dirname']);
+ $baseDir = array_shift($dirs);
+ if ($commonBaseDir === false) {
+ $commonBaseDir = $baseDir;
+ } else {
+ if ($commonBaseDir != $baseDir) {
+ throw new ClientException(_("Invalid theme: bad directory structure."));
+ }
+ }
+
+ foreach ($dirs as $dir) {
+ $this->validateFileOrFolder($dir);
+ }
+
+ // Is this a safe or skippable file?
+ if ($this->skippable($path['filename'], $path['extension'])) {
+ // Documentation and such... booooring
+ continue;
+ } else {
+ $this->validateFile($path['filename'], $path['extension']);
+ }
+
+ $fullPath = $dirs;
+ $fullPath[] = $path['basename'];
+ $localFile = implode('/', $fullPath);
+ if ($localFile == 'css/display.css') {
+ $hasMain = true;
+ }
+
+ $size = $data['size'];
+ $estSize = $blockSize * max(1, intval(ceil($size / $blockSize)));
+ $totalSize += $estSize;
+ if ($totalSize > $sizeLimit) {
+ $msg = sprintf(_("Uploaded theme is too large; " .
+ "must be less than %d bytes uncompressed."),
+ $sizeLimit);
+ throw new ClientException($msg);
+ }
+
+ if ($outdir) {
+ $this->extractFile($zip, $data['name'], "$outdir/$localFile");
+ }
+ }
+
+ if (!$hasMain) {
+ throw new ClientException(_("Invalid theme archive: " .
+ "missing file css/display.css"));
+ }
+ }
+
+ protected function skippable($filename, $ext)
+ {
+ $skip = array('txt', 'rtf', 'doc', 'docx', 'odt');
+ if (strtolower($filename) == 'readme') {
+ return true;
+ }
+ if (in_array(strtolower($ext), $skip)) {
+ return true;
+ }
+ return false;
+ }
+
+ protected function validateFile($filename, $ext)
+ {
+ $this->validateFileOrFolder($filename);
+ $this->validateExtension($ext);
+ // @fixme validate content
+ }
+
+ protected function validateFileOrFolder($name)
+ {
+ if (!preg_match('/^[a-z0-9_-]+$/i', $name)) {
+ $msg = _("Theme contains invalid file or folder name. " .
+ "Stick with ASCII letters, digits, underscore, and minus sign.");
+ throw new ClientException($msg);
+ }
+ return true;
+ }
+
+ protected function validateExtension($ext)
+ {
+ $allowed = array('css', 'png', 'gif', 'jpg', 'jpeg');
+ if (!in_array(strtolower($ext), $allowed)) {
+ $msg = sprintf(_("Theme contains file of type '.%s', " .
+ "which is not allowed."),
+ $ext);
+ throw new ClientException($msg);
+ }
+ return true;
+ }
+
+ /**
+ * @return ZipArchive
+ */
+ protected function openArchive()
+ {
+ $zip = new ZipArchive;
+ $ok = $zip->open($this->sourceFile);
+ if ($ok !== true) {
+ common_log(LOG_ERR, "Error opening theme zip archive: " .
+ "{$this->sourceFile} code: {$ok}");
+ throw new Exception(_("Error opening theme archive."));
+ }
+ return $zip;
+ }
+
+ /**
+ * @param ZipArchive $zip
+ * @param string $from original path inside ZIP archive
+ * @param string $to final destination path in filesystem
+ */
+ protected function extractFile($zip, $from, $to)
+ {
+ $dir = dirname($to);
+ if (!file_exists($dir)) {
+ $this->quiet();
+ $ok = mkdir($dir, 0755, true);
+ $this->loud();
+ if (!$ok) {
+ common_log(LOG_ERR, "Failed to mkdir $dir while uploading theme");
+ throw new ServerException(_("Failed saving theme."));
+ }
+ } else if (!is_dir($dir)) {
+ common_log(LOG_ERR, "Output directory $dir not a directory while uploading theme");
+ throw new ServerException(_("Failed saving theme."));
+ }
+
+ // ZipArchive::extractTo would be easier, but won't let us alter
+ // the directory structure.
+ $in = $zip->getStream($from);
+ if (!$in) {
+ common_log(LOG_ERR, "Couldn't open archived file $from while uploading theme");
+ throw new ServerException(_("Failed saving theme."));
+ }
+ $this->quiet();
+ $out = fopen($to, "wb");
+ $this->loud();
+ if (!$out) {
+ common_log(LOG_ERR, "Couldn't open output file $to while uploading theme");
+ throw new ServerException(_("Failed saving theme."));
+ }
+ while (!feof($in)) {
+ $buffer = fread($in, 65536);
+ fwrite($out, $buffer);
+ }
+ fclose($in);
+ fclose($out);
+ }
+
+ private function quiet()
+ {
+ $this->prevErrorReporting = error_reporting();
+ error_reporting($this->prevErrorReporting & ~E_WARNING);
+ }
+
+ private function loud()
+ {
+ error_reporting($this->prevErrorReporting);
+ }
+
+ private function recursiveRmdir($dir)
+ {
+ $list = dir($dir);
+ while (($file = $list->read()) !== false) {
+ if ($file == '.' || $file == '..') {
+ continue;
+ }
+ $full = "$dir/$file";
+ if (is_dir($full)) {
+ $this->recursiveRmdir($full);
+ } else {
+ unlink($full);
+ }
+ }
+ $list->close();
+ rmdir($dir);
+ }
+
+}
$transports[] = 'jabber';
}
- // @fixme move these checks into QueueManager and/or individual handlers
- if ($notice->is_local == Notice::LOCAL_PUBLIC ||
- $notice->is_local == Notice::LOCAL_NONPUBLIC) {
+ // We can skip these for gatewayed notices.
+ if ($notice->isLocal()) {
$transports = array_merge($transports, $localTransports);
if ($xmpp) {
$transports[] = 'public';
function onStartEnqueueNotice($notice, &$transports)
{
- if (self::hasKeys()) {
+ if (self::hasKeys() && $notice->isLocal()) {
array_push($transports, 'facebook');
}
return true;
public $controlport = null;
public $controlserver = null;
public $channelbase = null;
+ public $persistent = true;
protected $_socket = null;
function __construct($webserver=null, $webport=4670, $controlport=4671, $controlserver=null, $channelbase='')
function _connect()
{
$controlserver = (empty($this->controlserver)) ? $this->webserver : $this->controlserver;
+
+ $errno = $errstr = null;
+ $timeout = 5;
+ $flags = STREAM_CLIENT_CONNECT;
+ if ($this->persistent) $flags |= STREAM_CLIENT_PERSISTENT;
+
// May throw an exception.
- $this->_socket = stream_socket_client("tcp://{$controlserver}:{$this->controlport}");
+ $this->_socket = stream_socket_client("tcp://{$controlserver}:{$this->controlport}", $errno, $errstr, $timeout, $flags);
if (!$this->_socket) {
throw new Exception("Couldn't connect to {$controlserver} on {$this->controlport}");
}
function _disconnect()
{
- $cnt = fwrite($this->_socket, "QUIT\n");
- @fclose($this->_socket);
+ if (!$this->persistent) {
+ $cnt = fwrite($this->_socket, "QUIT\n");
+ @fclose($this->_socket);
+ }
}
// Meteord flips out with default '/' separator
// Outgoing from our internal PuSH hub
$qm->connect('hubconf', 'HubConfQueueHandler');
+ $qm->connect('hubprep', 'HubPrepQueueHandler');
+
$qm->connect('hubout', 'HubOutQueueHandler');
// Outgoing Salmon replies (when we don't need a return value)
*/
function onStartEnqueueNotice($notice, &$transports)
{
- // put our transport first, in case there's any conflict (like OMB)
- array_unshift($transports, 'ostatus');
+ if ($notice->isLocal()) {
+ // put our transport first, in case there's any conflict (like OMB)
+ array_unshift($transports, 'ostatus');
+ }
return true;
}
$retries = intval(common_config('ostatus', 'hub_retries'));
}
+ if (common_config('ostatus', 'local_push_bypass')) {
+ // If target is a local site, bypass the web server and drop the
+ // item directly into the target's input queue.
+ $url = parse_url($this->callback);
+ $wildcard = common_config('ostatus', 'local_wildcard');
+ $site = Status_network::getFromHostname($url['host'], $wildcard);
+
+ if ($site) {
+ if ($this->secret) {
+ $hmac = 'sha1=' . hash_hmac('sha1', $atom, $this->secret);
+ } else {
+ $hmac = '';
+ }
+
+ // Hack: at the moment we stick the subscription ID in the callback
+ // URL so we don't have to look inside the Atom to route the subscription.
+ // For now this means we need to extract that from the target URL
+ // so we can include it in the data.
+ $parts = explode('/', $url['path']);
+ $subId = intval(array_pop($parts));
+
+ $data = array('feedsub_id' => $subId,
+ 'post' => $atom,
+ 'hmac' => $hmac);
+ common_log(LOG_DEBUG, "Cross-site PuSH bypass enqueueing straight to $site->nickname feed $subId");
+ $qm = QueueManager::get();
+ $qm->enqueue($data, 'pushin', $site->nickname);
+ return;
+ }
+ }
+
// We dare not clone() as when the clone is discarded it'll
// destroy the result data for the parent query.
// @fixme use clone() again when it's safe to copy an
$qm->enqueue($data, 'hubout');
}
+ /**
+ * Queue up a large batch of pushes to multiple subscribers
+ * for this same topic update.
+ *
+ * If queues are disabled, this will run immediately.
+ *
+ * @param string $atom well-formed Atom feed
+ * @param array $pushCallbacks list of callback URLs
+ */
+ function bulkDistribute($atom, $pushCallbacks)
+ {
+ $data = array('atom' => $atom,
+ 'topic' => $this->topic,
+ 'pushCallbacks' => $pushCallbacks);
+ common_log(LOG_INFO, "Queuing PuSH batch: $this->topic to " .
+ count($pushCallbacks) . " sites");
+ $qm = QueueManager::get();
+ $qm->enqueue($data, 'hubprep');
+ }
+
/**
* Send a 'fat ping' to the subscriber's callback endpoint
* containing the given Atom feed chunk.
*/
class OStatusQueueHandler extends QueueHandler
{
+ // If we have more than this many subscribing sites on a single feed,
+ // break up the PuSH distribution into smaller batches which will be
+ // rolled into the queue progressively. This reduces disruption to
+ // other, shorter activities being enqueued while we work.
+ const MAX_UNBATCHED = 50;
+
+ // Each batch (a 'hubprep' entry) will have this many items.
+ // Selected to provide a balance between queue packet size
+ // and number of batches that will end up getting processed.
+ // For 20,000 target sites, 1000 should work acceptably.
+ const BATCH_SIZE = 1000;
+
function transport()
{
return 'ostatus';
/**
* Queue up direct feed update pushes to subscribers on our internal hub.
+ * If there are a large number of subscriber sites, intermediate bulk
+ * distribution triggers may be queued.
+ *
* @param string $atom update feed, containing only new/changed items
* @param HubSub $sub open query of subscribers
*/
function pushFeedInternal($atom, $sub)
{
common_log(LOG_INFO, "Preparing $sub->N PuSH distribution(s) for $sub->topic");
+ $n = 0;
+ $batch = array();
while ($sub->fetch()) {
- $sub->distribute($atom);
+ $n++;
+ if ($n < self::MAX_UNBATCHED) {
+ $sub->distribute($atom);
+ } else {
+ $batch[] = $sub->callback;
+ if (count($batch) >= self::BATCH_SIZE) {
+ $sub->bulkDistribute($atom, $batch);
+ $batch = array();
+ }
+ }
+ }
+ if (count($batch) >= 0) {
+ $sub->bulkDistribute($atom, $batch);
}
}
function onStartEnqueueNotice($notice, &$transports)
{
- array_push($transports, 'rsscloud');
+ if ($notice->isLocal()) {
+ array_push($transports, 'rsscloud');
+ }
return true;
}
- /**
- * Determine whether the notice was locally created
- *
- * @param Notice $notice the notice in question
- *
- * @return boolean locality
- */
-
- function _isLocal($notice)
- {
- return ($notice->is_local == Notice::LOCAL_PUBLIC ||
- $notice->is_local == Notice::LOCAL_NONPUBLIC);
- }
-
/**
* Create the rsscloud_subscription table if it's not
* already in the DB
*/
function onStartEnqueueNotice($notice, &$transports)
{
- if (self::hasKeys()) {
+ if (self::hasKeys() && $notice->isLocal()) {
// Avoid a possible loop
if ($notice->source != 'twitter') {
array_push($transports, 'twitter');
projects / quix0rs-gnu-social.git / commitdiff