Class: Beanstalk::Connection

Inherits:
Object
  • Object
show all
Defined in:
lib/beanstalk-client/connection.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(addr, default_tube = nil) ⇒ Connection

Returns a new instance of Connection.



29
30
31
32
33
34
35
36
37
 
# File 'lib/beanstalk-client/connection.rb', line 29

def initialize(addr, default_tube=nil)
  @waiting = false
  @addr = addr
  connect
  @last_used = 'default'
  @watch_list = [@last_used]
  self.use(default_tube) if default_tube
  self.watch(default_tube) if default_tube
end

Instance Attribute Details

#addr ⇒ Object (readonly)

Returns the value of attribute addr.



27
28
29
 
# File 'lib/beanstalk-client/connection.rb', line 27

def addr
  @addr
end

Instance Method Details

#bury(id, pri) ⇒ Object 



104
105
106
107
 
# File 'lib/beanstalk-client/connection.rb', line 104

def bury(id, pri)
  interact("bury #{id} #{pri}\r\n", %w(BURIED))
  :ok
end

#close ⇒ Object 



47
48
49
50
 
# File 'lib/beanstalk-client/connection.rb', line 47

def close
  @socket.close
  @socket = nil
end

#connect ⇒ Object 



39
40
41
42
43
44
45
 
# File 'lib/beanstalk-client/connection.rb', line 39

def connect
  host, port = addr.split(':')
  @socket = TCPSocket.new(host, port.to_i)

  # Don't leak fds when we exec.
  @socket.fcntl(Fcntl::F_SETFD, Fcntl::FD_CLOEXEC)
end

#delete(id) ⇒ Object 



94
95
96
97
 
# File 'lib/beanstalk-client/connection.rb', line 94

def delete(id)
  interact("delete #{id}\r\n", %w(DELETED))
  :ok
end

#ignore(tube) ⇒ Object 



125
126
127
128
129
130
 
# File 'lib/beanstalk-client/connection.rb', line 125

def ignore(tube)
  return @watch_list.size if !@watch_list.include?(tube)
  r = interact("ignore #{tube}\r\n", %w(WATCHING))[0].to_i
  @watch_list -= [tube]
  return r
end

#job_stats(id) ⇒ Object 



136
137
138
 
# File 'lib/beanstalk-client/connection.rb', line 136

def job_stats(id)
  interact("stats-job #{id}\r\n", :yaml)
end

#list_tube_used ⇒ Object 



148
149
150
 
# File 'lib/beanstalk-client/connection.rb', line 148

def list_tube_used()
  interact("list-tube-used\r\n", %w(USING))[0]
end

#list_tubes ⇒ Object 



144
145
146
 
# File 'lib/beanstalk-client/connection.rb', line 144

def list_tubes()
  interact("list-tubes\r\n", :yaml)
end

#list_tubes_watched(cached = false) ⇒ Object 



152
153
154
155
 
# File 'lib/beanstalk-client/connection.rb', line 152

def list_tubes_watched(cached=false)
  return @watch_list if cached
  @watch_list = interact("list-tubes-watched\r\n", :yaml)
end

#peek_buried ⇒ Object 



73
74
75
 
# File 'lib/beanstalk-client/connection.rb', line 73

def peek_buried()
  interact("peek-buried\r\n", :job)
end

#peek_delayed ⇒ Object 



69
70
71
 
# File 'lib/beanstalk-client/connection.rb', line 69

def peek_delayed()
  interact("peek-delayed\r\n", :job)
end

#peek_job(id) ⇒ Object 



61
62
63
 
# File 'lib/beanstalk-client/connection.rb', line 61

def peek_job(id)
  interact("peek #{id}\r\n", :job)
end

#peek_ready ⇒ Object 



65
66
67
 
# File 'lib/beanstalk-client/connection.rb', line 65

def peek_ready()
  interact("peek-ready\r\n", :job)
end

#put(body, pri = 65536, delay = 0, ttr = 120) ⇒ Object 



52
53
54
55
 
# File 'lib/beanstalk-client/connection.rb', line 52

def put(body, pri=65536, delay=0, ttr=120)
  interact("put #{pri} #{delay} #{ttr} #{body.size}\r\n#{body}\r\n",
           %w(INSERTED BURIED))[0].to_i
end

#release(id, pri, delay) ⇒ Object 



99
100
101
102
 
# File 'lib/beanstalk-client/connection.rb', line 99

def release(id, pri, delay)
  interact("release #{id} #{pri} #{delay}\r\n", %w(RELEASED))
  :ok
end

#reserve ⇒ Object

Raises:



77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
 
# File 'lib/beanstalk-client/connection.rb', line 77

def reserve()
  raise WaitingForJobError if @waiting
  @socket.write("reserve\r\n")

  begin
    @waiting = true
    # Give the user a chance to select on multiple fds.
    Beanstalk.select.call([@socket]) if Beanstalk.select
  rescue WaitingForJobError
    # just continue
  ensure
    @waiting = false
  end

  Job.new(self, *read_job('RESERVED'))
end

#stats ⇒ Object 



132
133
134
 
# File 'lib/beanstalk-client/connection.rb', line 132

def stats()
  interact("stats\r\n", :yaml)
end

#stats_tube(tube) ⇒ Object 



140
141
142
 
# File 'lib/beanstalk-client/connection.rb', line 140

def stats_tube(tube)
  interact("stats-tube #{tube}\r\n", :yaml)
end

#use(tube) ⇒ Object 



109
110
111
112
113
114
 
# File 'lib/beanstalk-client/connection.rb', line 109

def use(tube)
  return tube if tube == @last_used
  @last_used = interact("use #{tube}\r\n", %w(USING))[0]
rescue BadFormatError
  raise InvalidTubeName.new(tube)
end

#watch(tube) ⇒ Object 



116
117
118
119
120
121
122
123
 
# File 'lib/beanstalk-client/connection.rb', line 116

def watch(tube)
  return @watch_list.size if @watch_list.include?(tube)
  r = interact("watch #{tube}\r\n", %w(WATCHING))[0].to_i
  @watch_list += [tube]
  return r
rescue BadFormatError
  raise InvalidTubeName.new(tube)
end

#yput(obj, pri = 65536, delay = 0, ttr = 120) ⇒ Object 



57
58
59
 
# File 'lib/beanstalk-client/connection.rb', line 57

def yput(obj, pri=65536, delay=0, ttr=120)
  put(YAML.dump(obj), pri, delay, ttr)
end