| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,125 @@ | ||
| import logging | ||
| import sys | ||
|
|
||
| from geonode.geoserver.signals import geoserver_post_save2 | ||
| from geonode.security.views import send_email_consumer, send_email_owner_on_view | ||
| from geonode.social.signals import notification_post_save_resource2 | ||
| from geonode.layers.views import layer_view_counter | ||
| from kombu.mixins import ConsumerMixin | ||
| from django.conf import settings | ||
|
|
||
| from queues import queue_email_events, queue_geoserver_events,\ | ||
| queue_notifications_events, queue_all_events,\ | ||
| queue_geoserver_catalog, queue_geoserver_data,\ | ||
| queue_geoserver, queue_layer_viewers | ||
|
|
||
| logger = logging.getLogger(__package__) | ||
| logger.addHandler(logging.StreamHandler(sys.stdout)) | ||
| logger.setLevel(logging.DEBUG) | ||
|
|
||
|
|
||
| class Consumer(ConsumerMixin): | ||
| def __init__(self, connection): | ||
| self.connection = connection | ||
| return | ||
|
|
||
| def get_consumers(self, Consumer, channel): | ||
| return [ | ||
| Consumer(queue_all_events, | ||
| callbacks=[self.on_message]), | ||
| Consumer(queue_email_events, | ||
| callbacks=[self.on_email_messages]), | ||
| Consumer(queue_geoserver_events, | ||
| callbacks=[self.on_geoserver_messages]), | ||
| Consumer(queue_notifications_events, | ||
| callbacks=[self.on_notifications_messages]), | ||
| Consumer(queue_geoserver_catalog, | ||
| callbacks=[self.on_geoserver_catalog]), | ||
| Consumer(queue_geoserver_data, | ||
| callbacks=[self.on_geoserver_data]), | ||
| Consumer(queue_geoserver, | ||
| callbacks=[self.on_geoserver_all]), | ||
| Consumer(queue_layer_viewers, | ||
| callbacks=[self.on_layer_viewer]), | ||
| ] | ||
|
|
||
| def on_consume_end(self, connection, channel): | ||
| super(Consumer, self).on_consume_end(connection, channel) | ||
| logger.info("finished.") | ||
|
|
||
| def on_message(self, body, message): | ||
| logger.info("broadcast: RECEIVED MSG - body: %r" % (body,)) | ||
| message.ack() | ||
| return | ||
|
|
||
| def on_email_messages(self, body, message): | ||
| logger.info("on_email_messages: RECEIVED MSG - body: %r" % (body,)) | ||
| layer_uuid = body.get("layer_uuid") | ||
| user_id = body.get("user_id") | ||
| send_email_consumer(layer_uuid, user_id) | ||
| # Not sure if we need to send ack on this fanout version. | ||
| message.ack() | ||
| logger.info("on_email_messages: finished") | ||
| return | ||
|
|
||
| def on_geoserver_messages(self, body, message): | ||
| logger.info("on_geoserver_messages: RECEIVED MSG - body: %r" % (body,)) | ||
| layer_id = body.get("id") | ||
| geoserver_post_save2(layer_id) | ||
| # Not sure if we need to send ack on this fanout version. | ||
| message.ack() | ||
| logger.info("on_geoserver_messages: finished") | ||
| return | ||
|
|
||
| def on_notifications_messages(self, body, message): | ||
| logger.info("on_notifications_message: RECEIVED MSG - body: %r" % (body,)) | ||
| instance_id = body.get("id") | ||
| app_label = body.get("app_label") | ||
| model = body.get("model") | ||
| created = body.get("created") | ||
| notification_post_save_resource2(instance_id, app_label, model, created) | ||
| message.ack() | ||
| logger.info("on_notifications_message: finished") | ||
| return | ||
|
|
||
| def on_geoserver_all(self, body, message): | ||
| logger.info("on_geoserver_all: RECEIVED MSG - body: %r" % (body,)) | ||
| message.ack() | ||
| logger.info("on_geoserver_all: finished") | ||
| # TODO:Adding consurmer's producers. | ||
| return | ||
|
|
||
| def on_geoserver_catalog(self, body, message): | ||
| logger.info("on_geoserver_catalog: RECEIVED MSG - body: %r" % (body,)) | ||
| message.ack() | ||
| logger.info("on_geoserver_catalog: finished") | ||
| return | ||
|
|
||
| def on_geoserver_data(self, body, message): | ||
| logger.info("on_geoserver_data: RECEIVED MSG - body: %r" % (body,)) | ||
| message.ack() | ||
| logger.info("on_geoserver_data: finished") | ||
| return | ||
|
|
||
| def on_consume_ready(self, connection, channel, consumers, **kwargs): | ||
| logger.info(">>> Ready:") | ||
| logger.info(connection) | ||
| logger.info("{} consumers:".format(len(consumers))) | ||
| for i, consumer in enumerate(consumers, start=1): | ||
| logger.info("{0} {1}".format(i, consumer)) | ||
|
|
||
| super(Consumer, self).on_consume_ready(connection, channel, consumers, | ||
| **kwargs) | ||
|
|
||
| def on_layer_viewer(self, body, message): | ||
| logger.info("on_layer_viewer: RECEIVED MSG - body: %r" % (body,)) | ||
| viewer = body.get("viewer") | ||
| owner_layer = body.get("owner_layer") | ||
| layer_id = body.get("layer_id") | ||
| layer_view_counter(layer_id) | ||
| if settings.EMAIL_ENABLE: | ||
| send_email_owner_on_view(owner_layer, viewer, layer_id) | ||
| message.ack() | ||
| logger.info("on_layer_viewer: finished") | ||
|
|
||
| return |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,39 @@ | ||
| # -*- coding: utf-8 -*- | ||
| ######################################################################### | ||
| # | ||
| # Copyright (C) 2016 OSGeo | ||
| # | ||
| # This program is free software: you can redistribute it and/or modify | ||
| # it under the terms of the GNU 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 General Public License for more details. | ||
| # | ||
| # You should have received a copy of the GNU General Public License | ||
| # along with this program. If not, see <http://www.gnu.org/licenses/>. | ||
| # | ||
| ######################################################################### | ||
|
|
||
| from django.conf import settings | ||
| from django.db.models import signals | ||
| from django.utils.translation import ugettext_noop as _ | ||
| import logging | ||
| logger = logging.getLogger(__name__) | ||
|
|
||
| if "notification" in settings.INSTALLED_APPS: | ||
| import notification | ||
|
|
||
| def create_notice_types(app, created_models, verbosity, **kwargs): | ||
| notification.models.NoticeType.create("layer_uploaded", _("Layer Uploaded"), _("A layer was uploaded")) | ||
| notification.models.NoticeType.create("layer_comment", _("Comment on Layer"), _("A layer was commented on")) | ||
| notification.models.NoticeType.create("layer_rated", _("Rating for Layer"), _("A rating was given to a layer")) | ||
|
|
||
| signals.post_syncdb.connect(create_notice_types, sender=notification) | ||
| logger.info("Notifications Configured for geonode.layers.managment.commands") | ||
| else: | ||
| logger.info("Skipping creation of NoticeTypes for geonode.layers.management.commands, since notification app was \ | ||
| not found.") |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,19 @@ | ||
| # -*- coding: utf-8 -*- | ||
| ######################################################################### | ||
| # | ||
| # Copyright (C) 2016 OSGeo | ||
| # | ||
| # This program is free software: you can redistribute it and/or modify | ||
| # it under the terms of the GNU 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 General Public License for more details. | ||
| # | ||
| # You should have received a copy of the GNU General Public License | ||
| # along with this program. If not, see <http://www.gnu.org/licenses/>. | ||
| # | ||
| ######################################################################### |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,40 @@ | ||
| # -*- coding: utf-8 -*- | ||
| ######################################################################### | ||
| # | ||
| # Copyright (C) 2016 OSGeo | ||
| # | ||
| # This program is free software: you can redistribute it and/or modify | ||
| # it under the terms of the GNU 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 General Public License for more details. | ||
| # | ||
| # You should have received a copy of the GNU General Public License | ||
| # along with this program. If not, see <http://www.gnu.org/licenses/>. | ||
| # | ||
| ######################################################################### | ||
|
|
||
| from django.core.management.base import BaseCommand | ||
|
|
||
| from geonode.messaging.queues import queue_email_events, queue_geoserver_events, \ | ||
| queue_notifications_events, queue_all_events, \ | ||
| queue_geoserver_catalog, queue_geoserver_data, \ | ||
| queue_geoserver, queue_layer_viewers | ||
|
|
||
|
|
||
| class Command(BaseCommand): | ||
| help = 'Start the MQ consumer to perform non blocking tasks' | ||
|
|
||
| def handle(self, **options): | ||
| queue_geoserver_events.purge() | ||
| queue_notifications_events.purge() | ||
| queue_email_events.purge() | ||
| queue_all_events.purge() | ||
| queue_geoserver_catalog.purge() | ||
| queue_geoserver_data.purge() | ||
| queue_geoserver.purge() | ||
| queue_layer_viewers.purge() |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,55 @@ | ||
| # -*- coding: utf-8 -*- | ||
| ######################################################################### | ||
| # | ||
| # Copyright (C) 2016 OSGeo | ||
| # | ||
| # This program is free software: you can redistribute it and/or modify | ||
| # it under the terms of the GNU 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 General Public License for more details. | ||
| # | ||
| # You should have received a copy of the GNU General Public License | ||
| # along with this program. If not, see <http://www.gnu.org/licenses/>. | ||
| # | ||
| ######################################################################### | ||
|
|
||
| import logging | ||
| import sys | ||
| from optparse import make_option | ||
|
|
||
| from django.core.management.base import BaseCommand | ||
| from geonode.settings import BROKER_URL | ||
|
|
||
| logger = logging.getLogger(__package__) | ||
| logger.addHandler(logging.StreamHandler(sys.stdout)) | ||
| logger.setLevel(logging.DEBUG) | ||
|
|
||
|
|
||
| class Command(BaseCommand): | ||
| help = 'Start the MQ consumer to perform non blocking tasks' | ||
| option_list = BaseCommand.option_list + ( | ||
| make_option( | ||
| '-i', | ||
| '--ignore-errors', | ||
| action='store_true', | ||
| dest='ignore_errors', | ||
| default=False, | ||
| help='Stop after any errors are encountered.'), | ||
| ) | ||
|
|
||
| def handle(self, **options): | ||
| from kombu import BrokerConnection | ||
| from geonode.messaging.consumer import Consumer | ||
|
|
||
| with BrokerConnection(BROKER_URL) as connection: | ||
| try: | ||
| logger.info("Consumer starting.") | ||
| worker = Consumer(connection) | ||
| worker.run() | ||
| except KeyboardInterrupt: | ||
| logger.info("Consumer stopped.") |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,65 @@ | ||
| from __future__ import with_statement | ||
|
|
||
| from geonode.settings import BROKER_URL | ||
| from kombu import BrokerConnection | ||
| from kombu.common import maybe_declare | ||
| from kombu.pools import producers | ||
| from queues import queue_email_events, queue_geoserver_events,\ | ||
| queue_notifications_events, queue_layer_viewers | ||
|
|
||
|
|
||
| connection = BrokerConnection(BROKER_URL) | ||
|
|
||
|
|
||
| def send_email_producer(layer_uuid, user_id): | ||
| with producers[connection].acquire(block=True) as producer: | ||
| maybe_declare(queue_email_events, producer.channel) | ||
| payload = { | ||
| "layer_uuid": layer_uuid, | ||
| "user_id": user_id | ||
|
|
||
| } | ||
| producer.publish( | ||
| payload, | ||
| exchange='geonode', | ||
| serializer='json', | ||
| routing_key='email' | ||
| ) | ||
|
|
||
|
|
||
| def geoserver_upload_layer(payload): | ||
| with producers[connection].acquire(block=True) as producer: | ||
| maybe_declare(queue_geoserver_events, producer.channel) | ||
| producer.publish( | ||
| payload, | ||
| exchange='geonode', | ||
| serializer='json', | ||
| routing_key='geonode.geoserver' | ||
| ) | ||
|
|
||
|
|
||
| def notifications_send(payload, created=None): | ||
| with producers[connection].acquire(block=True) as producer: | ||
| maybe_declare(queue_notifications_events, producer.channel) | ||
| payload['created'] = created | ||
| producer.publish( | ||
| payload, | ||
| exchange='geonode', | ||
| serializer='json', | ||
| routing_key='notifications' | ||
| ) | ||
|
|
||
|
|
||
| def viewing_layer(user, owner, layer_id): | ||
| with producers[connection].acquire(block=True) as producer: | ||
| maybe_declare(queue_layer_viewers, producer.channel) | ||
|
|
||
| payload = {"viewer": user, | ||
| "owner_layer": owner, | ||
| "layer_id": layer_id} | ||
| producer.publish( | ||
| payload, | ||
| exchange='geonode', | ||
| serializer='json', | ||
| routing_key='geonode.viewer' | ||
| ) |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,12 @@ | ||
| from kombu import Exchange, Queue | ||
|
|
||
| geonode_exchange = Exchange("geonode", type="topic", durable=False) | ||
|
|
||
| queue_all_events = Queue("broadcast", geonode_exchange, routing_key="#") | ||
| queue_email_events = Queue("email.events", geonode_exchange, routing_key="email") | ||
| queue_geoserver = Queue("all.geoserver", geonode_exchange, routing_key="geoserver.#") | ||
| queue_geoserver_catalog = Queue("geoserver.catalog", geonode_exchange, routing_key="geoserver.catalog") | ||
| queue_geoserver_data = Queue("geoserver.data", geonode_exchange, routing_key="geoserver.catalog") | ||
| queue_geoserver_events = Queue("geoserver.events", geonode_exchange, routing_key="geonode.geoserver") | ||
| queue_notifications_events = Queue("notifications.events", geonode_exchange, routing_key="notifications") | ||
| queue_layer_viewers = Queue("geonode.layer.viewer", geonode_exchange, routing_key="geonode.viewer") |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,30 @@ | ||
| # -*- coding: utf-8 -*- | ||
| ######################################################################### | ||
| # | ||
| # Copyright (C) 2016 OSGeo | ||
| # | ||
| # This program is free software: you can redistribute it and/or modify | ||
| # it under the terms of the GNU 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 General Public License for more details. | ||
| # | ||
| # You should have received a copy of the GNU General Public License | ||
| # along with this program. If not, see <http://www.gnu.org/licenses/>. | ||
| # | ||
| ######################################################################### | ||
|
|
||
| from django.test import TestCase | ||
|
|
||
|
|
||
| class MessagingTest(TestCase): | ||
| """ | ||
| Tests geonode.messaging | ||
| """ | ||
| def setUp(self): | ||
| self.adm_un = "admin" | ||
| self.adm_pw = "admin" |