Cloudenvoy

Cross-application messaging framework for GCP Pub/Sub.

Cloudenvoy provides an easy to use interface to GCP Pub/Sub. Using Cloudenvoy you can simplify cross-application event messaging by using a publish/subscribe approach. Pub/Sub is particularly suited for micro-service architectures where a great number of components need to be aware of other components’ activities. In these architectures using point to point communication via API can quickly become messy and hard to maintain due to the number of interconnections to maintain.

Pub/Sub solves that event distribution problem by allowing developers to define topics, publishers and subscribers to distribute and process event messages. Cloudenvoy furthers simplifies the process of setting up Pub/Sub by giving developers an object-oriented way of managing publishers and subscribers.

Cloudenvoy works with the local pub/sub emulator as well, meaning that you can work offline without access to GCP.

Summary

Installation
Get started with Rails
Configuring Cloudenvoy
1. Pub/Sub authentication & permissions
2. Cloudenvoy initializer
Creating topics and subscriptions
1. Sending messages
2. Publisher implementation layout
Receiving messages
Error Handling
Testing

Installation

Add this line to your application’s Gemfile:

gem 'cloudenvoy'

And then execute:

$ bundle install

Or install it yourself as:

$ gem install cloudenvoy

Get started with Rails

Cloudenvoy is pre-integrated with Rails. Follow the steps below to get started.

Install the pub/sub local emulator

gcloud components install pubsub-emulator
gcloud components update

Add the following initializer

# config/initializers/cloudenvoy.rb
Cloudenvoy.configure do |config|
  #
  # GCP Configuration
  #
  config.gcp_project_id = 'some-project'
  config.gcp_sub_prefix = 'my-app'
  #
  # Adapt the server port to be the one used by your Rails web process
  #
  config.processor_host = 'http://localhost:3000'
  #
  # If you do not have any Rails secret_key_base defined, uncomment the following
  # This secret is used to authenticate messages sent to the processing endpoint
  # of your application.
  #
  # config.secret = 'some-long-token'
end

Define a publisher or use generator: rails generate cloudenvoy:publisher Dummy

# app/publishers/dummy_publisher.rb
class DummyPublisher
  include Cloudenvoy::Publisher
  cloudenvoy_options topic: 'test-msgs'
  # Format the message payload. The payload can be a hash
  # or a string.
  def payload(msg)
    {
      type: 'message',
      content: msg
    }
  end
end

Define a subscriber or use generator: rails generate cloudenvoy:subscriber Dummy

# app/subscribers/dummy_subscriber.rb
class HelloSubscriber
  include Cloudenvoy::Subscriber
  cloudenvoy_options topics: ['test-msgs']
  # Do something with the message
  def process(message)
    logger.info("Received message #{message.payload.dig('content')}")
  end
end

Launch the pub/sub emulator:

gcloud beta emulators pubsub start

Use cloudenvoy to setup your topic and subscription

bundle exec rake cloudenvoy:setup

Launch Rails

rails s -p 3000

Open a Rails console and send a message

  # One message at a time
  DummyPublisher.publish('Hello pub/sub')
  # Publish multiple messages in one batch
  # Only available since v0.6.rc1
  DummyPublisher.publish_all(['Hello pub/sub', 'Hello again!', 'Hello again and again!'])

Your Rails logs should display the following:

Started POST "/cloudenvoy/receive?token=1234" for 66.102.6.140 at 2020-09-16 11:12:47 +0200
Processing by Cloudenvoy::SubscriberController#receive as JSON
  Parameters: {"message"=>{"attributes"=>{"kind"=>"hello"}, "data"=>"eyJ0eXBlIjoibWVzc2FnZSIsImNvbnRlbnQiOiJIZWxsbyBmcmllbmQifQ==", "messageId"=>"1501653492745522", "message_id"=>"1501653492745522", "publishTime"=>"2020-09-16T09:12:45.214Z", "publish_time"=>"2020-09-16T09:12:45.214Z"}, "subscription"=>"projects/keypup-dev/subscriptions/my-app.hello_subscriber.test-msgs", "token"=>"eyJhbGciOiJIUzI1NiJ9.eyJpYXQiOjE2MDAyNDc0OTh9.5SbVsDCZLcyoFXseCpPuvE7KY7WXqIQtO6ceoFXcrdw", "subscriber"=>{"message"=>{"attributes"=>{"kind"=>"hello"}, "data"=>"eyJ0eXBlIjoibWVzc2FnZSIsImNvbnRlbnQiOiJIZWxsbyBmcmllbmQifQ==", "messageId"=>"1501653492745522", "message_id"=>"1501653492745522", "publishTime"=>"2020-09-16T09:12:45.214Z", "publish_time"=>"2020-09-16T09:12:45.214Z"}, "subscription"=>"projects/keypup-dev/subscriptions/my-app.hello_subscriber.test-msgs"}}
