Merge remote-tracking branch 'refs/remotes/spaetz/436_celery_push'

mediagoblin/plugins/api/views.py

@@ -86,7 +86,10 @@ def post_entry(request):
     #
     # (... don't change entry after this point to avoid race
     # conditions with changes to the document via processing code)
-    run_process_media(entry)
+    feed_url = request.urlgen(
+        'mediagoblin.user_pages.atom_feed',
+        qualified=True, user=request.user.username)
+    run_process_media(entry, feed_url)
 
     return json_response(get_entry_serializable(entry, request.urlgen))
 

mediagoblin/processing/task.py

@@ -15,8 +15,10 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 import logging
+import urllib
+import urllib2
 
-from celery.task import Task
+from celery import registry, task
 
 from mediagoblin import mg_globals as mgg
 from mediagoblin.db.models import MediaEntry
@@ -28,18 +30,51 @@ logging.basicConfig()
 _log.setLevel(logging.DEBUG)
 
 
+@task.task(default_retry_delay=2 * 60)
+def handle_push_urls(feed_url):
+    """Subtask, notifying the PuSH servers of new content
+
+    Retry 3 times every 2 minutes if run in separate process before failing."""
+    if not mgg.app_config["push_urls"]:
+        return # Nothing to do
+    _log.debug('Notifying Push servers for feed {0}'.format(feed_url))
+    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 mgg.app_config["push_urls"]:
+        hubrequest = urllib2.Request(huburl, hubdata, hubheaders)
+        try:
+            hubresponse = urllib2.urlopen(hubrequest)
+        except (urllib2.HTTPError, urllib2.URLError) as exc:
+            # We retry by default 3 times before failing
+            _log.info("PuSH url %r gave error %r", huburl, exc)
+            try:
+                return handle_push_urls.retry(exc=exc, throw=False)
+            except Exception as e:
+                # All retries failed, Failure is no tragedy here, probably.
+                _log.warn('Failed to notify PuSH server for feed {0}. '
+                          'Giving up.'.format(feed_url))
+                return False
+
 ################################
 # Media processing initial steps
 ################################
 
-class ProcessMedia(Task):
+class ProcessMedia(task.Task):
     """
     Pass this entry off for processing.
     """
-    def run(self, media_id):
+    def run(self, media_id, feed_url):
         """
         Pass the media entry off to the appropriate processing function
         (for now just process_image...)
+
+        :param feed_url: The feed URL that the PuSH server needs to be
+            updated for.
         """
         entry = MediaEntry.query.get(media_id)
 
@@ -58,6 +93,10 @@ class ProcessMedia(Task):
             entry.state = u'processed'
             entry.save()
 
+            # Notify the PuSH servers as async task
+            if mgg.app_config["push_urls"] and feed_url:
+                handle_push_urls.subtask().delay(feed_url)
+
             json_processing_callback(entry)
         except BaseProcessingFail as exc:
             mark_entry_failed(entry.id, exc)
@@ -97,3 +136,7 @@ class ProcessMedia(Task):
 
         entry = mgg.database.MediaEntry.query.filter_by(id=entry_id).first()
         json_processing_callback(entry)
+
+# Register the task
+process_media = registry.tasks[ProcessMedia.name]
+

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,17 @@ def prepare_queue_task(app, entry, filename):
     return queue_file
 
 
-def run_process_media(entry):
+def run_process_media(entry, feed_url=None):
-    process_media = registry.tasks[ProcessMedia.name]
+    """Process the media asynchronously
+
+    :param entry: MediaEntry() instance to be processed.
+    :param feed_url: A string indicating the feed_url that the PuSH servers
+        should be notified of. This will be sth like: `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 +78,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)

mediagoblin/submit/views.py

@@ -32,8 +32,7 @@ from mediagoblin.submit import forms as submit_forms
 from mediagoblin.messages import add_message, SUCCESS
 from mediagoblin.media_types import sniff_media, \
     InvalidFileType, FileTypeNotSupported
-from mediagoblin.submit.lib import handle_push_urls, run_process_media, \
+from mediagoblin.submit.lib import run_process_media, prepare_queue_task
-    prepare_queue_task
 
 
 @require_active_login
@@ -90,10 +89,10 @@ def submit_start(request):
                 #
                 # (... don't change entry after this point to avoid race
                 # conditions with changes to the document via processing code)
-                run_process_media(entry)
+                feed_url = request.urlgen(
-
+                    'mediagoblin.user_pages.atom_feed',
-                handle_push_urls(request)
+                    qualified=True, user=request.user.username)
-
+                run_process_media(entry, feed_url)
                 add_message(request, SUCCESS, _('Woohoo! Submitted!'))
 
                 return redirect(request, "mediagoblin.user_pages.user_home",