Performance improvements

lib/gush/client.rb

@@ -73,7 +73,7 @@ def next_free_workflow_id
 
     def all_workflows
       connection_pool.with do |redis|
-        redis.keys("gush.workflows.*").map do |key|
+        redis.scan_each(match: "gush.workflows.*").map do |key|
           id = key.sub("gush.workflows.", "")
           find_workflow(id)
         end
@@ -86,7 +86,7 @@ def find_workflow(id)
 
         unless data.nil?
           hash = Gush::JSON.decode(data, symbolize_keys: true)
-          keys = redis.keys("gush.jobs.#{id}.*")
+          keys = redis.scan_each(match: "gush.jobs.#{id}.*")
           nodes = redis.mget(*keys).map { |json| Gush::JSON.decode(json, symbolize_keys: true) }
           workflow_from_hash(hash, nodes)
         else
@@ -116,7 +116,7 @@ def find_job(workflow_id, job_id)
       hypen = '-' if job_name_match.nil?
 
       keys = connection_pool.with do |redis|
-        redis.keys("gush.jobs.#{workflow_id}.#{job_id}#{hypen}*")
+        redis.scan_each(match: "gush.jobs.#{workflow_id}.#{job_id}#{hypen}*").to_a
       end
 
       return nil if keys.nil?
@@ -153,14 +153,14 @@ def enqueue_job(workflow_id, job)
 
     private
 
-    def workflow_from_hash(hash, nodes = nil)
+    def workflow_from_hash(hash, nodes = [])
       flow = hash[:klass].constantize.new(*hash[:arguments])
       flow.jobs = []
       flow.stopped = hash.fetch(:stopped, false)
       flow.id = hash[:id]
 
-      (nodes || hash[:nodes]).each do |node|
-        flow.jobs << Gush::Job.from_hash(node)
+      flow.jobs = nodes.map do |node|
+        Gush::Job.from_hash(node)
       end
 
       flow

lib/gush/job.rb

@@ -88,8 +88,8 @@ def ready_to_start?
     end
 
     def parents_succeeded?
-      incoming.all? do |name|
-        client.find_job(workflow_id, name).succeeded?
+      !incoming.any? do |name|
+        !client.find_job(workflow_id, name).succeeded?
       end
     end
 

lib/gush/workflow.rb

@@ -168,7 +168,6 @@ def to_hash
         total: jobs.count,
         finished: jobs.count(&:finished?),
         klass: name,
-        jobs: jobs.map(&:as_json),
         status: status,
         stopped: stopped,
         started_at: started_at,

spec/gush/workflow_spec.rb

@@ -13,8 +13,7 @@ def configure(*args)
       end
 
       expect_any_instance_of(klass).to receive(:configure).with("arg1", "arg2")
-      flow = klass.new("arg1", "arg2")
-
+      klass.new("arg1", "arg2")
     end
   end
 
@@ -102,35 +101,7 @@ def configure(*args)
           "started_at" => nil,
           "finished_at" => nil,
           "stopped" => false,
-          "arguments" => ["arg1", "arg2"],
-          "jobs" => [
-              {
-                  "name"=>a_string_starting_with('FetchFirstJob'),
-                  "klass"=>"FetchFirstJob",
-                  "incoming"=>[],
-                  "outgoing"=>[a_string_starting_with('PersistFirstJob')],
-                  "finished_at"=>nil,
-                  "started_at"=>nil,
-                  "enqueued_at"=>nil,
-                  "failed_at"=>nil,
-                  "params" => {},
-                  "output_payload" => nil,
-                  "workflow_id" => an_instance_of(String)
-              },
-              {
-                  "name"=>a_string_starting_with('PersistFirstJob'),
-                  "klass"=>"PersistFirstJob",
-                  "incoming"=>["FetchFirstJob"],
-                  "outgoing"=>[],
-                  "finished_at"=>nil,
-                  "started_at"=>nil,
-                  "enqueued_at"=>nil,
-                  "failed_at"=>nil,
-                  "params" => {},
-                  "output_payload" => nil,
-                  "workflow_id" => an_instance_of(String)
-              }
-          ]
+          "arguments" => ["arg1", "arg2"]
       }
       expect(result).to match(expected)
     end