[Cloudenvoy][HelloSubscriber][1501653492745522] Processing message... -- {:id=>"1501653492745522", :metadata=>{}, :topic=>"test-msgs"}
[Cloudenvoy][HelloSubscriber][1501653492745522] Received message Hello pub/sub -- {:id=>"1501653492745522", :metadata=>{}, :topic=>"test-msgs"}
[Cloudenvoy][HelloSubscriber][1501653492745522] Processing done after 0.001s -- {:id=>"1501653492745522", :metadata=>{}, :topic=>"test-msgs", :duration=>0.001}
Completed 204 No Content in 1ms (ActiveRecord: 0.0ms | Allocations: 500)

Hurray! Your published message was immediately processed by the subscriber.

Configuring Cloudenvoy

Pub/Sub authentication & permissions

The Google Cloud library authenticates via the Google Cloud SDK by default. If you do not have it setup then we recommend you install it.

Other options are available such as using a service account. You can see all authentication options in the Google Cloud Authentication guide.

In order to function properly Cloudenvoy requires the authenticated account to have the following IAM permissions:

pubsub.subscriptions.create
pubsub.subscriptions.get
pubsub.topics.create
pubsub.topics.get
pubsub.topics.publish

To get started quickly you can add the roles/pubsub.admin role to your account via the IAM Console. This is not required if your account is a project admin account.

Cloudenvoy initializer

The gem can be configured through an initializer. See below all the available configuration options.

# config/initializers/cloudenvoy.rb
Cloudenvoy.configure do |config|
  #
  # If you do not have any Rails secret_key_base defined, uncomment the following.
  # This secret is used to authenticate messages sent to the processing endpoint
  # of your application.
  #
  # Default with Rails: Rails.application.credentials.secret_key_base
  #
  # config.secret = 'some-long-token'
  #
  # GCP Configuration
  #
  config.gcp_project_id = 'some-project'
  #
  # Specify the namespace for your subscriptions
  #
  # The gem attempts to keep GCP subscriptions organized by
  # properly namespacing them. Each subscription has the following
  # format:
  # > projects/<gcp_project_id>/subscriptions/<gcp_sub_prefix>.<subscriber_class>.<topic>
  #
  config.gcp_sub_prefix = 'my-app'
  #
  # Specify the publicly accessible host for your application
  #
  # > E.g. in development, using the pub/sub local emulator
  # config.processor_host = 'http://localhost:3000'
  #
  # > E.g. in development, using `config.mode = :production` and ngrok
  # config.processor_host = 'https://111111.ngrok.io'
  #
  config.processor_host = 'https://app.mydomain.com'
  #
  # Specify the mode of operation:
  # - :development => messages will be pushed to the local pub/sub emulator
  # - :production => messages will be pushed to Google Cloud Pub/Sub. Requires a publicly accessible domain.
  #
  # Defaults to :development unless CLOUDENVOY_ENV or RAILS_ENV or RACK_ENV is set to something else.
  #
  # config.mode = Rails.env.production? || Rails.env.my_other_env? ? :production : :development
  #
  # Specify the logger to use
  #
  # Default with Rails: Rails.logger
  # Default without Rails: Logger.new(STDOUT)
  #
  # config.logger = MyLogger.new(STDOUT)
end

Creating topics and subscriptions

Topics and subscriptions can be created using the provided Rake tasks:

