Class: Postburner::Connection

Inherits:
Object
  • Object
show all
Defined in:
lib/postburner/connection.rb

Overview

Wrapper around Beaneater connection with automatic reconnection.

Provides a simplified interface to Beanstalkd via Beaneater, with automatic connection management and reconnection on failures. Each worker thread creates its own connection instance for thread safety.

Examples:

Direct usage

conn = Postburner::Connection.new
conn.tubes['default'].put('job data')

With reconnection

conn = Postburner::Connection.new
conn.reconnect!  # Force fresh connection

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(url = nil) ⇒ Connection

Returns a new instance of Connection.

Parameters:

  • url (String) (defaults to: nil) —

    Beanstalkd URL (e.g., ‘beanstalk://localhost:11300’)



23
24
25
26
27
 
# File 'lib/postburner/connection.rb', line 23

def initialize(url = nil)
  @url = url || Postburner.configuration.beanstalk_url
  @pool = nil
  connect!
end

Instance Attribute Details

#url ⇒ Object (readonly)

Returns the value of attribute url.



19
20
21
 
# File 'lib/postburner/connection.rb', line 19

def url
  @url
end

Instance Method Details

#beanstalk ⇒ Beaneater::Pool

Returns the underlying Beaneater pool.

Examples:

conn.beanstalk.stats

Returns:

  • (Beaneater::Pool) —

    Beaneater connection pool



53
54
55
56
 
# File 'lib/postburner/connection.rb', line 53

def beanstalk
  ensure_connected!
  @pool
end

#clear_tubes!(tube_names = nil) ⇒ Hash

Clears jobs from specified tubes or collects stats for all tubes.

Low-level method that returns data only (no output to stdout). Delegates to Postburner.stats for collecting statistics. For user-facing output, use Postburner.clear_jobs! instead.

SAFETY: Only allows clearing tubes that are defined in the loaded configuration (watched_tube_names) or the scheduler tube. This prevents accidentally clearing tubes from other applications or environments.

Examples:

Collect stats only (no clearing)

result = conn.clear_tubes!
result[:totals][:total]  # => 42

Clear configured tubes only

result = conn.clear_tubes!(Postburner.watched_tube_names)
result[:cleared]  # => true

Invalid tube raises error

conn.clear_tubes!(['random-tube'])
# => ArgumentError: Cannot clear tubes not in configuration

Parameters:

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

    Array of tube names to clear, or nil to only collect stats

Returns:

  • (Hash) —

    Statistics and results with keys:

    • tubes: Array of hashes with per-tube stats

    • totals: Hash with aggregated counts across all tubes

    • cleared: Boolean indicating if tubes were actually cleared

Raises:

  • (ArgumentError) —

    if tube_names contains tubes not in watched_tube_names



123
124
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
 
# File 'lib/postburner/connection.rb', line 123

def clear_tubes!(tube_names = nil)
  ensure_connected!

  # Validate that tubes to clear are in the loaded configuration (or scheduler tube)
  if tube_names&.any?
    allowed = Postburner.watched_tube_names + [Postburner.scheduler_tube_name]
    invalid_tubes = tube_names - allowed

    if invalid_tubes.any?
      raise ArgumentError, <<~ERROR
        Cannot clear tubes not in configuration.
        Invalid tubes: #{invalid_tubes.join(', ')}
        Configured tubes: #{allowed.join(', ')}
      ERROR
    end
  end

  # Get stats using Postburner.stats
  result = Postburner.stats(tube_names)
  result[:cleared] = tube_names&.any? ? true : false

  # Actually clear if tube names were provided and validated
  if tube_names&.any?
    tube_names.each do |tube_name|
      tubes[tube_name].clear
    end
  end

  result
end

#close ⇒ void

This method returns an undefined value.

Closes the Beanstalkd connection.



87
88
89
90
 
# File 'lib/postburner/connection.rb', line 87

def close
  @pool&.close rescue nil
  @pool = nil
end

#connected? ⇒ Boolean

Checks if currently connected to Beanstalkd.

Returns:

  • (Boolean) —

    true if connected, false otherwise



62
63
64
65
66
 
# File 'lib/postburner/connection.rb', line 62

def connected?
  @pool && @pool.respond_to?(:connection) && @pool.connection
rescue
  false
end

#reconnect! ⇒ void

This method returns an undefined value.

Forces reconnection to Beanstalkd.

Closes existing connection (if any) and establishes a fresh one. Use this when connection has gone stale or after network issues.

Examples:

conn.reconnect!


78
79
80
81
 
# File 'lib/postburner/connection.rb', line 78

def reconnect!
  close
  connect!
end

#tubes ⇒ Beaneater::Tubes

Returns the tubes interface for Beanstalkd operations.

Automatically ensures connection is active before returning.

Examples:

conn.tubes['critical'].put('{"job": "data"}')
conn.tubes.watch!('default', 'critical')

Returns:

  • (Beaneater::Tubes) —

    Tubes interface

Raises:

  • (Beaneater::NotConnected) —

    if connection fails



41
42
43
44
 
# File 'lib/postburner/connection.rb', line 41

def tubes
  ensure_connected!
  @pool.tubes
end