HermesMessengerOfTheGods
Installation
Add this line to your application's Gemfile:
gem 'hermes_messenger_of_the_gods', git: 'url_of_repo'
And then execute:
$ bundle
Or install it yourself as:
$ gem install hermes_messenger_of_the_gods
Usage
The Message
A HMOTG Message includes ActiveModel::Base, so ou get alot of the joys related to that. Initialization can be setup with:
Creating a Message
When creating a message the most common parameters passed are the attributes which represent the content of the message.
class Foo
include HermesMessengerOfTheGods::Concerns::Message
attr_accessor :name
end
Foo.new(name: 'Dude').name
# => Dude
Validations
HMOTG Supports the validation syntax provided by ActiveModel::Validations.
class Foo
include HermesMessengerOfTheGods::Concerns::Message
attr_accessor :name
validates :name, presence: true
end
instance = Foo.new(name: '')
instance.valid?
# => false
instance.errors
# => {name: ["can't be blank"]}
Configuring a Endpoint
Endpoints are the recepients of message, and are typically configured per message. The Endpoints section provides configuration details for each endpoint type.
Setting Endpoints
You may set a single endpoint for a mesage by assigning it to a class attribute endpoitns
class Foo
include HermesMessengerOfTheGods::Concerns::Message
attr_accessor :name
validates :name, presence: true
self.endpoints = {default: sns_endpoint("sns:arn", jitter: false) }
end
Alternatively you can set multiple endpoints using a hash syntax:
class Foo
include HermesMessengerOfTheGods::Concerns::Message
attr_accessor :name
validates :name, presence: true
self.endpoints = {
receiver1: sns_endpoint("sns:arn", jitter: false),
receiver2: sns_endpoint("sns:arn", jitter: true),
}
end
Important: When using multiple endpoints, if one endpoint of many fails, the entire message transmission is considered to have failed. Since we can't take a message back from an endpoint we can't do too much about it.
Sending the message
Messages can be dispatched to the respective endpoints using the dispatch
or
the error raising dispatch!
method. By default the Message also includes a
to_message
method which returns a hash of the object's attributes.
class Foo
include HermesMessengerOfTheGods::Concerns::Message
attr_accessor :name
validates :name, presence: true
self.endpoints = sns_endpoint("sns:arn", jitter: false)
end
Foo.new(name: 'omg').dispatch!
# => true
The Worker
The worker provides a generic class to provide access for reading from endpoints. Only some endpoints actually provide the ability to read messages.
Configuration
Workers can be configured with a single endpoint and a single MessageClass for deserialization.
class MyWorker
include HermesMessengerOfTheGods::Concerns::Worker
self.endpoint = sns_endpoint("sns::arn", jitter: false)
self.deserialize_with = MessageType
end
Getting the work done
When working off a messages coming from an endpoint, there are a few possible paths the message can take.
- You may define a
perform
method on the worker which takes the message instance as the only argument. - If a
perform
method is not defined on the worker, aperform
method is expected to be defined on the message itself. This method is called with no arguments. - If no
perform
method is found on the Worker or the Message an error is raised.
Note: A perform
method on the worker is perfered to a perform method on
the message.
Error Handling
When a messsage fails to run, the handle_failure
on the worker is invoked.
This method is called with two arguments:
- The message itself
- The exception itself.
By default this method simply passes the job and exception to the endpoint so
it can be handled as appropiate. If you overload this method you probably want
to ensure you call super
.
Message Deserialization
Messages coming off the endpoints need to be deserialized into the domain message. By default we expect you are deserializing into a HermesMessengerOfTheGods::Concerns::Message, but it can be anything really.
In order to turn the received message into the Object we expect a class method
.from_message
to be defined. The HermesMessengerOfTheGods::Message superclass
defines a basic one that expects all attributes present in the message to be
attr_accessor
s on the message object.
The method used to create the message object can be overloaded by setting
deserialize_method
class variable of the worker.
class MessageType
include HermesMessengerOfTheGods::Concerns::Message
attr_accessor :dude, :hrm
delf self.omg_dude(msg)
new(dude: msg["old_dude"], hrm: nil)
end
end
class AwesomeWorker < HermesMessengerOfTheGods::Worker
self.endpoint = sns_endpoint("sns::arn", jitter: false)
self.deserialize_with = MessageType
self.deserialize_method = :omg_dude
end
If the endpoint emits a hash with only the keys :dude
and :hrm
you can get
away without defining any deserialize_with
setting. However if you need to do
some translations, you can define the custom builder function such as omg_dude
above.
Execution
For a single queue ./bin/fly_hermes start --worker={worker job}
For pooled operation `./bin/fly_hermes start --pool job--count job--count job--count
Message & Worker in one class
When your message meets a few constraints you can combine the logic of the worker and the message into a single class.
- You may only use one endpoint (due to limitation on the worker)
- The endpoint must be both readable and writable (No SNS for example)
- You do not need to customize callbacks much
class MonoMessageWorker
include HermesMessengerOftheGods::Concerns::MonoMessage
self.endpoint = sqs_endpoint("arn")
def perform
# Do magic Here
end
end
You may then start the message execution with:
./bin/fly_hermes start --pool=MonoMessageWorker--1
rotocol Buffers
Hermes supports using rotoBuffers over the wire using the
HermesMessengerOfTheGods::Concerns::GrpcProtobuf
mixin. Currently only JSON
encoding is allowed.
Example:
class MessageA
include HermesMessengerOfTheGods::Concerns::Message
include HermesMessengerOfTheGods::Concerns::GrpcProtobuf
self.protobuf_class = Helloworld::HelloRequest
end
# Or
class MessageB
include HermesMessengerOftheGods::Concerns::MonoMessage
include HermesMessengerOfTheGods::Concerns::GrpcProtobuf
self.protobuf_class = Helloworld::HelloRequest
end
This will use the Protobuf encoder and decoder for transmission on the wire.
Notes:
1) Message initialization now requires the first parameter be an instance
of the provided protobuf.
2) If you are are using Dispatch helpers, you probably need to define a
specific builder.
3) If you are lazy, you can use .from_message
and pass a hash.
Health Check HTTP Server
Hermes comes with a built in HTTP Health check server that may be optionally enabled. The server responds with 200 OK
when everything is going okay, and 500 Internal Server Error
when things seem out of wack.
The basic logic of the sever is as follows:
1) The server becomes unhealthy if no work has been performed in 60 seconds (customizable with the environment
variable HERMES_MINIMUM_WORK_FREQUENCY
) AND the endpoint reports there is work to do.
2) The server always reports healthy when there is no pending work to be done.
Pending work for an SQS queue is when there is at least one message visible in the queue.
In order to enable the HTTP Health Check server, you should set the environment variable ENABLE_HERMES_HEALTH_CHECK
to
and value except false
. The port the server listens on is configurable using HERMES_WORKER_STATUS_PORT
defaulting to 4242.
Endpoints
Endpoints provide the external communication layer to the outside world. Generally speaking all endpoints should inherit from the base endpoint to provide common behaviors.
Endpoint creation takes two parameters, the first is the "endpoint", the second is a options hash. Endpoint is required, while options provide more fine grained control of the endpoint behavior. The content of both parameters is dependent on the backing endpoint.
General Endpoint Behavior
Each end point needs to define a transmit
method. Beyond that the basic
behavior common to all end points will be automatically added.
Dispatching a message to an endpoint
Dispatching a message into a endpoint can be accomplished by calling dispatch
and passing the message in as the first parameter.
dispatch(message)
returns true or false to represent success or failuredispatch!(message)
returns true on success, and raises the last error received on failure
Message Transformations
The endpoint will request the content to deliver from the message object itself. A heirachy of methods will be used to determine wich method to use.
- You may provide an options during creation of the endpoint,
:transformer
and that will be used. This can be a proc, or a method name. The proc will be called with the object as the first parameter to_<endpoint_type>_message
- For any endpoint, the class name will be transformed into a dynamic method call. For example the SnsFoo endpoint will look forto_sns_foo_message
.to_message
- Generic to_message handler for the class.- Finally the transmitted message itself. You probably don't want this since most message objects do not serialize well.
Backoffs
By default a linear backoff is used of 1 second per number of tries. You can
configure this by passing the :backoff
option with :linear
or
:exponential
. If you prefer to write your own backoff, you can also pass a
proc to the backoff method. A :backoff
value of nil will disable any backoffs.
Jitter is added by default to the backoff time. This can be disabled by passing
jitter: false
in with the options.
Error Handling
By default all standard errors are caught and retried, with the exceptoin of
HermesMessengerOfTheGods::Endpoints::FatalError
. If this error is raised
there will be no further retries.
All errors raised during execution will be stored in the endpoint.errors
array.
SNS Endpoint
endpoint
- the ARN of the SNS endpoint to publish to.- Avaialable options:
:client_options
- a hash of options to pass to the SNS::Topic during creation. If your ENV is setup with all required AWS keys you won't need to set anything here.:publish_options
- a hash of options to pass into the publish command for the SNS::Topic. The message key will always be overwritten. You can also provide a proc, which will be passed the message per transmission and is expected to return a hash with options. A proc can be used to set message_attributes that rely on variables from the message body.
SQS Endpoint
endpoint
- the URL of the SQS queue to poll.- Avaialable options:
:client_options
- a hash of options to pass to the SNS::Queue during creation. If your ENV is setup with all required AWS keys you won't need to set anything here.:poll_options
- a hash of options to pass into the poll command for the SQS::QueuePoller.:send_options
- a hash of options to pass into the send_message command for the SQS::Queue. The message_body key will always be overwritten. You can also provide a proc, which will be passed the message per transmission and is expected to return a hash with options.:jsonify
- converts the trasnmitted data to json prior to transmission:from_sns
- reads and converts from JSON the serialized Message key
Global Configuration
When configuring HMOTG you can set global configuration opions like so:
HermesMessengerOfTheGods.config do |config|
config. = wassup
end
Allowed configuration options:
Configuration Option | Default | Allowed Values |
---|---|---|
logger | Ruby Logger to STDOUT | A single or array of logger like objects |
quiet | false | Set this to true and all logging will be disabled |
delay_strategy | :smart | This value can be a symbol representing the delay strategy, or a proc which will be expected to do the delaying |
delay_options | {} | A set of default options to use when delaying. The contents of this hash are dependent on the delay_strategy used |
Logging
HermesMessengerOfTheGods::Message
, HermesMessengerOfTheGods::Worker
have
access to a unified set of logging helpers which respect the global logging
configuration. These helpers correspond to the configured Logger::Severity
constant. debug
, info
, warn
, error
, and fatal
are provided by default.
All methods which correspond to a severity level take a single message parameter, or a block which is expected to return the message. With the same behavior as the default logger.
Additionally the say
method is provided which takes the desired level contsant
as the first parameter. say(Logger::ERROR, "foo")
Upgrading
V2 -> V3
Upgrading to V2 to V3 includes the following breaking changes:
1) The worker / message endpoints
setting has been changed to just endpoint
and takes a single endpoint definition.
2) Outputters are no longer used. You need to log anything you want or need.
3) There is no longer instrumentation outputted using ActiveSupport::Notifications
4) Async Dispatches are now removed (Test helper wait_for_async_dispatches
also removed)
5) Partial and Total failure errors are no longer used, the superclass MessageDispatchFailed is used.
6) Dispatch Errors / Successes are no longer included
7) Callback system has been removed
Development
After checking out the repo, run bin/setup
to install dependencies. Then, run rake spec
to run the tests. You can also run bin/console
for an interactive prompt that will allow you to experiment.
To install this gem onto your local machine, run bundle exec rake install
. To release a new version, update the version number in version.rb
, and then run bundle exec rake release
, which will create a git tag for the version, push git commits and tags, and push the .gem
file to rubygems.org.
Contributing
Bug reports and pull requests are welcome on GitHub at https://github.com/[USERNAME]/hermes-messenger-of-the-gods.