# Setup publishers (topics) and subscribers (subscriptions) in one go
bundle exec rake cloudenvoy:setup
# Or set them up individually
bundle exec rake cloudenvoy:setup_publishers
bundle exec rake cloudenvoy:setup_subscribers

For non-rails applications you can run the following in a console to setup your publishers and subscribers:

DummyPublisher.setup
DummySubscriber.setup

Publishing messages

Sending messages

Note: The publish_all method is only available since v0.6.rc1

Cloudenvoy provides a helper method to publish arbitrary messages to any topic.

# Publish a single message
Cloudenvoy.publish('my-topic', { 'some' => 'payload' }, { 'optional' => 'message attribute' })
# Publish multiple messages in one batch
# Only available since v0.6.rc1
Cloudenvoy.publish_all('my-topic', [
  # Message 1
  [{ 'some' => 'msg1 payload' }, { 'optional' => 'msg1 attribute' }],
  # Message 2
  [{ 'some' => 'msg2 payload' }, { 'optional' => 'msg2 attribute' }]
])

This helper is useful for sending basic messages however it is not the preferred way of sending messages as you will quickly clutter your application with message formatting logic over time.

Cloudenvoy provides an object-oriented way of sending messages allowing developers to separate their core business logic from any kind of message formatting logic. These are called Publishers.

The example below shows you how to publish new users to a topic using Cloudenvoy publishers:

# app/publishers/user_publisher.rb
# The publisher is responsible for configuring and formatting
# the pub/sub message.
class UserPublisher
  include Cloudenvoy::Publisher
  cloudenvoy_options topic: 'system-users'
  # Publishers must at least implement the `payload` method,
  # which specifies how the message should be formatted.
  def payload(user)
    {
      id: user.id,
      name: user.name,
      email: user.email
    }
  end
end

Then in your user model you can do the following:

# app/users/user_publisher.rb
class User < ApplicationRecord
  after_create :publish_user
  # Example: publish multiple messages in one batch
  # Only available since v0.6.rc1
  # 
  #
  # Set user status to 'active' without involving callbacks
  # then manually publish the new state of users in one batch
  def self.enable_all_users
    User.update_all(status: 'active', updated_at: Time.current)
    UserPublisher.publish_all(User.all)
  end
  private
  # Example: publish one message at a time
  #
  # Publish users after they have been created
  def publish_user
    UserPublisher.publish(self)
  end
end

Publisher implementation layout

A full publisher implementation looks like this:

class MyPublisher
  include Cloudenvoy::Publisher
  # The topic option defines the default topic messages will be
  # sent to. The publishing topic can be overriden on a per message
  # basis. See the #topic method below.
  cloudenvoy_options topic: 'my-topic'
  # Evaluate the topic at runtime based on publishing arguments.
  # Returning `nil` makes the publisher use the default topic
  # defined via cloudenvoy_options.
  #
  # Note: runtime topics do not get created by the rake tasks. You
  # must create them manually at this stage.
  def topic(arg1, arg2)
    arg1 == 'other' ? 'some-other-topic' : nil
  end
  # Attach pub/sub attributes to the message. Pub/sub attributes
  # can be used for message filtering.
  def metadata(arg1, arg2)
    { reference: "#{arg1}_#{arg2}" }
  end
  # Publishers must at least implement the `payload` method,
  # which specifies how arguments should be transformed into
  # a message payload (Hash or String).
  def payload(arg1, arg2)
    {
      foo: arg1,
      bar: arg2
    }
  end
  # This hook is invoked when the message fails to be formatted and published.
  # If something wrong happens in the methods above, this hook will be triggered.
  def on_error(error)
    logger.error("Oops! Something wrong happened!")
  end
end

Receiving messages

After you have subscribed to a topic, Pub/Sub sends messages to your application via webhook on the /cloudenvoy/receive endpoint. Cloudenvoy then automatically dispatches the message to the right subscriber for processing.

Following up on the previous user publishing example, you might define the following subscriber in another Rails application:

