summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--classes/Notice.php12
-rw-r--r--classes/Status_network.php29
-rw-r--r--lib/liberalstomp.php27
-rw-r--r--lib/stompqueuemanager.php9
-rw-r--r--lib/util.php5
-rw-r--r--plugins/Facebook/FacebookPlugin.php2
-rw-r--r--plugins/Meteor/MeteorPlugin.php15
-rw-r--r--plugins/OStatus/OStatusPlugin.php8
-rw-r--r--plugins/OStatus/classes/HubSub.php51
-rw-r--r--plugins/OStatus/lib/ostatusqueuehandler.php31
-rw-r--r--plugins/RSSCloud/RSSCloudPlugin.php18
-rw-r--r--plugins/TwitterBridge/TwitterBridgePlugin.php2
12 files changed, 169 insertions, 40 deletions
diff --git a/classes/Notice.php b/classes/Notice.php
index 0838ca2a2..9ac9e10c1 100644
--- a/classes/Notice.php
+++ b/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);
+ }
+
}
diff --git a/classes/Status_network.php b/classes/Status_network.php
index a452c32ce..4a1f2c374 100644
--- a/classes/Status_network.php
+++ b/classes/Status_network.php
@@ -149,21 +149,15 @@ class Status_network extends Safe_DataObject
$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)) {
@@ -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)) {
diff --git a/lib/liberalstomp.php b/lib/liberalstomp.php
index 3d38953fd..70c22c17e 100644
--- a/lib/liberalstomp.php
+++ b/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);
+ }
+ }
+ }
diff --git a/lib/stompqueuemanager.php b/lib/stompqueuemanager.php
index de4ba7f01..91faa8c36 100644
--- a/lib/stompqueuemanager.php
+++ b/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);
diff --git a/lib/util.php b/lib/util.php
index 59d5132ec..049001aba 100644
--- a/lib/util.php
+++ b/lib/util.php
@@ -1235,9 +1235,8 @@ function common_enqueue_notice($notice)
$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';
diff --git a/plugins/Facebook/FacebookPlugin.php b/plugins/Facebook/FacebookPlugin.php
index 5dba73a5d..19989a952 100644
--- a/plugins/Facebook/FacebookPlugin.php
+++ b/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;
diff --git a/plugins/Meteor/MeteorPlugin.php b/plugins/Meteor/MeteorPlugin.php
index 5600d5fcc..ec8c9e217 100644
--- a/plugins/Meteor/MeteorPlugin.php
+++ b/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,8 +131,10 @@ class MeteorPlugin extends RealtimePlugin
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
diff --git a/plugins/OStatus/OStatusPlugin.php b/plugins/OStatus/OStatusPlugin.php
index 5b153216e..c61e2cc5f 100644
--- a/plugins/OStatus/OStatusPlugin.php
+++ b/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)
{
- // 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;
}
diff --git a/plugins/OStatus/classes/HubSub.php b/plugins/OStatus/classes/HubSub.php
index cdace3c1f..7db528a4e 100644
--- a/plugins/OStatus/classes/HubSub.php
+++ b/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
@@ -274,6 +305,26 @@ class HubSub extends Memcached_DataObject
}
/**
+ * 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.
*
diff --git a/plugins/OStatus/lib/ostatusqueuehandler.php b/plugins/OStatus/lib/ostatusqueuehandler.php
index d1e58f1d6..8905d2e21 100644
--- a/plugins/OStatus/lib/ostatusqueuehandler.php
+++ b/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()) {
- $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);
}
}
diff --git a/plugins/RSSCloud/RSSCloudPlugin.php b/plugins/RSSCloud/RSSCloudPlugin.php
index 661c32141..c1951cdbf 100644
--- a/plugins/RSSCloud/RSSCloudPlugin.php
+++ b/plugins/RSSCloud/RSSCloudPlugin.php
@@ -192,25 +192,13 @@ class RSSCloudPlugin extends Plugin
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
*
diff --git a/plugins/TwitterBridge/TwitterBridgePlugin.php b/plugins/TwitterBridge/TwitterBridgePlugin.php
index 1a0a69682..65b3a6b38 100644
--- a/plugins/TwitterBridge/TwitterBridgePlugin.php
+++ b/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');
generated by cgit v1.2.3-54-g00ecf (git 2.45.2) at 2024-07-05 06:41:02 +0000