Ruby on Rails: track model changes from controller action

Question

I want to generate events sent by the server whenever the update action in the controller is called (or whenever my model is updated). I currently have a valid watch action in the same controller with a dummy emitter:

 def watch self.response.headers["Content-Type"] = "text/event-stream" self.response.headers["Last-Modified"] = Time.now.ctime.to_json self.response_body = Enumerator.new do |y| 100.times do |i| sleep 5 y << ["event: message", "data: #{i}\n\n"].join("\n") end end # TODO catch IO error when client disconnects end 

How can I get an enumerated object that returns / returns a value when update called? (NB: it really isn't necessarily Enumerable, but it must answer before #each for this streaming technology to work.) In some ways, I'm trying to implement an event-driven architecture in Rails, I think.

I know Observable , but I cannot figure out how to get my observer to be enumerated as required for this ... or how to put an observer in an Enumerator (as mentioned above) without a loop and sleep timer.

The goal is to make changes to the database sent to all other users who are currently logged in, so that each user always has the current reflection of the database.

Thanks -

Answer 1

This is not necessarily the best solution, but I created the message queue as follows: I created a new table ("SSE") in my database, and in the model I wanted to watch, I added callbacks

 class MyModel < ActiveRecord::Base after_save :queue_sse after_delete :queue_sse # ... private def queue_sse connected_users.each do |user| SSE.create(:changed_record_id => this.id, :user_id => user) end end end 

Then in watch action:

 def watch connected_users << current_user # pseudo for a mutex-synched accessor self.response.headers["Content-Type"] = "text/event-stream" self.response.headers["Last-Modified"] = Time.now.ctime.to_json self.response_body = Enumerator.new do |y| loop do sleep 5 ActiveRecord::Base.uncached do updates = SSE.find_all_by_user_id(current_user) updates.each do |update| puts "update found: #{update.id}\n" y << ["event: message", "data: #{update.id}\n\n"].join("\n") update.destroy end end end end # TODO add error catching, and on IOError, remove the current_user end 

This very often gets into the database. It should probably be built on memcached, a mutex class variable or similar.

(NB - requires streaming, for example, config.threadsafe! And a streaming server, of course.)

Answer 2

I created a gem that allows you to "broadcast" events using SSE. It allows you to push data to connected clients from the update action. It is called R4S ( https://github.com/biggihs/r4s )

Example

 def update #do some updating R4S.push_data("key",{data:"data"},:event=>"SomeJsEvent") end 

This will cause the data to be transferred to all browsers connected to the sse stream keyword