Module: PGMQ::Client::Topics

Included in:
PGMQ::Client
Defined in:
lib/pgmq/client/topics.rb

Overview

Note:

Requires PGMQ v1.11.0+

Topic routing operations (AMQP-like patterns)

This module provides AMQP-style topic routing for PGMQ, allowing messages to be routed to multiple queues based on pattern matching.

Topic patterns support wildcards:

  • ‘*` matches exactly one word between dots (e.g., `orders.*` matches `orders.new`)

  • ‘#` matches zero or more words (e.g., `orders.#` matches `orders.new.urgent`)

Instance Method Summary collapse

Instance Method Details

#bind_topic(pattern, queue_name) ⇒ void

This method returns an undefined value.

Binds a topic pattern to a queue

Messages sent with routing keys matching this pattern will be delivered to the specified queue.

Examples:

Bind exact routing key

client.bind_topic("orders.new", "new_orders")

Bind with single-word wildcard

client.bind_topic("orders.*", "all_order_events")

Bind with multi-word wildcard

client.bind_topic("orders.#", "order_audit_log")

Parameters:

  • pattern (String) —

    topic pattern with optional wildcards (* or #)

  • queue_name (String) —

    name of the queue to bind



32
33
34
35
36
37
38
39
40
41
42
43
 
# File 'lib/pgmq/client/topics.rb', line 32

def bind_topic(pattern, queue_name)
  validate_queue_name!(queue_name)

  with_connection do |conn|
    conn.exec_params(
      "SELECT pgmq.bind_topic($1::text, $2::text)",
      [pattern, queue_name]
    )
  end

  nil
end

#list_topic_bindings(queue_name: nil) ⇒ Array<Hash>

Lists all topic bindings

Examples:

List all bindings

bindings = client.list_topic_bindings
# => [{ pattern: "orders.*", queue_name: "orders", bound_at: "..." }, ...]

List bindings for specific queue

bindings = client.list_topic_bindings(queue_name: "orders")

Parameters:

  • queue_name (String, nil) (defaults to: nil) —

    optional queue name to filter by

Returns:

  • (Array<Hash>) —

    array of binding hashes with pattern, queue_name, bound_at



183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
 
# File 'lib/pgmq/client/topics.rb', line 183

def list_topic_bindings(queue_name: nil)
  result = with_connection do |conn|
    if queue_name
      validate_queue_name!(queue_name)
      conn.exec_params(
        "SELECT pattern, queue_name, bound_at FROM pgmq.list_topic_bindings($1::text)",
        [queue_name]
      )
    else
      conn.exec("SELECT pattern, queue_name, bound_at FROM pgmq.list_topic_bindings()")
    end
  end

  result.map do |row|
    {
      pattern: row["pattern"],
      queue_name: row["queue_name"],
      bound_at: row["bound_at"]
    }
  end
end

#produce_batch_topic(routing_key, messages, headers: nil, delay: 0) ⇒ Array<Hash>

Sends multiple messages via topic routing

All messages will be delivered to all queues whose bound patterns match the routing key.

Examples:

Batch topic send

results = client.produce_batch_topic("orders.new", [
  '{"order_id":1}',
  '{"order_id":2}'
])
# => [{ queue_name: "new_orders", msg_id: "1" }, ...]

Parameters:

  • routing_key (String) —

    dot-separated routing key

  • messages (Array<String>) —

    array of message payloads as JSON strings

  • headers (Array<String>, nil) (defaults to: nil) —

    optional array of headers as JSON strings

  • delay (Numeric, Time) (defaults to: 0) —

    delay in seconds before messages become visible (integer or float), or an absolute Time at which the messages become visible, including ActiveSupport::TimeWithZone (PGMQ v1.10.0+)

Returns:

  • (Array<Hash>) —

    array of hashes with :queue_name and :msg_id



125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
 
# File 'lib/pgmq/client/topics.rb', line 125

def produce_batch_topic(routing_key, messages, headers: nil, delay: 0)
  return [] if messages.empty?

  if headers && headers.length != messages.length
    raise ArgumentError,
      "headers array length (#{headers.length}) must match messages array length (#{messages.length})"
  end

  result = with_connection do |conn|
    encoder = PG::TextEncoder::Array.new
    encoded_messages = encoder.encode(messages)

    if headers && !delay.is_a?(Numeric)
      encoded_headers = encoder.encode(headers)
      conn.exec_params(
        "SELECT * FROM pgmq.send_batch_topic($1::text, $2::jsonb[], $3::jsonb[], $4::timestamptz)",
        [routing_key, encoded_messages, encoded_headers, delay.to_time.utc.iso8601(6)]
      )
    elsif headers
      encoded_headers = encoder.encode(headers)
      conn.exec_params(
        "SELECT * FROM pgmq.send_batch_topic($1::text, $2::jsonb[], $3::jsonb[], $4::integer)",
        [routing_key, encoded_messages, encoded_headers, delay]
      )
    elsif !delay.is_a?(Numeric)
      conn.exec_params(
        "SELECT * FROM pgmq.send_batch_topic($1::text, $2::jsonb[], $3::timestamptz)",
        [routing_key, encoded_messages, delay.to_time.utc.iso8601(6)]
      )
    elsif delay > 0
      conn.exec_params(
        "SELECT * FROM pgmq.send_batch_topic($1::text, $2::jsonb[], $3::integer)",
        [routing_key, encoded_messages, delay]
      )
    else
      conn.exec_params(
        "SELECT * FROM pgmq.send_batch_topic($1::text, $2::jsonb[])",
        [routing_key, encoded_messages]
      )
    end
  end

  result.map do |row|
    { queue_name: row["queue_name"], msg_id: row["msg_id"] }
  end
end

#produce_topic(routing_key, message, headers: nil, delay: 0) ⇒ Integer

Sends a message via topic routing

The message will be delivered to all queues whose bound patterns match the routing key.

Examples:

Basic topic send

count = client.produce_topic("orders.new", '{"order_id":123}')

With headers and delay

count = client.produce_topic("orders.new.priority",
  '{"order_id":123}',
  headers: '{"trace_id":"abc"}',
  delay: 30)

Parameters:

  • routing_key (String) —

    dot-separated routing key (e.g., “orders.new.priority”)

  • message (String) —

    message as JSON string

  • headers (String, nil) (defaults to: nil) —

    optional headers as JSON string

  • delay (Integer) (defaults to: 0) —

    delay in seconds before message becomes visible

Returns:

  • (Integer) —

    count of queues the message was delivered to



84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
 
# File 'lib/pgmq/client/topics.rb', line 84

def produce_topic(routing_key, message, headers: nil, delay: 0)
  result = with_connection do |conn|
    if headers
      conn.exec_params(
        "SELECT pgmq.send_topic($1::text, $2::jsonb, $3::jsonb, $4::integer)",
        [routing_key, message, headers, delay]
      )
    elsif delay > 0
      conn.exec_params(
        "SELECT pgmq.send_topic($1::text, $2::jsonb, $3::integer)",
        [routing_key, message, delay]
      )
    else
      conn.exec_params(
        "SELECT pgmq.send_topic($1::text, $2::jsonb)",
        [routing_key, message]
      )
    end
  end

  result[0]["send_topic"].to_i
end

#test_routing(routing_key) ⇒ Array<Hash>

Tests which queues a routing key would match

Useful for debugging topic routing configurations.

Examples:

Test routing

matches = client.test_routing("orders.new.priority")
# => [{ pattern: "orders.*", queue_name: "new_orders" }, ...]

Parameters:

  • routing_key (String) —

    routing key to test

Returns:

  • (Array<Hash>) —

    array of matched bindings with pattern and queue_name



215
216
217
218
219
220
221
222
223
224
225
226
 
# File 'lib/pgmq/client/topics.rb', line 215

def test_routing(routing_key)
  result = with_connection do |conn|
    conn.exec_params(
      "SELECT pattern, queue_name FROM pgmq.test_routing($1::text)",
      [routing_key]
    )
  end

  result.map do |row|
    { pattern: row["pattern"], queue_name: row["queue_name"] }
  end
end

#unbind_topic(pattern, queue_name) ⇒ Boolean

Unbinds a topic pattern from a queue

Examples:

client.unbind_topic("orders.new", "new_orders")

Parameters:

  • pattern (String) —

    topic pattern to unbind

  • queue_name (String) —

    name of the queue to unbind from

Returns:

  • (Boolean) —

    true if the binding was removed, false if it didn’t exist



53
54
55
56
57
58
59
60
61
62
63
64
 
# File 'lib/pgmq/client/topics.rb', line 53

def unbind_topic(pattern, queue_name)
  validate_queue_name!(queue_name)

  result = with_connection do |conn|
    conn.exec_params(
      "SELECT pgmq.unbind_topic($1::text, $2::text)",
      [pattern, queue_name]
    )
  end

  result[0]["unbind_topic"] == "t"
end

#validate_routing_key(routing_key) ⇒ Boolean

Validates a routing key

Routing keys are dot-separated words (no wildcards allowed). Returns false for invalid routing keys (PGMQ raises an error for invalid keys).

Examples:

client.validate_routing_key("orders.new.priority")  # => true
client.validate_routing_key("orders.*")             # => false (wildcards not allowed)

Parameters:

  • routing_key (String) —

    routing key to validate

Returns:

  • (Boolean) —

    true if valid, false if invalid



239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
 
# File 'lib/pgmq/client/topics.rb', line 239

def validate_routing_key(routing_key)
  result = with_connection do |conn|
    conn.exec_params(
      "SELECT pgmq.validate_routing_key($1::text)",
      [routing_key]
    )
  end

  result[0]["validate_routing_key"] == "t"
rescue PGMQ::Errors::ConnectionError => e
  # PGMQ raises an error for invalid routing keys
  return false if e.message.include?("invalid characters")

  raise
end

#validate_topic_pattern(pattern) ⇒ Boolean

Validates a topic pattern

Topic patterns can include wildcards: * (single word) or # (zero or more words).

Examples:

client.validate_topic_pattern("orders.*")    # => true
client.validate_topic_pattern("orders.#")    # => true
client.validate_topic_pattern("orders.new")  # => true

Parameters:

  • pattern (String) —

    topic pattern to validate

Returns:

  • (Boolean) —

    true if valid



266
267
268
269
270
271
272
273
274
275
 
# File 'lib/pgmq/client/topics.rb', line 266

def validate_topic_pattern(pattern)
  result = with_connection do |conn|
    conn.exec_params(
      "SELECT pgmq.validate_topic_pattern($1::text)",
      [pattern]
    )
  end

  result[0]["validate_topic_pattern"] == "t"
end