Class: Tuber::Connection

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

Overview

Represents a connection to a beanstalkd instance.

Constant Summary collapse

MAX_RETRIES =

Default number of retries to send a command to a connection

3
DEFAULT_RETRY_INTERVAL =

Default retry interval

1
DEFAULT_PORT =

Default port value for beanstalk connection

11300
NON_IDEMPOTENT_COMMANDS =

Commands that must not be retransmitted after a dropped connection. The socket dying between write and readline leaves the first send's fate unknown: a re-sent put can insert a duplicate job, and a re-sent delete/release/bury/touch acts on a job whose reservation died with the old socket — the server answers NOT_FOUND for work that actually succeeded. For these verbs the connection is healed but the original error is re-raised so the caller decides. Everything else (reserve, watch, stats, peek, ...) converges to the same state on a re-send and keeps the transparent retry.

%w[put delete delete-batch release bury touch touch-all kick kick-job].freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(address) ⇒ Connection

Initializes new connection.

Examples:

Tuber::Connection.new('127.0.0.1')
Tuber::Connection.new('127.0.0.1:11300')

ENV['TUBER_URL'] = '127.0.0.1:11300'
@b = Tuber.new
@b.connection.host # => '127.0.0.1'
@b.connection.port # => '11300'

Parameters:

  • address (String) —

    beanstalkd instance address.



64
65
66
67
68
69
70
71
72
73
 
# File 'lib/tuber/connection.rb', line 64

def initialize(address)
  @address = address || _host_from_env || Tuber.configuration.tuber_url
  @mutex = Mutex.new
  @tube_used = 'default'
  @tubes_watched = ['default']

  establish_connection
rescue
  _raise_not_connected!
end

Instance Attribute Details

#address ⇒ String

Returns Beanstalkd server address

Examples:

@conn.address # => "localhost:11300"

Returns:

  • (String) —

    returns Beanstalkd server address



30
31
32
 
# File 'lib/tuber/connection.rb', line 30

def address
  @address
end

#connection ⇒ Object

Returns the value of attribute connection.



30
 
# File 'lib/tuber/connection.rb', line 30

attr_reader :address, :host, :port, :connection

#host ⇒ String

Returns Beanstalkd server host

Examples:

@conn.host # => "localhost"

Returns:

  • (String) —

    returns Beanstalkd server host



30
 
# File 'lib/tuber/connection.rb', line 30

attr_reader :address, :host, :port, :connection

#port ⇒ Integer

Returns Beanstalkd server port

Examples:

@conn.port # => "11300"

Returns:

  • (Integer) —

    returns Beanstalkd server port



30
 
# File 'lib/tuber/connection.rb', line 30

attr_reader :address, :host, :port, :connection

#tube_used ⇒ Object 



36
 
# File 'lib/tuber/connection.rb', line 36

attr_accessor :tubes_watched, :tube_used

#tubes_watched ⇒ Object 



36
37
38
 
# File 'lib/tuber/connection.rb', line 36

def tubes_watched
  @tubes_watched
end

Instance Method Details

#add_to_watched(tube_name) ⇒ Object 



188
189
190
191
 
# File 'lib/tuber/connection.rb', line 188

def add_to_watched(tube_name)
  @tubes_watched << tube_name
  @tubes_watched.uniq
end

#close ⇒ Object

Close connection with beanstalkd server.

Examples:

@conn.close


171
172
173
174
175
176
 
# File 'lib/tuber/connection.rb', line 171

def close
  if @connection
    @connection.close
    @connection = nil
  end
end

#config ⇒ Tuber::Configuration (protected)

Returns configuration options for tuber

Returns:



275
276
277
 
# File 'lib/tuber/connection.rb', line 275

def config
  Tuber.configuration
end

#delete_batch(ids) ⇒ Hash

Deletes a batch of jobs atomically.

Parameters:

  • ids (Array<Integer, String>) —

    Job IDs to delete

Returns:

  • (Hash) —

    with :deleted and :not_found counts



148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
 
# File 'lib/tuber/connection.rb', line 148

def delete_batch(ids)
  _with_retry(retransmit: false) do
    @mutex.synchronize do
      _raise_not_connected! unless connection

      cmd = "delete-batch #{ids.join(' ')}"
      connection.write(cmd + "\r\n")

      res = connection.readline.chomp
      status, deleted, not_found = res.split(/\s/)

      raise UnexpectedResponse.from_status(status, cmd) unless status == "DELETED_BATCH"

      { deleted: deleted.to_i, not_found: not_found.to_i }
    end
  end
end

#establish_connection ⇒ Net::TCPSocket (protected)

Establish a connection based on beanstalk address.

Examples:

establish_connection('localhost:3005')

Returns:

  • (Net::TCPSocket) —

    connection for specified address.

Raises:



206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
 
# File 'lib/tuber/connection.rb', line 206

def establish_connection
  @address = address.first if address.is_a?(Array)
  match = address.split(':')
  @host, @port = match[0], Integer(match[1] || DEFAULT_PORT)

  tcp_opts = { connect_timeout: config.connect_timeout, resolv_timeout: config.resolv_timeout }.compact

  socket = if RUBY_VERSION >= "3.0" && tcp_opts.any?
    TCPSocket.new(@host, @port, **tcp_opts)
  else
    TCPSocket.new(@host, @port)
  end

  begin
    socket.setsockopt(Socket::SOL_SOCKET, Socket::SO_RCVTIMEO, _timeval_for(config.read_timeout)) if config.read_timeout
    socket.setsockopt(Socket::SOL_SOCKET, Socket::SO_SNDTIMEO, _timeval_for(config.write_timeout)) if config.write_timeout
    @connection = socket
  rescue
    socket.close rescue nil
    raise
  end
