+ @connection_count = 0
+ end
+
+ # Push out any pending events to the connection +ws+
+ # +notify_id+ the id of the most recent row in the log table, may be nil
+ #
+ # This accepts a websocket and a notify_id (this is the row id from Postgres
+ # LISTEN/NOTIFY, it may be nil if called from somewhere else)
+ #
+ # It queries the database for log rows that are either
+ # a) greater than ws.last_log_id, which is the last log id which was a candidate to be sent out
+ # b) if ws.last_log_id is nil, then it queries the row notify_id
+ #
+ # Regular Arvados permissions are applied using readable_by() and filters using record_filters().
+ def push_events ws, notify_id
+ begin
+ # Must have at least one filter set up to receive events
+ if ws.filters.length > 0
+ # Start with log rows readable by user, sorted in ascending order
+ logs = Log.readable_by(ws.user).order("id asc")
+
+ cond_id = nil
+ cond_out = []
+ param_out = []
+
+ if not ws.last_log_id.nil?
+ # We are catching up from some starting point.
+ cond_id = "logs.id > ?"
+ param_out << ws.last_log_id
+ elsif not notify_id.nil?
+ # Get next row being notified.
+ cond_id = "logs.id = ?"
+ param_out << notify_id
+ else
+ # No log id to start from, nothing to do, return
+ return
+ end
+
+ # Now build filters provided by client
+ ws.filters.each do |filter|
+ ft = record_filters filter.filters, Log
+ if ft[:cond_out].any?
+ # Join the clauses within a single subscription filter with AND
+ # so it is consistent with regular queries
+ cond_out << "(#{ft[:cond_out].join ') AND ('})"
+ param_out += ft[:param_out]
+ end
+ end
+
+ # Add filters to query
+ if cond_out.any?
+ # Join subscriptions with OR
+ logs = logs.where(cond_id + " AND ((#{cond_out.join ') OR ('}))", *param_out)
+ else
+ logs = logs.where(cond_id, *param_out)
+ end
+
+ # Execute query and actually send the matching log rows
+ logs.each do |l|
+ if not ws.sent_ids.include?(l.id)
+ # only send if not a duplicate
+ ws.send(l.as_api_response.to_json)
+ end
+ if not ws.last_log_id.nil?
+ # record ids only when sending "catchup" messages, not notifies
+ ws.sent_ids << l.id
+ end
+ end
+ ws.last_log_id = nil
+ end
+ rescue ArgumentError => e
+ # There was some kind of user error.
+ Rails.logger.warn "Error publishing event: #{$!}"
+ ws.send ({status: 500, message: $!}.to_json)
+ ws.close
+ rescue => e
+ Rails.logger.warn "Error publishing event: #{$!}"
+ Rails.logger.warn "Backtrace:\n\t#{e.backtrace.join("\n\t")}"
+ ws.send ({status: 500, message: $!}.to_json)
+ ws.close
+ # These exceptions typically indicate serious server trouble:
+ # out of memory issues, database connection problems, etc. Go ahead and
+ # crash; we expect that a supervisor service like runit will restart us.
+ raise
+ end