Merge pull request #4027 from coolo/publish_on_exchange

[frontend] Publish on the rabbitmq exchange

src/api/app/models/event/base.rb

@@ -79,7 +79,7 @@ def inherited(subclass)
         subclass.receiver_roles(*receiver_roles)
       end
 
-      def message_bus_queue
+      def message_bus_routing_key
         raise NotImplementedError
       end
     end
@@ -265,7 +265,7 @@ def _roles(role, project, package = nil)
     end
 
     def send_to_bus
-      RabbitmqBus.publish(self.class.message_bus_queue, read_attribute(:payload))
+      RabbitmqBus.publish(self.class.message_bus_routing_key, read_attribute(:payload))
     rescue Bunny::Exception => e
       logger.error "Publishing to RabbitMQ failed: #{e.message}"
     end

src/api/app/models/event/build.rb

@@ -24,7 +24,7 @@ class Event::BuildSuccess < Event::Build
   self.description = 'Package has succeeded building'
   after_create_commit :send_to_bus
 
-  def self.message_bus_queue
+  def self.message_bus_routing_key
     "#{Configuration.amqp_namespace}.package.build_success"
   end
 end
@@ -36,7 +36,7 @@ class Event::BuildFail < Event::Build
   receiver_roles :maintainer, :bugowner, :reader, :watcher
   after_create_commit :send_to_bus
 
-  def self.message_bus_queue
+  def self.message_bus_routing_key
     "#{Configuration.amqp_namespace}.package.build_fail"
   end
 
@@ -83,7 +83,7 @@ class Event::BuildUnchanged < Event::Build
   self.description = 'Package has succeeded building with unchanged result'
   after_create_commit :send_to_bus
 
-  def self.message_bus_queue
+  def self.message_bus_routing_key
     "#{Configuration.amqp_namespace}.package.build_unchanged"
   end
 end

src/api/app/models/event/comment.rb

@@ -34,7 +34,7 @@ class Event::CommentForProject < ::Event::Project
   receiver_roles :maintainer, :watcher
   after_create_commit :send_to_bus
 
-  def self.message_bus_queue
+  def self.message_bus_routing_key
     "#{Configuration.amqp_namespace}.project.comment"
   end
 
@@ -50,7 +50,7 @@ class Event::CommentForPackage < ::Event::Package
   receiver_roles :maintainer, :watcher
   after_create_commit :send_to_bus
 
-  def self.message_bus_queue
+  def self.message_bus_routing_key
     "#{Configuration.amqp_namespace}.package.comment"
   end
 
@@ -68,7 +68,7 @@ class Event::CommentForRequest < ::Event::Request
   receiver_roles :source_maintainer, :target_maintainer, :creator, :reviewer, :source_watcher, :target_watcher
   after_create_commit :send_to_bus
 
-  def self.message_bus_queue
+  def self.message_bus_routing_key
     "#{Configuration.amqp_namespace}.request.comment"
   end
 

src/api/app/models/event/package.rb

@@ -8,7 +8,7 @@ class CreatePackage < Package
     self.description = 'Package was created'
     after_create_commit :send_to_bus
 
-    def self.message_bus_queue
+    def self.message_bus_routing_key
       "#{Configuration.amqp_namespace}.package.create"
     end
 
@@ -21,7 +21,7 @@ class UpdatePackage < Package
     self.description = 'Package meta data was updated'
     after_create_commit :send_to_bus
 
-    def self.message_bus_queue
+    def self.message_bus_routing_key
       "#{Configuration.amqp_namespace}.package.update"
     end
   end
@@ -31,7 +31,7 @@ class UndeletePackage < Package
     payload_keys :comment
     after_create_commit :send_to_bus
 
-    def self.message_bus_queue
+    def self.message_bus_routing_key
       "#{Configuration.amqp_namespace}.package.undelete"
     end
 
