Cloudenvoy

gem 'cloudenvoy'

# 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

# 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

# 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

  # 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!'])

# 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

DummyPublisher.setup
DummySubscriber.setup

# 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' }]
])

# 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

# 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

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

# 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

# config/initializers/cloudenvoy.rb

Cloudenvoy.configure do |config|
  config.logger = SemanticLogger[Cloudenvoy]
end

# 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

# 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) }

# config/initializers/cloudenvoy.rb

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

# 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

# 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

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!

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

# 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

# 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

# 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')

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

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