# app/subscribers/user_subscriber.rb
class UserSubscriber
  include Cloudenvoy::Subscriber
  # Subscribers can subscribe to multiple topics
  #
  # You can subscribe to multiple topics:
  # > cloudenvoy_options topics: ['system-users', 'events']
  #
  # You can specify subscription options for each topic
  # by passing a hash (target: v0.2.0)
  #
  # > cloudenvoy_options topics: ['system-users', { name: 'events', retain_acked: true }]
  #
  # See the Pub/Sub documentation of the list of available subscription options: 
  # https://googleapis.dev/ruby/google-cloud-pubsub/latest/Google/Cloud/PubSub/Topic.html#subscribe-instance_method
  #
  cloudenvoy_options topic: 'system-users'
  # Create the user locally if it does not exist already
  #
  # A message has the following attributes:
  #   id: the pub/sub message id
  #   payload: the content of the message (String or Hash)
  #   metadata: the pub/sub message attributes
  #   sub_uri: the pub/sub subscription URI
  #   topic: the topic the message comes from
  #
  def process(message)
    payload = message.payload
    User.create_or_find_by(system_id: payload['id']) do |u|
      u.first_name = payload['name']
      u.email = payload['email']
    end
  end
  # This hook will be invoked if the message processing fails
  def on_error(error)
    logger.error("The following error happened: #{error}")
  end
end

Logging

There are several options available to configure logging and logging context.

Configuring a logger

Cloudenvoy uses Rails.logger if Rails is available and falls back on a plain ruby logger Logger.new(STDOUT) if not.

It is also possible to configure your own logger. For example you can setup Cloudenvoy with semantic_logger by doing the following in your initializer:

# config/initializers/cloudenvoy.rb
Cloudenvoy.configure do |config|
  config.logger = SemanticLogger[Cloudenvoy]
end

Logging context

Cloudenvoy provides publisher/subscriber contextual information to the logger methods.

For example:

# app/subscribers/dummy_subscriber.rb
class DummySubscriber
  include Cloudenvoy::Subscriber
  cloudenvoy_options topics: ['my-topic']
  def process(message)
    logger.info("Subscriber processed with #{message.inspect}. This is working!")
  end
end

Will generate the following log with context {:id=>..., :metadata=>..., :topic=>...}

[Cloudenvoy][DummySubscriber][1501678353930997] Subscriber processed with ###. This is working! -- {:id=>"1501678353930997", :metadata=>{"some"=>"meta"}, :topic=>"my-topic"}

The way contextual information is displayed depends on the logger itself. For example with semantic_logger contextual information might not appear in the log message but show up as payload data on the log entry itself (e.g. using the fluentd adapter).

Contextual information can be customised globally and locally using a log context_processor. By default the loggers are configured this way:

# Publishers
Cloudenvoy::PublisherLogger.log_context_processor = ->(publisher) { publisher.message&.to_h&.slice(:id, :metadata, :topic) || {} }
# Subscribers
Cloudenvoy::SubscriberLogger.log_context_processor = ->(subscriber) { subscriber.message.to_h.slice(:id, :metadata, :topic) }

You can decide to add a global identifier for your publisher logs using the following:

# config/initializers/cloudenvoy.rb
Cloudenvoy::PublisherLogger.log_context_processor = lambda { |publisher|
  publisher.message.to_h.slice(:id, :metadata, :topic).merge(app: 'my-app')
}

You could also decide to log all available context - including the message payload - for specific subscribers only:

# app/subscribers/full_context_subscriber.rb
class FullContextSubscriber
  include Cloudenvoy::Subscriber
  cloudenvoy_options topics: ['my-topic'], log_context_processor: ->(s) { s.message.to_h }
  def process(message)
    logger.info("This log entry will have full context!")
  end
end

See the Cloudenvoy::Publisher, Cloudenvoy::Subscriber and Cloudenvoy::Message for more information on attributes available to be logged in your log_context_processor proc.

Error Handling

Message failures will return an HTTP error to Pub/Sub and trigger a retry at a later time. By default Pub/Sub will retry sending the message until the acknowledgment deadline expires. A number of retries can be explicitly configured by setting up a dead-letter queue.

HTTP Error codes

