OStatus: do PuSH subscription setup from subscribe/join event hooks, so resubscribing directly from a profile/group list works correctly if there aren't active subscriptions at the moment.

lib/activity.php

@@ -823,7 +823,9 @@ class Activity
         if ($namespace) {
             $attrs = array('xmlns' => 'http://www.w3.org/2005/Atom',
                            'xmlns:activity' => 'http://activitystrea.ms/spec/1.0/',
-                           'xmlns:ostatus' => 'http://ostatus.org/schema/1.0');
+                           'xmlns:georss' => 'http://www.georss.org/georss',
+                           'xmlns:ostatus' => 'http://ostatus.org/schema/1.0',
+                           'xmlns:poco' => 'http://portablecontacts.net/spec/1.0');
         } else {
             $attrs = array();
         }

plugins/OStatus/OStatusPlugin.php

@@ -251,58 +251,6 @@ class OStatusPlugin extends Plugin
         return true;
     }
 
-    /**
-     * Notify remote server and garbage collect unused feeds on unsubscribe.
-     * @fixme send these operations to background queues
-     *
-     * @param User $user
-     * @param Profile $other
-     * @return hook return value
-     */
-    function onEndUnsubscribe($profile, $other)
-    {
-        $user = User::staticGet('id', $profile->id);
-
-        if (empty($user)) {
-            return true;
-        }
-
-        $oprofile = Ostatus_profile::staticGet('profile_id', $other->id);
-
-        if (empty($oprofile)) {
-            return true;
-        }
-
-        // Drop the PuSH subscription if there are no other subscribers.
-
-        if ($other->subscriberCount() == 0) {
-            common_log(LOG_INFO, "Unsubscribing from now-unused feed $oprofile->feeduri");
-            $oprofile->unsubscribe();
-        }
-
-        $act = new Activity();
-
-        $act->verb = ActivityVerb::UNFOLLOW;
-
-        $act->id   = TagURI::mint('unfollow:%d:%d:%s',
-                                  $profile->id,
-                                  $other->id,
-                                  common_date_iso8601(time()));
-
-        $act->time    = time();
-        $act->title   = _("Unfollow");
-        $act->content = sprintf(_("%s stopped following %s."),
-                               $profile->getBestName(),
-                               $other->getBestName());
-
-        $act->actor   = ActivityObject::fromProfile($profile);
-        $act->object  = ActivityObject::fromProfile($other);
-
-        $oprofile->notifyActivity($act);
-
-        return true;
-    }
-
     /**
      * Make sure necessary tables are filled out.
      */
@@ -366,6 +314,50 @@ class OStatusPlugin extends Plugin
         }
     }
 
+    /**
+     * When about to subscribe to a remote user, start a server-to-server
+     * PuSH subscription if needed. If we can't establish that, abort.
+     *
+     * @fixme If something else aborts later, we could end up with a stray
+     *        PuSH subscription. This is relatively harmless, though.
+     *
+     * @param Profile $subscriber
+     * @param Profile $other
+     *
+     * @return hook return code
+     *
+     * @throws Exception
+     */
+    function onStartSubscribe($subscriber, $other)
+    {
+        $user = User::staticGet('id', $subscriber->id);
+
+        if (empty($user)) {
+            return true;
+        }
+
+        $oprofile = Ostatus_profile::staticGet('profile_id', $other->id);
+
+        if (empty($oprofile)) {
+            return true;
+        }
+
+        if (!$oprofile->subscribe()) {
+            throw new Exception(_m('Could not set up remote subscription.'));
+        }
+    }
+
+    /**
+     * Having established a remote subscription, send a notification to the
+     * remote OStatus profile's endpoint.
+     *
+     * @param Profile $subscriber
+     * @param Profile $other
+     *
+     * @return hook return code
+     *
+     * @throws Exception
+     */
     function onEndSubscribe($subscriber, $other)
     {
         $user = User::staticGet('id', $subscriber->id);
@@ -403,6 +395,54 @@ class OStatusPlugin extends Plugin
         return true;
     }
 
