Class: MessageBus::Backends::Base Abstract

Inherits:
Object
  • Object
show all
Defined in:
lib/message_bus/backends/base.rb

Overview

This class is abstract.

Backends provide a consistent API over a variety of options for persisting published messages. The API they present is around the publication to and reading of messages from those backlogs in a manner consistent with message_bus’ philosophy.

The heart of the message bus, a backend acts as two things:

  1. A channel multiplexer

  2. Backlog storage per-multiplexed channel.

Backends manage and expose multiple backlogs:

  • A backlog for each channel, in which messages that were published to that channel are stored.

  • A global backlog, which conceptually stores all published messages, regardless of the channel to which they were published.

Backlog storage mechanisms and schemas are up to each individual backend implementation, and some backends store messages very differently than others. It is not necessary in order to be considered a valid backend, to, for example, store each channel backlog as a separate collection. As long as the API is presented per this documentation, the backend is free to make its own storage and performance optimisations.

The concept of a per-channel backlog permits for lookups of messages in a manner that is optimised for the use case of a subscriber catching up from a message pointer, while a global backlog allows for optimising the case where another system subscribes to the firehose of messages, for example a message_bus server receiving all publications for delivery to subscribed clients.

Backends are fully responsible for maintaining their storage, including any pruning or expiration of that storage that is necessary. message_bus allows for several options for limiting the required storage capacity by either backlog size or the TTL of messages in a backlog. Backends take these settings and effect them either forcibly or by delegating to their storage mechanism.

Message which are published to message_bus have two IDs; one which they are known by in the channel-specific backlog that they are published to, and another (the “global ID”) which is unique across all channels and by which the message can be found in the global backlog. IDs are all sequential integers starting at 0.

Direct Known Subclasses

Memory, Postgres, Redis

Constant Summary collapse

ConcreteClassMustImplementError =

Raised to indicate that the concrete backend implementation does not implement part of the API

Class.new(StandardError)
UNSUB_MESSAGE = 
"$$UNSUBSCRIBE"

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(config = {}, max_backlog_size = 1000) ⇒ Base 



77
 
# File 'lib/message_bus/backends/base.rb', line 77

def initialize(config = {}, max_backlog_size = 1000); end

Instance Attribute Details

#clear_every ⇒ Integer

Typically, backlogs are trimmed whenever we publish to them. This setting allows some tolerance in order to improve performance.



71
72
73
 
# File 'lib/message_bus/backends/base.rb', line 71

def clear_every
  @clear_every
end

#max_backlog_age ⇒ Integer 



68
69
70
 
# File 'lib/message_bus/backends/base.rb', line 68

def max_backlog_age
  @max_backlog_age
end

#max_backlog_size ⇒ Integer 



64
65
66
 
# File 'lib/message_bus/backends/base.rb', line 64

def max_backlog_size
  @max_backlog_size
end

#max_global_backlog_size ⇒ Integer 



66
67
68
 
# File 'lib/message_bus/backends/base.rb', line 66

def max_global_backlog_size
  @max_global_backlog_size
end

#max_in_memory_publish_backlog ⇒ Integer 



73
74
75
 
# File 'lib/message_bus/backends/base.rb', line 73

def max_in_memory_publish_backlog
  @max_in_memory_publish_backlog
end

#subscribed ⇒ Boolean (readonly) 



62
63
64
 
# File 'lib/message_bus/backends/base.rb', line 62

def subscribed
  @subscribed
end

Instance Method Details

#after_fork ⇒ Object

Performs routines specific to the backend that are necessary after a process fork, typically triggerd by a forking webserver. Typically this re-opens sockets to the backend.

Raises:



80
81
82
 
# File 'lib/message_bus/backends/base.rb', line 80

def after_fork
  raise ConcreteClassMustImplementError
end

#backlog(channel, last_id = 0) ⇒ Array<MessageBus::Message>

Get messages from a channel backlog

Raises:



124
125
126
 
# File 'lib/message_bus/backends/base.rb', line 124

def backlog(channel, last_id = 0)
  raise ConcreteClassMustImplementError
end

#expire_all_backlogs! ⇒ Object

This method is abstract.

Deletes all backlogs and their data. Does not delete non-backlog data that message_bus may persist, depending on the concrete backend implementation. Use with extreme caution.

Raises:



91
92
93
 
# File 'lib/message_bus/backends/base.rb', line 91

def expire_all_backlogs!
  raise ConcreteClassMustImplementError
end

#get_message(channel, message_id) ⇒ MessageBus::Message?

Get a specific message from a channel

Raises:



143
144
145
 
# File 'lib/message_bus/backends/base.rb', line 143

def get_message(channel, message_id)
  raise ConcreteClassMustImplementError
end

#global_backlog(last_id = 0) ⇒ Array<MessageBus::Message>

Get messages from the global backlog

Raises:



133
134
135
 
# File 'lib/message_bus/backends/base.rb', line 133

def global_backlog(last_id = 0)
  raise ConcreteClassMustImplementError
end

#global_subscribe(last_id = nil) {|message| ... } ⇒ nil

Subscribe to messages on all channels. Each message since the last ID specified will be delivered by yielding to the passed block as soon as it is available. This will block until subscription is terminated.

Yields:

  • (message) —

    a message-handler block

Yield Parameters:

Raises:



177
178
179
 
# File 'lib/message_bus/backends/base.rb', line 177

def global_subscribe(last_id = nil)
  raise ConcreteClassMustImplementError
end

#global_unsubscribe ⇒ Object

Causes all subscribers to the bus to unsubscribe, and terminates the local connection. Typically used to reset tests.

Raises:



163
164
165
 
# File 'lib/message_bus/backends/base.rb', line 163

def global_unsubscribe
  raise ConcreteClassMustImplementError
end

#last_id(channel) ⇒ Integer

Get the ID of the last message published on a channel

Raises:



114
115
116
 
# File 'lib/message_bus/backends/base.rb', line 114

def last_id(channel)
  raise ConcreteClassMustImplementError
end

#publish(channel, data, opts = nil) ⇒ Integer

Publishes a message to a channel

Options Hash (opts):

  • :queue_in_memory (Boolean) — default: true —

    whether or not to hold the message in an in-memory buffer if publication fails, to be re-tried later

  • :max_backlog_age (Integer) — default: `self.max_backlog_age` —

    the longest amount of time a message may live in a backlog before beging removed, in seconds

  • :max_backlog_size (Integer) — default: `self.max_backlog_size` —

    the largest permitted size (number of messages) for the channel backlog; beyond this capacity, old messages will be dropped

Raises:



105
106
107
 
# File 'lib/message_bus/backends/base.rb', line 105

def publish(channel, data, opts = nil)
  raise ConcreteClassMustImplementError
end

#reset! ⇒ Object

Deletes all message_bus data from the backend. Use with extreme caution.

Raises:



85
86
87
 
# File 'lib/message_bus/backends/base.rb', line 85

def reset!
  raise ConcreteClassMustImplementError
end

#subscribe(channel, last_id = nil) {|message| ... } ⇒ nil

Subscribe to messages on a particular channel. Each message since the last ID specified will be delivered by yielding to the passed block as soon as it is available. This will block until subscription is terminated.

Yields:

  • (message) —

    a message-handler block

Yield Parameters:

Raises:



158
159
160
 
# File 'lib/message_bus/backends/base.rb', line 158

def subscribe(channel, last_id = nil)
  raise ConcreteClassMustImplementError
end