@@ -47,7 +47,7 @@ class DeletePackage < Package
     payload_keys :comment, :requestid
     after_create_commit :send_to_bus
 
-    def self.message_bus_queue
+    def self.message_bus_routing_key
       "#{Configuration.amqp_namespace}.package.delete"
     end
 
@@ -62,7 +62,7 @@ class BranchCommand < Package
     payload_keys :targetproject, :targetpackage, :user
     after_create_commit :send_to_bus
 
-    def self.message_bus_queue
+    def self.message_bus_routing_key
       "#{Configuration.amqp_namespace}.package.branch"
     end
 
@@ -76,7 +76,7 @@ class VersionChange < Package
     payload_keys :comment, :requestid, :files, :rev, :newversion, :user, :oldversion
     after_create_commit :send_to_bus
 
-    def self.message_bus_queue
+    def self.message_bus_routing_key
       "#{Configuration.amqp_namespace}.package.version_change"
     end
 
@@ -94,7 +94,7 @@ class Commit < Package
     create_jobs :update_backend_infos_job
     after_create_commit :send_to_bus
 
-    def self.message_bus_queue
+    def self.message_bus_routing_key
       "#{Configuration.amqp_namespace}.package.commit"
     end
 
@@ -114,7 +114,7 @@ class Upload < Package
     payload_keys :project, :package, :comment, :filename, :requestid, :target, :user
     after_create_commit :send_to_bus
 
-    def self.message_bus_queue
+    def self.message_bus_routing_key
       "#{Configuration.amqp_namespace}.package.upload"
     end
   end
@@ -126,7 +126,7 @@ class ServiceSuccess < Package
     create_jobs :update_backend_infos_job
     after_create_commit :send_to_bus
 
-    def self.message_bus_queue
+    def self.message_bus_routing_key
       "#{Configuration.amqp_namespace}.package.service_success"
     end
 
@@ -148,7 +148,7 @@ class ServiceFail < Package
     create_jobs :update_backend_infos_job
     after_create_commit :send_to_bus
 
-    def self.message_bus_queue
+    def self.message_bus_routing_key
       "#{Configuration.amqp_namespace}.package.service_fail"
     end
 

src/api/app/models/event/packtrack.rb

@@ -6,7 +6,7 @@ class Event::Packtrack < Event::Base
   create_jobs :update_released_binaries_job
   after_create_commit :send_to_bus
 
-  def self.message_bus_queue
+  def self.message_bus_routing_key
     "#{Configuration.amqp_namespace}.repo.packtrack"
   end
 end

src/api/app/models/event/project.rb

@@ -9,7 +9,7 @@ class CreateProject < Project
     payload_keys :sender
     after_create_commit :send_to_bus
 
-    def self.message_bus_queue
+    def self.message_bus_routing_key
       "#{Configuration.amqp_namespace}.project.create"
     end
 
@@ -23,7 +23,7 @@ class UpdateProjectConfig < Project
     payload_keys :sender, :files, :comment
     after_create_commit :send_to_bus
 
-    def self.message_bus_queue
+    def self.message_bus_routing_key
       "#{Configuration.amqp_namespace}.project.update_project_conf"
     end
   end
@@ -33,7 +33,7 @@ class UndeleteProject < Project
     payload_keys :comment, :sender
     after_create_commit :send_to_bus
 
-    def self.message_bus_queue
+    def self.message_bus_routing_key
       "#{Configuration.amqp_namespace}.project.undelete"
     end
   end
@@ -43,7 +43,7 @@ class UpdateProject < Project
     payload_keys :sender
     after_create_commit :send_to_bus
 
-    def self.message_bus_queue
+    def self.message_bus_routing_key
       "#{Configuration.amqp_namespace}.project.update"
     end
   end
@@ -53,7 +53,7 @@ class DeleteProject < Project
     payload_keys :comment, :requestid, :sender
     after_create_commit :send_to_bus
 
