Factor out the interactions handler in the worker code.

Author: hainesr

lib/taverna_player/worker.rb

@@ -109,65 +109,7 @@ def perform
             return
           end
 
-          run.notifications(:requests).each do |note|
-            if @run.has_parent?
-              next if note.has_reply? || note.is_notification?
-              int = Interaction.find_by_run_id_and_serial(@run.parent_id, note.serial)
-              new_int = Interaction.find_or_initialize_by_run_id_and_serial(@run.id, note.serial)
-              if new_int.new_record?
-                note.reply(int.feed_reply, int.data)
-                new_int.displayed = true
-                new_int.replied = true
-                new_int.feed_reply = int.feed_reply
-                new_int.data = int.data
-                new_int.save
-              end
-            else
-              int = Interaction.find_or_initialize_by_run_id_and_serial(@run.id, note.serial)
-
-              # Need to catch this here in case some other process has replied.
-              if note.has_reply? && !int.replied?
-                int.replied = true
-                int.save
-              end
-
-              unless int.replied?
-                if int.page.blank?
-                  page = server.read(note.uri, "text/html", credentials)
-
-                  INTERACTION_REGEX.match(page) do
-                    page_uri = $1
-
-                    if page_uri.starts_with?(server.uri.to_s)
-                      page = server.read(URI.parse(page_uri), "text/html", credentials)
-                      int.page = page.gsub("#{run.interactions_uri.to_s}/pmrpc.js",
-                        "/assets/taverna_player/application.js")
-                    else
-                      int.page_uri = page_uri
-                    end
-                  end
-                end
-
-                if note.is_notification? && !int.new_record?
-                  int.replied = true
-                end
-
-                if int.data.blank?
-                  int.data = note.input_data.force_encoding("UTF-8")
-                end
-
-                if !int.feed_reply.blank? && !int.data.blank?
-                  note.reply(int.feed_reply, int.data)
-
-                  int.replied = true
-                end
-
-                int.save
-              end
-
-              waiting = true unless int.replied?
-            end
-          end
+          waiting = interactions(run, credentials)
 
           status_message(waiting ? "interact" : "running")
         end
@@ -241,6 +183,75 @@ def create_run(server, workflow, credentials)
       failed(ce)
     end
 
+    def interactions(run, credentials)
+      wait = false
+
+      run.notifications(:requests).each do |note|
+        if @run.has_parent?
+          next if note.has_reply? || note.is_notification?
+
+          int = Interaction.find_by_run_id_and_serial(@run.parent_id, note.serial)
+          new_int = Interaction.find_or_initialize_by_run_id_and_serial(@run.id, note.serial)
+
+          if new_int.new_record?
+            note.reply(int.feed_reply, int.data)
+            new_int.displayed = true
+            new_int.replied = true
+            new_int.feed_reply = int.feed_reply
+            new_int.data = int.data
+            new_int.save
+          end
+        else
+          int = Interaction.find_or_initialize_by_run_id_and_serial(@run.id, note.serial)
+
+          # Need to catch this here in case some other process has replied.
+          if note.has_reply? && !int.replied?
+            int.replied = true
+            int.save
+          end
+
+          unless int.replied?
+            if int.page.blank?
+              server = run.server
+              page = server.read(note.uri, "text/html", credentials)
+
+              INTERACTION_REGEX.match(page) do
+                page_uri = $1
+
+                if page_uri.starts_with?(server.uri.to_s)
+                  page = server.read(URI.parse(page_uri), "text/html", credentials)
+                  int.page = page.gsub("#{run.interactions_uri.to_s}/pmrpc.js",
+                  "/assets/taverna_player/application.js")
+                else
+                  int.page_uri = page_uri
+                end
+              end
+            end
+
+            # If this is a pure notification that we've already seen then
+            # set it as replied as we don't want it blocking a proper
+            # interaction.
+            int.replied = true if note.is_notification? && !int.new_record?
+
+            if int.data.blank?
+              int.data = note.input_data.force_encoding("UTF-8")
+            end
+
+            if !int.feed_reply.blank? && !int.data.blank?
+              note.reply(int.feed_reply, int.data)
+              int.replied = true
+            end
+
+            int.save
+          end
+
+          wait = true unless int.replied?
+        end
+      end
+
+      wait
+    end
+
     # Run the specified callback and return false on error so that we know to
     # return out of the worker code completely.
     def run_callback(cb, message)

test/unit/taverna_player/worker_test.rb

@@ -39,6 +39,7 @@ class WorkerTest < ActiveSupport::TestCase
 
     # Stuff we can't test yet in TavernaPlayer::Worker.
     flexmock(TavernaPlayer::Worker).new_instances do |w|
+      w.should_receive(:interactions).and_return(false)
       w.should_receive(:download_outputs).and_return_undefined
       w.should_receive(:process_outputs).and_return([])
     end
@@ -102,7 +103,6 @@ class WorkerTest < ActiveSupport::TestCase
       r.should_receive(:name=).once.and_return(true)
       r.should_receive(:start).twice.and_return(false, true)
       r.should_receive(:start_time).and_return(Time.now)
-      r.should_receive(:notifications).and_return([])
       r.should_receive(:finish_time).and_return(Time.now)
       r.should_receive(:log).once.and_return(0)
       r.should_receive(:delete).and_return_undefined
@@ -296,7 +296,6 @@ class WorkerTest < ActiveSupport::TestCase
       r.should_receive(:name=).once.and_return(true)
       r.should_receive(:start).once.and_return(true)
       r.should_receive(:start_time).and_return(Time.now)
-      r.should_receive(:notifications).and_return([])
       r.should_receive(:finish_time).and_return(Time.now)
       r.should_receive(:log).once.and_return(0)
       r.should_receive(:delete).and_return_undefined
@@ -341,7 +340,6 @@ class WorkerTest < ActiveSupport::TestCase
       r.should_receive(:name=).once.and_return(true)
       r.should_receive(:start).once.and_return(true)
       r.should_receive(:start_time).and_return(Time.now)
-      r.should_receive(:notifications).and_return([])
       r.should_receive(:finish_time).and_return(Time.now)
       r.should_receive(:log).once.and_return(0)
       r.should_receive(:delete).and_return_undefined