+    /**
+     * Notify remote server and garbage collect unused feeds on unsubscribe.
+     * @fixme send these operations to background queues
+     *
+     * @param User $user
+     * @param Profile $other
+     * @return hook return value
+     */
+    function onEndUnsubscribe($profile, $other)
+    {
+        $user = User::staticGet('id', $profile->id);
+
+        if (empty($user)) {
+            return true;
+        }
+
+        $oprofile = Ostatus_profile::staticGet('profile_id', $other->id);
+
+        if (empty($oprofile)) {
+            return true;
+        }
+
+        // Drop the PuSH subscription if there are no other subscribers.
+        $oprofile->garbageCollect();
+
+        $act = new Activity();
+
+        $act->verb = ActivityVerb::UNFOLLOW;
+
+        $act->id   = TagURI::mint('unfollow:%d:%d:%s',
+                                  $profile->id,
+                                  $other->id,
+                                  common_date_iso8601(time()));
+
+        $act->time    = time();
+        $act->title   = _("Unfollow");
+        $act->content = sprintf(_("%s stopped following %s."),
+                               $profile->getBestName(),
+                               $other->getBestName());
+
+        $act->actor   = ActivityObject::fromProfile($profile);
+        $act->object  = ActivityObject::fromProfile($other);
+
+        $oprofile->notifyActivity($act);
+
+        return true;
+    }
+
     /**
      * When one of our local users tries to join a remote group,
      * notify the remote server. If the notification is rejected,
@@ -418,6 +458,10 @@ class OStatusPlugin extends Plugin
     {
         $oprofile = Ostatus_profile::staticGet('group_id', $group->id);
         if ($oprofile) {
+            if (!$oprofile->subscribe()) {
+                throw new Exception(_m('Could not set up remote group membership.'));
+            }
+
             $member = Profile::staticGet($user->id);
 
             $act = new Activity();
@@ -439,7 +483,8 @@ class OStatusPlugin extends Plugin
             if ($oprofile->notifyActivity($act)) {
                 return true;
             } else {
-                throw new ServerException(_m("Failed joining remote group."));
+                $oprofile->garbageCollect();
+                throw new Exception(_m("Failed joining remote group."));
             }
         }
     }
@@ -464,12 +509,7 @@ class OStatusPlugin extends Plugin
         $oprofile = Ostatus_profile::staticGet('group_id', $group->id);
         if ($oprofile) {
             // Drop the PuSH subscription if there are no other subscribers.
-    
+            $oprofile->garbageCollect();
-            $members = $group->getMembers(0, 1);
-            if ($members->N == 0) {
-                common_log(LOG_INFO, "Unsubscribing from now-unused group feed $oprofile->feeduri");
-                $oprofile->unsubscribe();
-            }
 
 
             $member = Profile::staticGet($user->id);

plugins/OStatus/actions/ostatussub.php

@@ -324,11 +324,6 @@ class OStatusSubAction extends Action
         // And subscribe the current user to the local profile
         $user = common_current_user();
 
-        if (!$this->oprofile->subscribe()) {
-            $this->showForm(_m("Failed to set up server-to-server subscription."));
-            return;
-        }
-
         if ($this->oprofile->isGroup()) {
             $group = $this->oprofile->localGroup();
             if ($user->isMember($group)) {

plugins/OStatus/classes/Ostatus_profile.php

@@ -346,6 +346,29 @@ class Ostatus_profile extends Memcached_DataObject
         }
     }
 
+    /**
+     * Check if this remote profile has any active local subscriptions, and
+     * if not drop the PuSH subscription feed.
+     *
+     * @return boolean
+     */
+    public function garbageCollect()
+    {
+        if ($this->isGroup()) {
+            $members = $this->localGroup()->getMembers(0, 1);
+            $count = $members->N;
+        } else {
+            $count = $this->localProfile()->subscriberCount();
+        }
+        if ($count == 0) {
+            common_log(LOG_INFO, "Unsubscribing from now-unused remote feed $oprofile->feeduri");
+            $this->unsubscribe();
+            return true;
+        } else {
+            return false;
+        }
+    }
+
     /**
      * Send an Activity Streams notification to the remote Salmon endpoint,
      * if so configured.
@@ -379,7 +402,8 @@ class Ostatus_profile extends Memcached_DataObject
                                 'xmlns:activity' => 'http://activitystrea.ms/spec/1.0/',
                                 'xmlns:thr' => 'http://purl.org/syndication/thread/1.0',
                                 'xmlns:georss' => 'http://www.georss.org/georss',
-                                'xmlns:ostatus' => 'http://ostatus.org/schema/1.0');
+                                'xmlns:ostatus' => 'http://ostatus.org/schema/1.0',
+                                'xmlns:poco' => 'http://portablecontacts.net/spec/1.0');
 
             $entry = new XMLStringer();
             $entry->elementStart('entry', $attributes);