#823: update PublishObserver

lib/abf_worker/build_lists_publish_task_manager.rb

@@ -33,6 +33,15 @@ module AbfWorker
       end
     end
 
+    def self.unlock_build_list(build_list)
+      Resque.redis.lrem(LOCKED_BUILD_LISTS, 0, build_list.id)
+    end
+
+    def self.unlock_rep_and_platform(build_list)
+      key = "#{build_list.save_to_repository_id}-#{build_list.build_for_platform_id}"
+      Resque.redis.lrem(LOCKED_REP_AND_PLATFORMS, 0, key)
+    end
+
     private
 
     def create_task(save_to_repository_id, build_for_platform_id)
@@ -71,20 +80,25 @@ module AbfWorker
 
       packages      = {:sources => [], :binaries => {:x86_64 => [], :i586 => []}}
       old_packages  = packages.clone
+      build_list_ids = []
 
       build_lists.each do |bl|
         fill_packages(bl, packages)
         bl.last_published.includes(:packages).limit(5).each{ |old_bl|
           fill_packages(old_bl, old_packages)
         }
+        build_list_ids << bl.id
         @redis.lpush(LOCKED_BUILD_LISTS, bl.id)
       end
-      options.merge!({:packages => packages, :old_packages => old_packages})
 
       Resque.push(
         worker_queue,
         'class' => worker_class,
-        'args' => [options]
+        'args' => [options.merge({
+          :packages => packages,
+          :old_packages => old_packages,
+          :build_list_ids => build_list_ids
+        })]
       )
       return true
     end

lib/abf_worker/publish_observer.rb

@@ -3,26 +3,27 @@ module AbfWorker
     @queue = :publish_observer
 
     def self.perform(options)
-      bl = BuildList.find options['id']
+      build_lists = BuildList.where(:id => options['build_list_ids'])
       status = options['status'].to_i
-      case status
+      build_lists.each do |bl| 
-      when COMPLETED
-        bl.published
-        update_results(bl, options)
-      when FAILED
-        bl.fail_publish
-        update_results(bl, options)
-      when CANCELED
-        bl.fail_publish
         update_results(bl, options)
+        case status
+        when COMPLETED
+          bl.published
+        when FAILED
+          bl.fail_publish
+        when CANCELED
+          bl.fail_publish
+        end
+        AbfWorker::BuildListsPublishTaskManager.unlock_build_list bl
       end
+      AbfWorker::BuildListsPublishTaskManager.unlock_rep_and_platform build_lists.first
     end
 
     def self.update_results(subject, options)
       results = (subject.results || []).
-        map{ |r| r if r['file_name'] !~ /^abfworker\:\:publish\-worker.*\.log$/ }.
+        select{ |r| r['file_name'] !~ /^abfworker\:\:publish\-worker.*\.log$/ }
-        compact
+      results |= options['results']
-      results += options['results']
       sort_results_and_save(subject, results)
     end