Merge branch '0.9.x' into 1.0.x

5 files changed, 138 insertions, 80 deletions

diff --git a/plugins/OStatus/lib/hubprepqueuehandler.php b/plugins/OStatus/lib/hubprepqueuehandler.php
new file mode 100644
index 000000000..0d585938f
--- /dev/null
+++ b/plugins/OStatus/lib/hubprepqueuehandler.php
@@ -0,0 +1,87 @@
+<?php
+/*
+ * StatusNet - the distributed open-source microblogging tool
+ * Copyright (C) 2010, StatusNet, Inc.
+ *
+ * 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/>.
+ */
+
+/**
+ * When we have a large batch of PuSH consumers, we break the data set
+ * into smaller chunks. Enqueue final destinations...
+ *
+ * @package Hub
+ * @author Brion Vibber <brion@status.net>
+ */
+class HubPrepQueueHandler extends QueueHandler
+{
+    // Enqueue this many low-level distributions before re-queueing the rest
+    // of the batch to be processed later. Helps to keep latency down for other
+    // things happening during a particularly long OStatus delivery session.
+    //
+    // [Could probably ditch this if we had working message delivery priorities
+    // for queueing, but this isn't supported in ActiveMQ 5.3.]
+    const ROLLING_BATCH = 20;
+
+    function transport()
+    {
+        return 'hubprep';
+    }
+
+    function handle($data)
+    {
+        $topic = $data['topic'];
+        $atom = $data['atom'];
+        $pushCallbacks = $data['pushCallbacks'];
+
+        assert(is_string($atom));
+        assert(is_string($topic));
+        assert(is_array($pushCallbacks));
+
+        // Set up distribution for the first n subscribing sites...
+        // If we encounter an uncatchable error, queue handling should
+        // automatically re-run the batch, which could lead to some dupe
+        // distributions.
+        //
+        // Worst case is if one of these hubprep entries dies too many
+        // times and gets dropped; the rest of the batch won't get processed.
+        try {
+            $n = 0;
+            while (count($pushCallbacks) && $n < self::ROLLING_BATCH) {
+                $n++;
+                $callback = array_shift($pushCallbacks);
+                $sub = HubSub::staticGet($topic, $callback);
+                if (!$sub) {
+                    common_log(LOG_ERR, "Skipping PuSH delivery for deleted(?) consumer $callback on $topic");
+                    continue;
+                }
+
+                $sub->distribute($atom);
+            }
+        } catch (Exception $e) {
+            common_log(LOG_ERR, "Exception during PuSH batch out: " .
+                                $e->getMessage() .
+                                " prepping $topic to $callback");
+        }
+
+        // And re-queue the rest of the batch!
+        if (count($pushCallbacks) > 0) {
+            $sub = new HubSub();
+            $sub->topic = $topic;
+            $sub->bulkDistribute($atom, $pushCallbacks);
+        }
+
+        return true;
+    }
+}
diff --git a/plugins/OStatus/lib/magicenvelope.php b/plugins/OStatus/lib/magicenvelope.php
index f39686b71..3bdf24b31 100644
--- a/plugins/OStatus/lib/magicenvelope.php
+++ b/plugins/OStatus/lib/magicenvelope.php
@@ -97,24 +97,18 @@ class MagicEnvelope
     }
 
     public function toXML($env) {
-        $dom = new DOMDocument();
-
-        $envelope = $dom->createElementNS(MagicEnvelope::NS, 'me:env');
-        $envelope->setAttribute('xmlns:me', MagicEnvelope::NS);
-        $data = $dom->createElementNS(MagicEnvelope::NS, 'me:data', $env['data']);
-        $data->setAttribute('type', $env['data_type']);
-        $envelope->appendChild($data);
-        $enc = $dom->createElementNS(MagicEnvelope::NS, 'me:encoding', $env['encoding']);
-        $envelope->appendChild($enc);
-        $alg = $dom->createElementNS(MagicEnvelope::NS, 'me:alg', $env['alg']);
-        $envelope->appendChild($alg);
-        $sig = $dom->createElementNS(MagicEnvelope::NS, 'me:sig', $env['sig']);
-        $envelope->appendChild($sig);
-
-        $dom->appendChild($envelope);
+        $xs = new XMLStringer();
+        $xs->startXML();
+        $xs->elementStart('me:env', array('xmlns:me' => MagicEnvelope::NS));
+        $xs->element('me:data', array('type' => $env['data_type']), $env['data']);
+        $xs->element('me:encoding', null, $env['encoding']);
+        $xs->element('me:alg', null, $env['alg']);
+        $xs->element('me:sig', null, $env['sig']);
+        $xs->elementEnd('me:env');
         
-        
-        return $dom->saveXML();
+        $string =  $xs->getString();
+        common_debug($string);
+        return $string;
     }
 
     
