[frontend] Publish events to a Rabbit MQ bus

src/api/Gemfile

@@ -66,6 +66,8 @@ gem 'peek-dalli'
 gem 'peek-mysql2'
 # for kerberos authentication
 gem "gssapi", require: false
+# for sending events to rabbitmq
+gem 'bunny'
 
 group :development, :production do
   # to have the delayed job daemon

src/api/Gemfile.lock

@@ -51,6 +51,7 @@ GEM
     airbrake (7.0.2)
       airbrake-ruby (~> 2.4)
     airbrake-ruby (2.4.0)
+    amq-protocol (2.2.0)
     annotate (2.7.2)
       activerecord (>= 3.2, < 6.0)
       rake (>= 10.4, < 13.0)
@@ -63,6 +64,8 @@ GEM
     bullet (5.6.1)
       activesupport (>= 3.0.0)
       uniform_notifier (~> 1.10.0)
+    bunny (2.7.1)
+      amq-protocol (>= 2.2.0)
     capybara (2.15.3)
       addressable
       mini_mime (>= 0.1.3)
@@ -414,6 +417,7 @@ DEPENDENCIES
   bcrypt
   bullet
   bundler
+  bunny
   capybara
   capybara_minitest_spec
   chunky_png

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

@@ -273,6 +273,12 @@ def _roles(role, project, package = nil)
       obj_roles(p, role)
     end
 
+    def send_to_bus
+      RabbitmqBus.publish(self.class.message_bus_queue, read_attribute(:payload))
+    rescue Bunny::Exception => e
+      logger.error "Publishing to RabbitMQ failed: #{e.message}"
+    end
+
     private
 
     def shorten_payload_if_necessary

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

@@ -32,6 +32,11 @@ def custom_headers
 class Event::CommentForProject < ::Event::Project
   include CommentEvent
   receiver_roles :maintainer
+  after_commit :send_to_bus
+
+  def self.message_bus_queue
+    'opensuse.obs.project.comment'
+  end
 
   self.description = 'New comment for project created'
 
@@ -43,6 +48,11 @@ def subject
 class Event::CommentForPackage < ::Event::Package
   include CommentEvent
   receiver_roles :maintainer
+  after_commit :send_to_bus
+
+  def self.message_bus_queue
+    'opensuse.obs.package.comment'
+  end
 
   self.description = 'New comment for package created'
 

src/api/app/models/rabbitmq_bus.rb

@@ -0,0 +1,23 @@
+class RabbitmqBus
+  def self.publish(event_queue_name, event_payload)
+    return unless CONFIG['amqp_options']
+    start_connection
+
+    queue = $rabbitmq_channel.queue(event_queue_name, CONFIG['amqp_queue_options'].try(:with_indifferent_access) || {})
+    $rabbitmq_exchange.publish(event_payload, routing_key: queue.name)
+  end
+
+  # Start one connection, channel and exchange per rails process
+  # and reuse them
+  def self.start_connection
+    $rabbitmq_conn ||= Bunny.new(CONFIG['amqp_options'].with_indifferent_access)
+    $rabbitmq_conn.start
+    $rabbitmq_channel ||= $rabbitmq_conn.create_channel
+    $rabbitmq_exchange = if CONFIG['amqp_exchange_name']
+      $rabbitmq_channel.exchange(CONFIG['amqp_exchange_name'], CONFIG['amqp_exchange_options'].try(:with_indifferent_access) || {})
+    else
+      $rabbitmq_channel.default_exchange
+    end
+  end
+  private_class_method :start_connection
+end

src/api/config/options.yml.example

@@ -159,3 +159,30 @@ backup_user: tux
 backup_location: /home/tux
 backup_filename: obs_production.sql.bz2
 backup_port: 22
+
+# Rabbitmq based message bus
+#
+# Connection options -> http://rubybunny.info/articles/connecting.html
+# amqp_options:
+#   host: rabbit.example.com
+#   port: 5672
+#   user: guest
+#   pass: guest
+#   vhost: vhost
+#
+# Exchange options -> http://rubybunny.info/articles/exchanges.html
+# amqp_exchange_name: pubsub
+# amqp_exchange_options:
+#  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