A Rack-based event receiver for the Routemaster event bus.
routemaster-drain
is a collection of Rack middleware to receive and
parse Routemaster events, filter them, and preemptively cache the corresponding
resources.
It provides prebuilt middleware stacks (Basic
, Mapping
, and Caching
) for
typical use cases, illustrated below, or you can easily roll your own by
combining middleware.
Add this line to your application's Gemfile:
gem 'routemaster-drain'
Configuration
This gem is configured through the environment, making 12factor compliance easier.
Required:
ROUTEMASTER_DRAIN_TOKENS
: a comma-separated list of valid authentication tokens, used by Routemaster to send you events.
Optional:
ROUTEMASTER_DRAIN_REDIS
: the URL of the Redis instance used for filtering and dirty mapping. Required if you use either feature, ignored otherwise. A namespace can be specified. Example:redis://user:s3cr3t@myhost:1234/12/ns
.ROUTEMASTER_CACHE_REDIS
: the URL of the Redis instance used for caching. Required if you use the feature, ignored otherwise. Formatted likeROUTEMASTER_DRAIN_REDIS
.ROUTEMASTER_CACHE_EXPIRY
: if using the cache, for how long to cache entries, in seconds. Default 1 year (31,536,000).ROUTEMASTER_CACHE_AUTH
: if using the cache, specifies what username/password pairs to use to fetch resources. The format is a comma-separated list of colon-separate lists of regexp, username, password values. Example:server1:user:p4ss,server2:user:p4ass
.ROUTEMASTER_QUEUE_NAME
: if using the cache, on which Resque queue the cache population jobs should be enqueued.ROUTEMASTER_CACHE_TIMEOUT
: if using the cache, how long before Faraday will timeout fetching the resource. Defaults to 1 second.ROUTEMASTER_CACHE_VERIFY_SSL
: if using the cache, whether to verify SSL when fetching the resource. Defaults to false.
Provide a listener for events:
class Listener
def on_events_received(batch)
batch.each do |event|
puts event.url
end
end
end
Each event is a Hashie::Mash
and responds to type
(one of create
,
update
, delete
, or noop
), url
(the resource), and t
(the event
timestamp, in milliseconds since the Epoch).
Create the app that will process events:
require 'routemaster/drain/basic'
$app = Routemaster::Drain::Basic.new
Bind the app to your listener:
$app.subscribe(Listener.new, prefix: true)
And finally, mount your app to your subscription path:
# typically in config.ru
map '/events' do
run $app
end
This relies on the excellent event bus from the wisper gem.
When reacting to changes of some resource, it's common to want to avoid receiving further change notifications until you've actually processed that resource.
Possibly you'll want to process changes in batches at regular time intervals.
For this purpose, use Routemaster::Drain::Mapping
:
require 'routemaster/drain/mapping'
$app = Routemaster::Drain::Mapping.new
And mount it as usual:
# in config.ru
map('/events') { run $app }
Instead of processing events, you'll check for changes in the dirty map:
require 'routemaster/dirty/map'
$map = Routemaster::Dirty::Map.new
every_5_minutes do
$map.sweep do |url|
# do something about this changed resource
true
end
end
Until you've called #sweep
and your block has returned true
, you won't be
bugged again — the dirty map acts as a buffer of changes (see below for
internals).
Notes:
- You can limit the number of resources to be swept (
$map.sweep(123) { ... }
). - You can count the number of resources to be swept with
$map.count
. - You're not told what is to be swept; entities won't be swept in the order of events received (much like Routemaster does not guarantee ordering).
- If your sweeper fails, the dirty map will not be cleaned, so you can have
leftovers. It's good practice to regularly run
$map.sweep { ... }
and perform cleanup regularly. - The map won't tell you if the resources has been changed, created, or deleted. You'll have to figure it out with an API call.
- You can still attach a listener to the app to get all events.
Another common use case is that you'll actually need the representation of the resources Routemaster tells you about.
The Caching
prebuilt app can do that for you, using Resque to populate the
cache as events are received.
For this purpose, use Routemaster::Drain::Caching
:
require 'routemaster/drain/machine'
$app = Routemaster::Drain::Caching.new
And mount it as usual:
# in config.ru
map('/events') { run $app }
You can still attach a listenenr if you want the incoming events. Typically, what you'll want is the cache:
require 'routemaster/cache'
$cache = Routemaster::Cache.new
response = @cache.fget('https://example.com/widgets/123')
puts response.body.id
In this example, is your app was notified by Routemaster about Widget #123, the cache will be very likely to be hit; and it will be invalidated automatically whenever the drain gets notified about a change on that widget.
Note that Cache#fget
is a future, so you can efficiently query many resources
and have any HTTP GET
requests (and cache queries) happen in parallel.
See
rubydoc
for more details on Cache
.
The more elaborate drains are built with two components which can also be used
independently,
Dirty::Map
and
Dirty::Filter
.
A dirty map collects entities that have been created, updated, or deleted (or rather, their URLs). It can be used to delay your service's reaction to events, for instance combined with Resque.
A dirty map map gets marked when an event about en entity gets processed that indicates a state change, and swept to process those changes.
Practically, instances of
Routemaster::Dirty::Map
will emit a :dirty_entity
event when a URL is marked as dirty, and can be
swept when an entity is "cleaned". If a URL is marked multiple times before
being swept (e.g. for very volatile entities), the event will only by broadcast
once.
To sweep the map, you can for instance listen to this event and call
#sweep_one
.
If you're not in a hurry and would rather run through batches you can call
#sweep
which will yield URLs until it runs out of dirty resources.
Routemaster::Dirty::Filter
is a simple event filter
that performs reordering. It ignores events older than the latest known
information on an entity.
It stores transient state in Redis and will emit :entity_changed
events
whenever an entity has changed. This event can usefully be fed into a dirty map,
as in Receiver::Filter
for instance.
- Fork it ( http://github.com/HouseTrip/routemaster-drain/fork )
- Create your feature branch (
git checkout -b my-new-feature
) - Commit your changes (
git commit -am 'Add some feature'
) - Push to the branch (
git push origin my-new-feature
) - Create new Pull Request