[ARVADOS] created: 21ccdae70baaca0c7aeca8542f2a0a431f06c313
git at public.curoverse.com
git at public.curoverse.com
Thu Oct 23 15:29:15 EDT 2014
at 21ccdae70baaca0c7aeca8542f2a0a431f06c313 (commit)
commit 21ccdae70baaca0c7aeca8542f2a0a431f06c313
Author: Peter Amstutz <peter.amstutz at curoverse.com>
Date: Thu Oct 23 15:29:09 2014 -0400
4295: Fixed tests
diff --git a/services/api/lib/eventbus.rb b/services/api/lib/eventbus.rb
index 080138b..a0c0b27 100644
--- a/services/api/lib/eventbus.rb
+++ b/services/api/lib/eventbus.rb
@@ -123,7 +123,7 @@ class EventBus
# another query to get more logs (will start from last_log_id
# reported by current query)
@channel.push nil
- elsif !notify_id.nil? and notify_id > ws.last_log_id
+ elsif !notify_id.nil? and (ws.last_log_id.nil? or notify_id > ws.last_log_id)
# Number of rows returned was less than cap, but the notify id is
# higher than the last id visible to the client, so update last_log_id
ws.last_log_id = notify_id
commit 74e7f636d3ec2f3a20640b721a3fd76fd23cd788
Author: Peter Amstutz <peter.amstutz at curoverse.com>
Date: Thu Oct 23 15:22:21 2014 -0400
4295: First pass at more efficient database utilization. A few tests fail.
diff --git a/services/api/lib/eventbus.rb b/services/api/lib/eventbus.rb
index bccbeea..080138b 100644
--- a/services/api/lib/eventbus.rb
+++ b/services/api/lib/eventbus.rb
@@ -49,65 +49,95 @@ class EventBus
end
# Push out any pending events to the connection +ws+
- # +id+ the id of the most recent row in the log table, may be nil
- def push_events ws, id = nil
- 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 ws.last_log_id
- # Client is only interested in log rows that are newer than the
- # last log row seen by the client.
- cond_id = "logs.id > ?"
- param_out << ws.last_log_id
- elsif id
- # No last log id, so only look at the most recently changed row
- cond_id = "logs.id = ?"
- param_out << id.to_i
- else
- return
- end
+ # +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 nil)
+ # 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 rows starting with notify_id
+ #
+ # Regular Arvados permissions are applied using readable_by() and filters using record_filters()
+ # To avoid clogging up the database, queries are limited to batches of 100. It will schedule a new
+ # push_events call if there are more log rows to send.
+ def push_events ws, notify_id
+ begin
+ if !notify_id.nil? and !ws.last_log_id.nil? and notify_id <= ws.last_log_id
+ # This notify is for a row we've handled already.
+ return
+ end
- # Now process 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
+ # 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")
- # 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
+ cond_id = nil
+ cond_out = []
+ param_out = []
- # Finally execute query and actually send the matching log rows
- logs.each do |l|
- ws.send(l.as_api_response.to_json)
- ws.last_log_id = l.id
+ if !ws.last_log_id.nil?
+ # Client is only interested in log rows that are newer than the
+ # last log row seen by the client.
+ cond_id = "logs.id > ?"
+ param_out << ws.last_log_id
+ elsif !notify_id.nil?
+ # No last log id, so look at rows starting with notify id
+ 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
- elsif id
- # No filters set up, so just record the sequence number
- ws.last_log_id = id.to_i
end
- rescue Exception => e
- Rails.logger.warn "Error publishing event: #{$!}"
- Rails.logger.warn "Backtrace:\n\t#{e.backtrace.join("\n\t")}"
- ws.send ({status: 500, message: 'error'}.to_json)
- ws.close
+
+ # 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
+ count = 0
+ limit = 100
+
+ logs.limit(limit).each do |l|
+ ws.send(l.as_api_response.to_json)
+ ws.last_log_id = l.id
+ count += 1
+ end
+
+ if count == limit
+ # Number of rows returned was capped by limit(), we need to schedule
+ # another query to get more logs (will start from last_log_id
+ # reported by current query)
+ @channel.push nil
+ elsif !notify_id.nil? and notify_id > ws.last_log_id
+ # Number of rows returned was less than cap, but the notify id is
+ # higher than the last id visible to the client, so update last_log_id
+ ws.last_log_id = notify_id
+ end
+ elsif !notify_id.nil?
+ # No filters set up, so just record the sequence number
+ ws.last_log_id = notify_id
end
+ rescue => e
+ Rails.logger.warn "Error publishing event: #{$!}"
+ Rails.logger.warn "Backtrace:\n\t#{e.backtrace.join("\n\t")}"
+ ws.send ({status: 500, message: 'error'}.to_json)
+ ws.close
+ end
end
# Handle inbound subscribe or unsubscribe message.
@@ -132,7 +162,7 @@ class EventBus
ws.send ({status: 200, message: 'subscribe ok', filter: p}.to_json)
# Send any pending events
- push_events ws
+ push_events ws, nil
else
ws.send ({status: 403, message: "maximum of #{MAX_FILTERS} filters allowed per connection"}.to_json)
end
@@ -153,7 +183,7 @@ class EventBus
end
rescue Oj::Error => e
ws.send ({status: 400, message: "malformed request"}.to_json)
- rescue Exception => e
+ rescue => e
Rails.logger.warn "Error handling message: #{$!}"
Rails.logger.warn "Backtrace:\n\t#{e.backtrace.join("\n\t")}"
ws.send ({status: 500, message: 'error'}.to_json)
@@ -215,7 +245,7 @@ class EventBus
# channel and calls #push_events to actually dispatch the
# events to the client.
conn.wait_for_notify do |channel, pid, payload|
- @channel.push payload
+ @channel.push payload.to_i
end
end
ensure
-----------------------------------------------------------------------
hooks/post-receive
--
More information about the arvados-commits
mailing list