Merge pull request #76 from gwu-libraries/t49-harvest-status-consume

T49 harvest status consume

docker/app-dev/Dockerfile

@@ -11,7 +11,10 @@ RUN apt-get update && apt-get install -y \
     apache2 \
     libapache2-mod-wsgi \
     wget \
-    zip 
+    zip
+# Not sure why the git install needs its own line,
+# but it seems to be ignored if added into the list above
+RUN apt-get install -y git
 #Upgrade pip
 RUN pip install --upgrade pip
 #This pre-fetches the most recent requirements.txt.

docker/app-dev/invoke.sh

@@ -24,6 +24,9 @@ echo "Collecting static files"
 echo "Loading fixtures"
 /opt/sfm-ui/sfm/manage.py loaddata /opt/sfm-setup/fixtures.json
 
+echo "Starting message consumer"
+/opt/sfm-ui/sfm/manage.py startconsumer &
+
 echo "Running server"
 #Not entirely sure why this is necessary, but it works.
 /etc/init.d/apache2 start

docker/app-master/Dockerfile

@@ -11,7 +11,10 @@ RUN apt-get update && apt-get install -y \
     apache2 \
     libapache2-mod-wsgi \
     wget \
-    zip 
+    zip
+# Not sure why the git install needs its own line,
+# but it seems to be ignored if added into the list above
+RUN apt-get install -y git
 #Upgrade pip
 RUN pip install --upgrade pip
 WORKDIR /tmp

docker/app-master/invoke.sh

@@ -20,6 +20,9 @@ echo "Collecting static files"
 echo "Loading fixtures"
 /opt/sfm-ui/sfm/manage.py loaddata /opt/sfm-setup/fixtures.json
 
+echo "Starting message consumer"
+/opt/sfm-ui/sfm/manage.py startconsumer &
+
 echo "Running server"
 #Not entirely sure why this is necessary, but it works.
 /etc/init.d/apache2 start

requirements/requirements.txt

@@ -5,3 +5,4 @@ pytz
 django-crispy-forms>=1.5,<1.6
 appdeps>=1.0,<1.1
 pika>=0.10,<0.11
+git+https://github.com/gwu-libraries/sfm-utils.git@0.1.0#egg=sfmutils

sfm/message_consumer/management/commands/startconsumer.py

@@ -0,0 +1,19 @@
+from django.conf import settings
+from django.core.management.base import BaseCommand
+from message_consumer.sfm_ui_consumer import SfmUiConsumer
+from sfmutils.consumer import MqConfig, EXCHANGE
+
+QUEUE = "sfm_ui"
+ROUTING_KEYS = ["harvest.status.*", "harvest.status.*.*"]
+
+
+class Command(BaseCommand):
+    help = 'Starts the message consumer'
+
+    def handle(self, *args, **options):
+        username = settings.RABBITMQ_USER
+        password = settings.RABBITMQ_PASSWORD
+        consumer = SfmUiConsumer(mq_config=MqConfig(settings.RABBITMQ_HOST,
+                                 username, password, EXCHANGE,
+                                 {QUEUE: ROUTING_KEYS}))
+        consumer.run()

sfm/message_consumer/sfm_ui_consumer.py

@@ -0,0 +1,23 @@
+import logging
+from sfmutils.consumer import BaseConsumer
+from ui.models import Harvest, SeedSet
+
+log = logging.getLogger(__name__)
+
+
+class SfmUiConsumer(BaseConsumer):
+    """
+    Class for the SFM UI Consumer, which subscribes to
+    messages from the queue and updates the models as appropriate.
+    """
+    def on_message(self):
+        m = self.message
+        m_id = m['id']
+        m_status = m['status']
+        m_date_started = m['date_started']
+        m_date_ended = m['date_ended']
+
+        ss = SeedSet(id=m_id)
+        h = Harvest(seed_set=ss, stats=m_status,
+                    date_started=m_date_started, date_ended=m_date_ended)
+        h.save()

sfm/sfm/settings/common.py

@@ -46,6 +46,7 @@
     'allauth.account',  # registration
     'allauth.socialaccount',  # registration
     'crispy_forms',  # for django crispy forms