end

#parse_response(cmd, res) ⇒ Array<Hash{String => String, Number}> (protected)

Parses the response and returns the useful beanstalk response. Will read the body if one is indicated by the status.

Examples:

parse_response("delete 56", "DELETED 56\nFOO")
 # => { :body => "FOO", :status => "DELETED", :id => 56 }

Parameters:

  • cmd (String) —

    Beanstalk command transmitted

  • res (String) —

    Telnet command response

Returns:

  • (Array<Hash{String => String, Number}>) —

    Beanstalk response with status, id, body

Raises:



240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
 
# File 'lib/tuber/connection.rb', line 240

def parse_response(cmd, res)
  status = res.chomp
  body_values = status.split(/\s/)
  status = body_values[0]
  if status == "DRAINING" && cmd.strip.start_with?("drain")
    return { status: status }
  end
  raise UnexpectedResponse.from_status(status, cmd) if UnexpectedResponse::ERROR_STATES.include?(status)
  body = nil
  if status == 'FLUSHED'
    return { status: status, id: body_values[1] }
  end
  if ['OK','FOUND', 'RESERVED'].include?(status)
    bytes_size = body_values[-1].to_i
    raw_body = connection.read(bytes_size)
    body = if status == 'OK'
      psych_v4_valid_body = raw_body.gsub(/^(.*?): (.*)$/) { "#{$1}: #{$2.gsub(/[\:\-\~]/, '_')}" }
      YAML.load(psych_v4_valid_body)
    else
      config.job_parser.call(raw_body)
    end
    crlf = connection.read(2) # \r\n
    raise ExpectedCrlfError.new('EXPECTED_CRLF', cmd) if crlf != "\r\n"
  end
  id = body_values[1]
  response = { :status => status }
  response[:id] = id if id
  response[:body] = body if body
  response[:state] = body_values[2] if status == 'INSERTED' && body_values[2]
  response
end

#remove_from_watched(tube_name) ⇒ Object 



193
194
195
 
# File 'lib/tuber/connection.rb', line 193

def remove_from_watched(tube_name)
  @tubes_watched.delete(tube_name)
end

#reserve_batch(count, timeout = nil) ⇒ Array<Hash>

Reserves a batch of jobs atomically.

Without a +timeout+ (or with +timeout+ of 0) the command is non-blocking: it returns whatever is ready immediately, possibly an empty array. With a positive +timeout+ it long-polls, blocking until the first job arrives (up to +timeout+ seconds) and then draining everything ready, up to +count+.

Parameters:

  • count (Integer) —

    Maximum number of jobs to reserve

  • timeout (Integer) (defaults to: nil) —

    Seconds to long-poll for the first job (nil = non-blocking)

Returns:

  • (Array<Hash>) —

    Array of job hashes with :status, :id, :body keys

Raises:



112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
 
# File 'lib/tuber/connection.rb', line 112

def reserve_batch(count, timeout = nil)
  _with_retry do
    @mutex.synchronize do
      _raise_not_connected! unless connection

      cmd = timeout ? "reserve-batch #{count} #{timeout}" : "reserve-batch #{count}"
      connection.write(cmd + "\r\n")

      header = connection.readline.chomp
      status, actual_count_str = header.split(/\s/, 2)

      raise UnexpectedResponse.from_status(status, cmd) unless status == "RESERVED_BATCH"

      actual_count = actual_count_str.to_i
      jobs = []
      actual_count.times do
        line = connection.readline.chomp
        _, job_id, bytes_str = line.split(/\s/)
        bytes = bytes_str.to_i
        body = connection.read(bytes)
        crlf = connection.read(2)
        raise ExpectedCrlfError.new("EXPECTED_CRLF", cmd) unless crlf == "\r\n"

        body = config.job_parser.call(body)
        jobs << { status: "RESERVED", id: job_id, body: body }
      end
      jobs
    end
  end
end

#to_s ⇒ Object Also known as: inspect

Returns string representation of job.

Examples:

@conn.inspect


183
184
185
 
# File 'lib/tuber/connection.rb', line 183

def to_s
  "#<Tuber::Connection host=#{host.inspect} port=#{port.inspect}>"
end

#transmit(command, **options) ⇒ Array<Hash{String => String, Number}>

Send commands to beanstalkd server via connection.

Examples:

@conn = Tuber::Connection.new
@conn.transmit('bury 123')
@conn.transmit('stats')

Parameters:

  • ] (Hash{String => String, Number}) —

    options Retained for compatibility

  • command (String) —

    Beanstalkd command

Returns:

  • (Array<Hash{String => String, Number}>) —

    Beanstalkd command response



85
86
87
88
89
90
91
92
93
94
95
96
97
98
 
# File 'lib/tuber/connection.rb', line 85

def transmit(command, **options)
  verb = command.to_s[/\A\S+/]
  retransmit = !NON_IDEMPOTENT_COMMANDS.include?(verb)
  _with_retry(retransmit: retransmit, **options.slice(:retry_interval, :init)) do
    @mutex.synchronize do
      _raise_not_connected! unless connection

      command = command.dup.force_encoding('ASCII-8BIT') if command.respond_to?(:force_encoding)
      connection.write(command.to_s + "\r\n")
      res = connection.readline
      parse_response(command, res)
    end
  end
end