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 



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

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.



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

def clear_every
  @clear_every
end

#max_backlog_age ⇒ Integer 



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

def max_backlog_age
  @max_backlog_age
end

#max_backlog_size ⇒ Integer 



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

def max_backlog_size
  @max_backlog_size
end

#max_global_backlog_size ⇒ Integer 



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

def max_global_backlog_size
  @max_global_backlog_size
end

#max_in_memory_publish_backlog ⇒ Integer 



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

def max_in_memory_publish_backlog
  @max_in_memory_publish_backlog
end

#subscribed ⇒ Boolean (readonly) 



59
60
61
 
# File 'lib/message_bus/backends/base.rb', line 59

def subscribed
  @subscribed
end

Instance Method Details

#after_fork ⇒ Object

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

Raises:



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

def after_fork
  raise ConcreteClassMustImplementError
end

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

Get messages from a channel backlog

Raises:



135
136
137
 
# File 'lib/message_bus/backends/base.rb', line 135

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

#destroy ⇒ Object

Closes all open connections to the storage.

Raises:



87
88
89
 
# File 'lib/message_bus/backends/base.rb', line 87

def destroy
  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:



93
94
95
 
# File 'lib/message_bus/backends/base.rb', line 93

def expire_all_backlogs!
  raise ConcreteClassMustImplementError
end

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

Get a specific message from a channel

Raises:



154
155
156
 
# File 'lib/message_bus/backends/base.rb', line 154

def get_message(channel, message_id)
  raise ConcreteClassMustImplementError
end

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

Get messages from the global backlog

Raises:



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

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:



188
189
190
 
# File 'lib/message_bus/backends/base.rb', line 188

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:



174
175
176
 
# File 'lib/message_bus/backends/base.rb', line 174

def global_unsubscribe
  raise ConcreteClassMustImplementError
end

#last_id(channel) ⇒ Integer

Get the ID of the last message published on a channel

Raises:



116
117
118
 
# File 'lib/message_bus/backends/base.rb', line 116

def last_id(channel)
  raise ConcreteClassMustImplementError
end

#last_ids(*channels) ⇒ Array<Integer>

Get the ID of the last message published on multiple channels

Raises:



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

def last_ids(*channels)
  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 being 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:



107
108
109
 
# File 'lib/message_bus/backends/base.rb', line 107

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

#reset! ⇒ Object

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

Raises:



82
83
84
 
# File 'lib/message_bus/backends/base.rb', line 82

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:



169
170
171
 
# File 'lib/message_bus/backends/base.rb', line 169

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