+    'message_consumer',
 ]
 
 MIDDLEWARE_CLASSES = (
@@ -128,3 +129,7 @@
 # a custom user model" under:
 # https://docs.djangoproject.com/en/1.8/topics/auth/customizing/
 AUTH_USER_MODEL = 'ui.User'
+
+RABBITMQ_HOST = env.get('SFM_RABBITMQ_HOST')
+RABBITMQ_USER = env.get('SFM_RABBITMQ_USER')
+RABBITMQ_PASSWORD = env.get('SFM_RABBITMQ_PASSWORD')

sfm/sfm/settings/docker_settings.py

@@ -12,4 +12,8 @@
 # A hashed version of `SITE_SUPERUSER_PASSWORD` will be store in superuser's `password` field.
 SITE_SUPERUSER_PASSWORD = env.get('SFM_SITE_ADMIN_PASSWORD', 'password')
 
-STATIC_ROOT = "/opt/sfm-static"
+STATIC_ROOT = "/opt/sfm-static"
+
+RABBITMQ_HOST = "mq"
+RABBITMQ_USER = env.get('MQ_ENV_RABBITMQ_DEFAULT_USER')
+RABBITMQ_PASSWORD = env.get('MQ_ENV_RABBITMQ_DEFAULT_PASS')

sfm/ui/admin.py

@@ -84,7 +84,16 @@ class Seed(a.ModelAdmin):
     search_fields = ['seed_set', 'token', 'uid', 'is_active',
                      'is_valid', 'stats', 'date_added', 'date_updated']
 
+
+class Harvest(a.ModelAdmin):
+    fields = ('seed_set', 'stats', 'date_started', 'date_ended')
+    list_display = ['id', 'seed_set', 'stats', 'date_started', 'date_ended']
+    list_filter = ['id', 'seed_set', 'stats', 'date_started', 'date_ended']
+    search_fields = ['id', 'seed_set', 'stats', 'date_started', 'date_ended']
+
+
 a.site.register(m.Credential, Credential)
 a.site.register(m.Collection, Collection)
 a.site.register(m.SeedSet, SeedSet)
 a.site.register(m.Seed, Seed)
+a.site.register(m.Harvest, Harvest)

sfm/ui/rabbit.py

@@ -1,28 +1,22 @@
 from django.apps import AppConfig
+from django.conf import settings
 import pika
-import os
+from sfmutils.consumer import EXCHANGE
 
 
 class RabbitWorker(AppConfig):
     name = 'ui'
     verbose_name = "ui"
     # Create a connection
     credentials = pika.PlainCredentials(
-        username=os.environ['MQ_ENV_RABBITMQ_DEFAULT_USER'],
-        password=os.environ['MQ_ENV_RABBITMQ_DEFAULT_PASS'])
+        username=settings.RABBITMQ_USER,
+        password=settings.RABBITMQ_PASSWORD)
     parameters = pika.ConnectionParameters(host='mq', credentials=credentials)
     connection = pika.BlockingConnection(parameters)
     # create channel
     channel = connection.channel()
 
     def ready(self):
         # Declare sfm_exchange
-        RabbitWorker.channel.exchange_declare(exchange="sfm_exchange",
+        RabbitWorker.channel.exchange_declare(exchange=EXCHANGE,
                                               type="topic", durable=True)
-        # Declare harvester queue
-        RabbitWorker.channel.queue_declare(queue="sfm_exchange", durable=True)
-        # Bind
-        RabbitWorker.channel.queue_bind(exchange="sfm_exchange",
-                                        queue="sfm_exchange",
-                                        routing_key="sfm_exchange")
-        pass  # startup code here

sfm/ui/views.py

@@ -6,7 +6,8 @@
 from .forms import CollectionForm, SeedSetForm, SeedForm
 from django.core.urlresolvers import reverse_lazy, reverse
 import json
-from .rabbit import RabbitWorker
+from .rabbit import RabbitWorker, EXCHANGE
+
 
 class CollectionListView(ListView):
     model = Collection
@@ -98,7 +99,7 @@ def post(self, request, *args, **kwargs):
         credential = json.loads(str(credential))
         options = json.loads(str(self.request.POST.get('harvest_options')))
         # Routing key
-        key = ''.join(['harvest.start.',str(media),'.',
+        key = ''.join(['harvest.start.', str(media), '.',
                        self.request.POST.get('harvest_type')])
         m = {
             'id': str(seedset.id),
@@ -111,7 +112,7 @@ def post(self, request, *args, **kwargs):
             },
             'seeds': seeds
         }
-        RabbitWorker.channel.basic_publish(exchange='sfm_exchange',
+        RabbitWorker.channel.basic_publish(exchange=EXCHANGE,
                                            routing_key=key,
                                            body=json.dumps(m))
         self.object = self.get_object()