Merge pull request #26 from eea/develop

Merged develop into master

Author: olimpiurob

Products/Reportek/EnvelopeInstance.py

@@ -551,6 +551,7 @@ def forwardState(self, REQUEST=None):
         """.."""
         result = {}
         engine = getattr(self, ENGINE_ID)
+        rmq = getattr(engine, 'env_fwd_rmq', False)
         if getattr(self, 'wf_status', None) == 'forward':
             wks = self.getListOfWorkitems()
             wk = wks[-1]
@@ -569,6 +570,12 @@ def forwardState(self, REQUEST=None):
             else:
                 if self.wf_status == 'forward':
                     wk.triggerApplication(wk.id, REQUEST)
+                    if rmq:
+                        activity = self.getActivity(wk.id)
+                        queue_msg(
+                            "{}|{}".format(self.absolute_url(),
+                                           self.get_freq(wk.id)),
+                            queue=self.get_rmq_queue(activity.getId()))
                     result['triggered'] = wk.activity_id
 
         return engine.jsonify(result)
@@ -636,6 +643,23 @@ def getDestinations(self, workitem_id, path=None):
                                  'process_to_id': process.id})
         return destinations
 
+    def get_rmq_queue(self, act_id):
+        queue = 'fwd_envelopes'
+        # Uncomment to allow for separate queues based on Activity
+        # if act_id in ['AutomaticQA', 'FMEConversionApplication']:
+        #     queue = 'poll_envelopes'
+        return queue
+
+    def get_freq(self, workitem_id):
+        """Return the frequency at which the application should be exec
+        """
+        freq = 1
+        application_url = self.getApplicationUrl(workitem_id)
+        application = self.unrestrictedTraverse(application_url, None)
+        if application:
+            freq = int(getattr(application, 'retryFrequency', freq))
+        return freq
+
     def handleWorkitem(self, workitem_id, REQUEST=None):
         # If it's a previously failed application, retry it, otherwise forward
         # it
@@ -653,9 +677,12 @@ def handleWorkitem(self, workitem_id, REQUEST=None):
                 if activity.isAutoStart():
                     self.wf_status = 'forward'
                     self.reindex_object()
-                    if rmq:
-                        queue_msg(self.absolute_url(), queue='fwd_envelopes')
                     self.startAutomaticApplication(workitem_id)
+                    if rmq:
+                        queue_msg(
+                            "{}|{}".format(self.absolute_url(),
+                                           self.get_freq(workitem_id)),
+                            queue=self.get_rmq_queue(activity.getId()))
                 else:
                     self.wf_status = 'manual'
                     self.reindex_object()
@@ -920,9 +947,12 @@ def manageWorkitemCreation(self, workitem_id):
                 if activity.isAutoStart():
                     self.wf_status = 'forward'
                     self.reindex_object()
-                    if rmq:
-                        queue_msg(self.absolute_url(), queue='fwd_envelopes')
                     self.startAutomaticApplication(workitem_id)
+                    if rmq:
+                        queue_msg(
+                            "{}|{}".format(self.absolute_url(),
+                                           self.get_freq(workitem_id)),
+                            queue=self.get_rmq_queue(activity.getId()))
                 else:
                     self.wf_status = 'manual'
                     self.reindex_object()

Products/Reportek/version.txt

@@ -1 +1 @@
-4.7
+4.8

Products/Reportek/views/envelope_utils.py

@@ -264,6 +264,14 @@ def env_long_running_aqa(self):
 
         return json.dumps(envelopes)
 
+    def get_rmq_queue(self, act_id):
+        queue = 'fwd_envelopes'
+        # Uncomment to allow for separate queues based on Activity
+        # if act_id in ['AutomaticQA', 'FMEConversionApplication']:
+        #     queue = 'poll_envelopes'
+
+        return queue
+
     def forwardable_envelopes(self):
         pub_envs = self.request.get('envelopes', [])
         if self.request.get('btn.publish') and not pub_envs:
@@ -273,6 +281,7 @@ def forwardable_envelopes(self):
             if self.rmq_fwd:
                 for env in pub_envs:
                     try:
+                        # TODO: dynamically set the queue here
                         send_message(env, queue='fwd_envelopes')
                         results.append({
                             'envelope': env,

docs/HISTORY.txt

@@ -1,6 +1,10 @@
 Changelog
 =========
 
+4.8 - (2022-01-25)
+---------------------------
+* Change: Develop [olimpiurob]
+
 4.3 - (2022-01-10)
 ---------------------------
   "documentation_url": "https://docs.github.com/rest/reference/repos#get-repository-content"