-    def self.message_bus_queue
+    def self.message_bus_routing_key
       "#{Configuration.amqp_namespace}.project.delete"
     end
   end

src/api/app/models/event/repo_publish_state.rb

@@ -3,7 +3,7 @@ class Event::RepoPublishState < Event::Base
   payload_keys :project, :repo, :state
   after_create_commit :send_to_bus
 
-  def self.message_bus_queue
+  def self.message_bus_routing_key
     "#{Configuration.amqp_namespace}.repo.publish_state"
   end
 end

src/api/app/models/event/repo_published.rb

@@ -3,7 +3,7 @@ class Event::RepoPublished < Event::Base
   payload_keys :project, :repo
   after_create_commit :send_to_bus
 
-  def self.message_bus_queue
+  def self.message_bus_routing_key
     "#{Configuration.amqp_namespace}.repo.published"
   end
 end

src/api/app/models/event/request.rb

@@ -141,7 +141,7 @@ class Event::RequestChange < Event::Request
   self.description = 'Request XML was updated (admin only)'
   after_create_commit :send_to_bus
 
-  def self.message_bus_queue
+  def self.message_bus_routing_key
     "#{Configuration.amqp_namespace}.request.change"
   end
 end
@@ -151,7 +151,7 @@ class Event::RequestCreate < Event::Request
   receiver_roles :source_maintainer, :target_maintainer, :source_watcher, :target_watcher
   after_create_commit :send_to_bus
 
-  def self.message_bus_queue
+  def self.message_bus_routing_key
     "#{Configuration.amqp_namespace}.request.create"
   end
 
@@ -176,7 +176,7 @@ class Event::RequestDelete < Event::Request
   self.description = 'Request was deleted (admin only)'
   after_create_commit :send_to_bus
 
-  def self.message_bus_queue
+  def self.message_bus_routing_key
     "#{Configuration.amqp_namespace}.request.delete"
   end
 end
@@ -187,7 +187,7 @@ class Event::RequestStatechange < Event::Request
   receiver_roles :source_maintainer, :target_maintainer, :creator, :reviewer, :source_watcher, :target_watcher
   after_create_commit :send_to_bus
 
-  def self.message_bus_queue
+  def self.message_bus_routing_key
     "#{Configuration.amqp_namespace}.request.state_change"
   end
 
@@ -203,7 +203,7 @@ class Event::ReviewWanted < Event::Request
   receiver_roles :reviewer
   after_create_commit :send_to_bus
 
-  def self.message_bus_queue
+  def self.message_bus_routing_key
     "#{Configuration.amqp_namespace}.request.review_wanted"
   end
 

src/api/config/options.yml.example

@@ -176,16 +176,8 @@ backup_port: 22
 # Exchange options -> http://rubybunny.info/articles/exchanges.html
 # amqp_exchange_name: pubsub
 # amqp_exchange_options:
-#  type: topic
+#  type: :topic
 #  auto_delete: false
 #  arguments:
 #    persistent: true
 #    passive: true
-#
-# Queue options -> http://rubybunny.info/articles/queues.html
-# amqp_queue_options:
-#   durable: false
-#   auto-delete: false
-#   exclusive: false
-#   arguments:
-#     extension_1: blah

src/api/lib/rabbitmq_bus.rb

@@ -1,10 +1,9 @@
 class RabbitmqBus
-  def self.publish(event_queue_name, event_payload)
+  def self.publish(event_routing_key, event_payload)
     return unless CONFIG['amqp_options']
     start_connection
 
-    queue = $rabbitmq_channel.queue(event_queue_name, CONFIG['amqp_queue_options'].try(:symbolize_keys) || {})
-    $rabbitmq_exchange.publish(event_payload, routing_key: queue.name)
+    $rabbitmq_exchange.publish(event_payload, routing_key: event_routing_key)
   end
 
   # Start one connection, channel and exchange per rails process