Make PuSHing the Pubhubsubbub server an async task (#436, #585)

Notifying the PuSH servers had 3 problems. 

1) it was done immediately after sending of the processing task to celery. So if celery was run in a separate
process we would notify the PuSH servers before the new media was processed/
visible. (#436)

2) Notification code was called in submit/views.py, so submitting via the
   API never resulted in notifications. (#585)

3) If Notifying the PuSH server failed, we would never retry.

The solution was to make the PuSH notification an asynchronous subtask. This
way: 1) it will only be called once async processing has finished, 2) it
is in the main processing code path, so even API calls will result in
notifications, and 3) We retry 3 times in case of failure before giving up.
If the server is in a separate process, we will wait 3x 2 minutes before
retrying the notification.

The only downside is that the celery server needs to have access to the internet
to ping the PuSH server. If that is a problem, we need to make the task belong
to a special group of celery servers that has access to the internet.

As a side effect, I believe I removed the limitation that prevented us from
upgrading celery.

Signed-off-by: Sebastian Spaeth <Sebastian@SSpaeth.de>

1 files changed, 6 insertions, 35 deletions

diff --git a/mediagoblin/submit/lib.py b/mediagoblin/submit/lib.py
index db5dfe53..ba07c6fa 100644
--- a/mediagoblin/submit/lib.py
+++ b/mediagoblin/submit/lib.py
@@ -14,16 +14,12 @@
 # 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/>.
 
-import urllib
-import urllib2
 import logging
 import uuid
-from celery import registry
 from werkzeug.utils import secure_filename
 
-from mediagoblin import mg_globals
 from mediagoblin.processing import mark_entry_failed
-from mediagoblin.processing.task import ProcessMedia
+from mediagoblin.processing.task import process_media
 
 
 _log = logging.getLogger(__name__)
@@ -58,11 +54,13 @@ def prepare_queue_task(app, entry, filename):
     return queue_file
 
 
-def run_process_media(entry):
-    process_media = registry.tasks[ProcessMedia.name]
+def run_process_media(entry, request):
+    feed_url = request.urlgen(
+        'mediagoblin.user_pages.atom_feed',
+        qualified=True, user=request.user.username)
     try:
         process_media.apply_async(
-            [unicode(entry.id)], {},
+            [entry.id, feed_url], {},
             task_id=entry.queued_task_id)
     except BaseException as exc:
         # The purpose of this section is because when running in "lazy"
@@ -76,30 +74,3 @@ def run_process_media(entry):
         mark_entry_failed(entry.id, exc)
         # re-raise the exception
         raise
-
-
-def handle_push_urls(request):
-    if mg_globals.app_config["push_urls"]:
-        feed_url = request.urlgen(
-                           'mediagoblin.user_pages.atom_feed',
-                           qualified=True,
-                           user=request.user.username)
-        hubparameters = {
-            'hub.mode': 'publish',
-            'hub.url': feed_url}
-        hubdata = urllib.urlencode(hubparameters)
-        hubheaders = {
-            "Content-type": "application/x-www-form-urlencoded",
-            "Connection": "close"}
-        for huburl in mg_globals.app_config["push_urls"]:
-            hubrequest = urllib2.Request(huburl, hubdata, hubheaders)
-            try:
-                hubresponse = urllib2.urlopen(hubrequest)
-            except urllib2.HTTPError as exc:
-                # This is not a big issue, the item will be fetched
-                # by the PuSH server next time we hit it
-                _log.warning(
-                    "push url %r gave error %r", huburl, exc.code)
-            except urllib2.URLError as exc:
-                _log.warning(
-                    "push url %r is unreachable %r", huburl, exc.reason)