+ rescue Oj::Error => e
+ ws.send ({status: 400, message: "malformed request"}.to_json)
+ rescue Exception => e
+ puts "Error handling message: #{$!}"
+ puts "Backtrace:\n\t#{e.backtrace.join("\n\t")}"
+ ws.send ({status: 500, message: 'error'}.to_json)
+ ws.close
+ end
+ end
+
+ # Constant maximum number of filters, to avoid silly huge database queries.
+ MAX_FILTERS = 16
+
+ # Called by RackSocket when a new websocket connection has been established.
+ def on_connect ws
+
+ # Disconnect if no valid API token.
+ # current_user is included from CurrentApiClient
+ if not current_user
+ ws.send ({status: 401, message: "Valid API token required"}.to_json)
+ ws.close
+ return
+ end
+
+ # Initialize our custom fields on the websocket connection object.
+ ws.user = current_user
+ ws.filters = []
+ ws.last_log_id = nil
+
+ # Subscribe to internal postgres notifications through @channel. This will
+ # call push_events when a notification comes through.
+ sub = @channel.subscribe do |msg|
+ push_events ws, msg
+ end
+
+ # Set up callback for inbound message dispatch.
+ ws.on :message do |event|
+ handle_message ws, event
+ end
+
+ # Set up socket close callback
+ ws.on :close do |event|
+ @channel.unsubscribe sub
+ ws = nil
+ end