diff --git a/plugins/OStatus/lib/salmon.php b/plugins/OStatus/lib/salmon.php
index 3d3341bc6..ef7719a40 100644
--- a/plugins/OStatus/lib/salmon.php
+++ b/plugins/OStatus/lib/salmon.php
@@ -28,9 +28,11 @@
  */
 class Salmon
 {
+    const REL_SALMON = 'salmon';
+    const REL_MENTIONED = 'mentioned';
 
+    // XXX: these are deprecated 
     const NS_REPLIES = "http://salmon-protocol.org/ns/salmon-replies";
-
     const NS_MENTIONS = "http://salmon-protocol.org/ns/salmon-mention";
     
     /**
diff --git a/plugins/OStatus/lib/xrd.php b/plugins/OStatus/lib/xrd.php
index 34b28790b..a10b9f427 100644
--- a/plugins/OStatus/lib/xrd.php
+++ b/plugins/OStatus/lib/xrd.php
@@ -106,44 +106,43 @@ class XRD
 
     public function toXML()
     {
-        $dom = new DOMDocument('1.0', 'UTF-8');
-        $dom->formatOutput = true;
-        
-        $xrd_dom = $dom->createElementNS(XRD::XRD_NS, 'XRD');
-        $dom->appendChild($xrd_dom);
+        $xs = new XMLStringer();
+
+        $xs->startXML();
+        $xs->elementStart('XRD', array('xmlns' => XRD::XRD_NS));
 
         if ($this->host) {
-            $host_dom = $dom->createElement('hm:Host', $this->host);
-            $xrd_dom->setAttributeNS(XRD::XML_NS, 'xmlns:hm', XRD::HOST_META_NS);
-            $xrd_dom->appendChild($host_dom);
+            $xs->element('hm:Host', array('xmlns:hm' => XRD::HOST_META_NS), $this->host);
         }
         
-		if ($this->expires) {
-			$expires_dom = $dom->createElement('Expires', $this->expires);
-			$xrd_dom->appendChild($expires_dom);
-		}
-
-		if ($this->subject) {
-			$subject_dom = $dom->createElement('Subject', $this->subject);
-			$xrd_dom->appendChild($subject_dom);
-		}
-
-		foreach ($this->alias as $alias) {
-			$alias_dom = $dom->createElement('Alias', $alias);
-			$xrd_dom->appendChild($alias_dom);
-		}
-
-		foreach ($this->types as $type) {
-			$type_dom = $dom->createElement('Type', $type);
-			$xrd_dom->appendChild($type_dom);
-		}
-
-		foreach ($this->links as $link) {
-			$link_dom = $this->saveLink($dom, $link);
-			$xrd_dom->appendChild($link_dom);
-		}
-
-        return $dom->saveXML();
+        if ($this->expires) {
+            $xs->element('Expires', null, $this->expires);
+        }
+
+        if ($this->subject) {
+            $xs->element('Subject', null, $this->subject);
+        }
+
+        foreach ($this->alias as $alias) {
+            $xs->element('Alias', null, $alias);
+        }
+
+        foreach ($this->links as $link) {
+            $titles = array();
+            if (isset($link['title'])) {
+                $titles = $link['title'];
+                unset($link['title']);
+            }
+            $xs->elementStart('Link', $link);
+            foreach ($titles as $title) {
+                $xs->element('Title', null, $title);
+            }
+            $xs->elementEnd('Link');
+        }
+        
+        $xs->elementEnd('XRD');
+
+        return $xs->getString();
     }
 
     function parseType($element)
@@ -169,32 +168,5 @@ class XRD
 
         return $link;
     }
-
-    function saveLink($doc, $link)
-    {
-        $link_element = $doc->createElement('Link');
-        if (!empty($link['rel'])) {
-            $link_element->setAttribute('rel', $link['rel']);
-        }
-        if (!empty($link['type'])) {
-            $link_element->setAttribute('type', $link['type']);
-        }
-        if (!empty($link['href'])) {
-            $link_element->setAttribute('href', $link['href']);
-        }
-        if (!empty($link['template'])) {
-            $link_element->setAttribute('template', $link['template']);
-        }
-
-        if (!empty($link['title']) && is_array($link['title'])) {
-            foreach($link['title'] as $title) {
-                $title = $doc->createElement('Title', $title);
-                $link_element->appendChild($title);
-            }
-        }
-
-        
-        return $link_element;
-    }
 }
 
diff --git a/plugins/OStatus/lib/xrdaction.php b/plugins/OStatus/lib/xrdaction.php
index f1a56e0a8..d8cf648d6 100644
--- a/plugins/OStatus/lib/xrdaction.php
+++ b/plugins/OStatus/lib/xrdaction.php
@@ -76,6 +76,9 @@ class XrdAction extends Action
         $salmon_url = common_local_url('usersalmon',
                                        array('id' => $this->user->id));
 
+        $xrd->links[] = array('rel' => Salmon::REL_SALMON,
+                              'href' => $salmon_url);
+        // XXX : Deprecated - to be removed.
         $xrd->links[] = array('rel' => Salmon::NS_REPLIES,
                               'href' => $salmon_url);
 
@@ -98,7 +101,7 @@ class XrdAction extends Action
         $xrd->links[] = array('rel' => 'http://ostatus.org/schema/1.0/subscribe',
                               'template' => $url );
 
-        header('Content-type: text/xml');
+        header('Content-type: application/xrd+xml');
         print $xrd->toXML();
     }