+ # Handle inbound subscribe or unsubscribe message.
+ def handle_message ws, event
+ begin
+ begin
+ # Parse event data as JSON
+ p = SafeJSON.load(event.data).symbolize_keys
+ filter = Filter.new(p)
+ rescue Oj::Error => e
+ send_message(ws, {status: 400, message: "malformed request"})
+ return
+ end
+
+ if p[:method] == 'subscribe'
+ # Handle subscribe event
+
+ if p[:last_log_id]
+ # Set or reset the last_log_id. The event bus only reports events
+ # for rows that come after last_log_id.
+ ws.last_log_id = p[:last_log_id].to_i
+ # Reset sent_ids for consistency
+ # (always re-deliver all matching messages following last_log_id)
+ ws.sent_ids = Set.new
+ end
+
+ if ws.filters.length < Rails.configuration.websocket_max_filters
+ # Add a filter. This gets the :filters field which is the same
+ # format as used for regular index queries.
+ ws.filters << filter
+ send_message(ws, {status: 200, message: 'subscribe ok', filter: p})
+
+ # Send any pending events
+ push_events ws, nil
+ else
+ send_message(ws, {status: 403, message: "maximum of #{Rails.configuration.websocket_max_filters} filters allowed per connection"})
+ end
+
+ elsif p[:method] == 'unsubscribe'
+ # Handle unsubscribe event
+
+ len = ws.filters.length
+ ws.filters.select! { |f| not ((f.filters == p[:filters]) or (f.filters.empty? and p[:filters].nil?)) }
+ if ws.filters.length < len
+ send_message(ws, {status: 200, message: 'unsubscribe ok'})
+ else
+ send_message(ws, {status: 404, message: 'filter not found'})
+ end
+
+ else
+ send_message(ws, {status: 400, message: "missing or unrecognized method"})
+ end
+ rescue => e
+ Rails.logger.warn "Error handling message: #{$!}"
+ Rails.logger.warn "Backtrace:\n\t#{e.backtrace.join("\n\t")}"
+ send_message(ws, {status: 500, message: 'error'})
+ ws.close
+ end
+ end
+
+ def overloaded?
+ @mtx.synchronize do
+ @connection_count >= Rails.configuration.websocket_max_connections
+ end
+ end