Merge branch 'testing' of gitorious.org:statusnet/mainline into testing

actions/designadminpanel.php

@@ -125,9 +125,19 @@ class DesignadminpanelAction extends AdminPanelAction
             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');
 
@@ -139,15 +149,13 @@ class DesignadminpanelAction extends AdminPanelAction
 
         $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'));
@@ -189,6 +197,13 @@ class DesignadminpanelAction extends AdminPanelAction
             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');
     }
 
@@ -262,6 +277,33 @@ class DesignadminpanelAction extends AdminPanelAction
         }
     }
 
+    /**
+     * 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
      *
@@ -370,7 +412,15 @@ class DesignAdminPanelForm extends AdminForm
 
     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'));
 
@@ -383,6 +433,11 @@ class DesignAdminPanelForm extends AdminForm
         $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'));
 
@@ -406,10 +461,23 @@ class DesignAdminPanelForm extends AdminForm
                              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' =>
@@ -483,6 +551,11 @@ class DesignAdminPanelForm extends AdminForm
 
         $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'));
@@ -490,6 +563,7 @@ class DesignAdminPanelForm extends AdminForm
         $this->out->elementStart('ul', 'form_data');
 
         try {
+            // @fixme avoid loop unrolling in non-performance-critical contexts like this
 
             $bgcolor = new WebColor($design->backgroundcolor);
 
@@ -557,6 +631,7 @@ class DesignAdminPanelForm extends AdminForm
             $this->unli();
 
         } catch (WebColorException $e) {
+            // @fixme normalize them individually!
             common_log(LOG_ERR, 'Bad color values in site design: ' .
                 $e->getMessage());
         }
@@ -566,6 +641,27 @@ class DesignAdminPanelForm extends AdminForm
         $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
      *

classes/Notice.php

@@ -1863,4 +1863,16 @@ class Notice extends Memcached_DataObject
         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);
+    }
+
 }

classes/Status_network.php

@@ -152,18 +152,12 @@ class Status_network extends Safe_DataObject
     
     /**
      * @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)) {
@@ -182,6 +176,23 @@ class Status_network extends Safe_DataObject
                 }
             }
         }
+        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)) {
 

lib/action.php

@@ -233,6 +233,16 @@ class Action extends HTMLOutputter // lawsuit
                 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));
+                }
+            }
         }
     }
 

lib/adminpanelaction.php

@@ -283,9 +283,10 @@ class AdminPanelAction extends Action
                 $this->clientError(_("Unable to delete design setting."));
                 return null;
             }
+            return $result;
         }
 
-        return $result;
+        return null;
     }
 
     function canAdmin($name)

lib/default.php

@@ -141,10 +141,17 @@ $default =
               '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
@@ -260,6 +267,9 @@ $default =
               'linkcolor' => null,
               'backgroundimage' => null,
               'disposition' => null),
+        'custom_css' =>
+        array('enabled' => true,
+              'css' => ''),
         'notice' =>
         array('contentlimit' => null),
         'message' =>

lib/liberalstomp.php

@@ -147,5 +147,30 @@ class LiberalStomp extends Stomp
         }
         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);
+        }
+    }
+ }
 

lib/stompqueuemanager.php

@@ -115,11 +115,12 @@ class StompQueueManager extends QueueManager
      *
      * @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')) {
@@ -134,7 +135,7 @@ class StompQueueManager extends QueueManager
         } else {
             $idx = $this->defaultIdx;
         }
-        return $this->_doEnqueue($object, $queue, $idx);
+        return $this->_doEnqueue($object, $queue, $idx, $siteNickname);
     }
 
     /**
@@ -144,10 +145,10 @@ class StompQueueManager extends QueueManager
      * @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);

lib/theme.php

@@ -39,6 +39,9 @@ if (!defined('STATUSNET') && !defined('LACONICA')) {
  * 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.
  *
@@ -76,7 +79,7 @@ class Theme
 
         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;
         }
 
@@ -89,11 +92,32 @@ class Theme
         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
+     */
+
+    protected function relativeThemePath($group, $fallbackSubdir, $name)
+    {
+        $path = common_config($group, 'path');
 
         if (empty($path)) {
-                $path = common_config('site', 'path') . '/theme/';
+            $path = common_config('site', 'path') . '/';
+            if ($fallbackSubdir) {
+                $path .= $fallbackSubdir . '/';
+            }
         }
 
         if ($path[strlen($path)-1] != '/') {
@@ -104,17 +128,17 @@ class Theme
             $path = '/'.$path;
         }
 
-            $server = common_config('theme', 'server');
+        $server = common_config($group, 'server');
 
         if (empty($server)) {
             $server = common_config('site', 'server');
         }
 
-            $ssl = common_config('theme', 'ssl');
+        $ssl = common_config($group, 'ssl');
 
         if (is_null($ssl)) { // null -> guess
             if (common_config('site', 'ssl') == 'always' &&
-                    !common_config('theme', 'server')) {
+                !common_config($group, 'server')) {
                 $ssl = true;
             } else {
                 $ssl = false;
@@ -123,8 +147,8 @@ class Theme
 
         $protocol = ($ssl) ? 'https' : 'http';
 
-            $this->path = $protocol . '://'.$server.$path.$name;
+        $path = $protocol . '://'.$server.$path.$name;
-        }
+        return $path;
     }
 
     /**
@@ -236,7 +260,13 @@ class Theme
 
     protected static function localRoot()
     {
-        return INSTALLDIR.'/local/theme';
+        $basedir = common_config('local', 'dir');
+
+        if (empty($basedir)) {
+            $basedir = INSTALLDIR . '/local';
+        }
+
+        return $basedir . '/theme';
     }
 
     /**

lib/themeuploader.php

@@ -0,0 +1,311 @@
+<?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);
+    }
+
+}

lib/util.php

@@ -1235,9 +1235,8 @@ function common_enqueue_notice($notice)
         $transports[] = 'jabber';
     }
 
-    // @fixme move these checks into QueueManager and/or individual handlers
+    // We can skip these for gatewayed notices.
-    if ($notice->is_local == Notice::LOCAL_PUBLIC ||
+    if ($notice->isLocal()) {
-        $notice->is_local == Notice::LOCAL_NONPUBLIC) {
         $transports = array_merge($transports, $localTransports);
         if ($xmpp) {
             $transports[] = 'public';

plugins/Facebook/FacebookPlugin.php

@@ -585,7 +585,7 @@ class FacebookPlugin extends Plugin
 
     function onStartEnqueueNotice($notice, &$transports)
     {
-        if (self::hasKeys()) {
+        if (self::hasKeys() && $notice->isLocal()) {
             array_push($transports, 'facebook');
         }
         return true;

plugins/Meteor/MeteorPlugin.php

@@ -50,6 +50,7 @@ class MeteorPlugin extends RealtimePlugin
     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='')
@@ -102,8 +103,14 @@ class MeteorPlugin extends RealtimePlugin
     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}");
         }
@@ -124,9 +131,11 @@ class MeteorPlugin extends RealtimePlugin
 
     function _disconnect()
     {
+        if (!$this->persistent) {
             $cnt = fwrite($this->_socket, "QUIT\n");
             @fclose($this->_socket);
         }
+    }
 
     // Meteord flips out with default '/' separator
 

plugins/OStatus/OStatusPlugin.php

@@ -87,6 +87,8 @@ class OStatusPlugin extends Plugin
 
         // 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)
@@ -102,8 +104,10 @@ class OStatusPlugin extends Plugin
      */
     function onStartEnqueueNotice($notice, &$transports)
     {
+        if ($notice->isLocal()) {
             // put our transport first, in case there's any conflict (like OMB)
             array_unshift($transports, 'ostatus');
+        }
         return true;
     }
 

plugins/OStatus/classes/HubSub.php

@@ -260,6 +260,37 @@ class HubSub extends Memcached_DataObject
             $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
@@ -273,6 +304,26 @@ class HubSub extends Memcached_DataObject
         $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.

plugins/OStatus/lib/ostatusqueuehandler.php

@@ -25,6 +25,18 @@
  */
 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';
@@ -147,14 +159,31 @@ class OStatusQueueHandler extends QueueHandler
 
     /**
      * 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()) {
+            $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);
         }
     }
 

plugins/RSSCloud/RSSCloudPlugin.php

@@ -192,22 +192,10 @@ class RSSCloudPlugin extends Plugin
 
     function onStartEnqueueNotice($notice, &$transports)
     {
+        if ($notice->isLocal()) {
             array_push($transports, 'rsscloud');
-        return true;
         }
-
+        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);
     }
 
     /**

plugins/TwitterBridge/TwitterBridgePlugin.php

@@ -221,7 +221,7 @@ class TwitterBridgePlugin extends Plugin
      */
     function onStartEnqueueNotice($notice, &$transports)
     {
-        if (self::hasKeys()) {
+        if (self::hasKeys() && $notice->isLocal()) {
             // Avoid a possible loop
             if ($notice->source != 'twitter') {
                 array_push($transports, 'twitter');