When Cloudenvoy fails to process a message it returns the following HTTP error code to Pub/Sub, based on the actual reason:

Code	Description
204	The message was processed successfully
404	The message subscriber does not exist.
422	An error occured during the processing of the message (`process` method)

Error callbacks

Publishers and subscribers can implement the on_error(error) callback to do things when a message fails to be published or received:

E.g.

# app/publisher/handle_error_publisher.rb
class HandleErrorPublisher
  include Cloudenvoy::Publisher
  cloudenvoy_options topic: 'my-topic'
  def payload(arg)
    raise(ArgumentError)
  end
  # The runtime error is passed as an argument.
  def on_error(error)
    logger.error("The following error occured: #{error}")
  end
end

Testing

Cloudenvoy provides several options to test your publishers and subscribers.

Test helper setup

Require cloudenvoy/testing in your rails_helper.rb (Rspec Rails) or spec_helper.rb (Rspec) or test unit helper file then enable one of the two modes:

require 'cloudenvoy/testing'
# Mode 1 (default): Push messages to GCP Pub/Sub (env != development)
Cloudenvoy::Testing.enable!
# Mode 2: Push message to in-memory queues. You will be responsible for clearing the
# topic queues using `Cloudenvoy::Testing.clear_all` or `Cloudenvoy::Testing.clear('my-topic')`
Cloudenvoy::Testing.fake!

You can query the current testing mode with:

Cloudenvoy::Testing.enabled?
Cloudenvoy::Testing.fake?

Each testing mode accepts a block argument to temporarily switch to it:

# Enable fake mode for all tests
Cloudenvoy::Testing.fake!
# Enable real mode temporarily for a given test
Cloudenvoy.enable! do
   MyPublisher.publish(1,2)
end

Note that extension middlewares - if any has been registered - run in test mode. You can disable middlewares in your tests by adding the following to your test helper:

# Remove all middlewares
Cloudenvoy.configure do |c|
  c.publisher_middleware.clear
  c.subscriber_middleware.clear
end
# Remove all specific middlewares
Cloudenvoy.configure do |c|
  c.publisher_middleware.remove(MyMiddleware::Publisher)
  c.subscriber_middleware.remove(MyMiddleware::Subscriber)
end

In-memory queues

The fake! modes uses in-memory queues for topics, which can be queried and controlled using the following methods:

# Clear all messages across all topics
Cloudenvoy::Testing.clear_all
# Remove all messages in a given topic
Cloudenvoy::Testing.clear('my-top')
# Get all messages for a given topic
Cloudenvoy::Testing.queue('my-top')

Unit tests

Below are examples of rspec tests. It is assumed that Cloudenvoy::Testing.fake! has been set in the test helper.

Example 1: Testing publishers

describe 'message publishing'
  subject(:publish_message) { MyPublisher.publish(1,2) }
  let(:queue) { Cloudenvoy::Testing.queue('my-topic') }
  it { expect { publish_message }.to change(queue, :size).by(1) }
  it { is_expected.to have_attributes(payload: { 'foo' => 'bar' }) }
end

Example 2: Testing subscribers

describe 'message processing'
  subject { VerifyDataViaApiSubscriber.new(message: message).execute } }
  let(:message) { Cloudenvoy::Message.new(payload: { 'some' => 'payload' }) }
  before { expect(MyApi).to receive(:fetch).and_return([]) }
  it { is_expected.to be_truthy }
end

Development

After checking out the repo, run bin/setup to install dependencies.

For tests, run rake to run the tests. Note that Rails is not in context by default, which means Rails-specific test will not run.
For tests including Rails-specific tests, run bundle exec appraisal rails-7.0 rake
For all context-specific tests (incl. Rails), run the appraisal tests using bundle exec appraisal rake.

You can 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/keypup-io/cloudenvoy. This project is intended to be a safe, welcoming space for collaboration, and contributors are expected to adhere to the code of conduct.

License

The gem is available as open source under the terms of the MIT License.

Code of Conduct

Everyone interacting in the Cloudenvoy project’s codebases, issue trackers, chat rooms and mailing lists is expected to follow the code of conduct.