Class: BlockGiven::Watcher

Inherits:
Object
  • Object
show all
Defined in:
lib/block_given/poller.rb

Overview

Background polling loop running in its own thread. Returned by the watch_* helpers; call #stop (alias #unwatch) to end it.

Every running watcher is registered under a unique id, so they can be listed and stopped even when the object reference was lost:

watcher = client.watch_block_number(id: "blocks") { |n| puts n }
BlockGiven.watchers                     # => [#<BlockGiven::Watcher blocks ...>]
BlockGiven::Watcher.find("blocks").stop
BlockGiven::Watcher.stop_all

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(interval:, name: "watcher", id: nil, logger: nil, on_error: nil, &tick) ⇒ Watcher

Returns a new instance of Watcher.

Raises:

  • (::ArgumentError) 


104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
 
# File 'lib/block_given/poller.rb', line 104

def initialize(interval:, name: "watcher", id: nil, logger: nil, on_error: nil, &tick)
  raise ::ArgumentError, "a block is required" unless tick

  @interval = interval
  @name = name.to_s
  @id = (id || self.class.generate_id(@name)).to_s
  @logger = logger
  @on_error = on_error
  @tick = tick
  @mutex = Mutex.new
  @cond = ConditionVariable.new
  @stopped = false
  @thread = nil
  @ticks = 0
  @started_at = nil
  @last_tick_at = nil
  @last_error = nil
  @last_error_at = nil
end

Instance Attribute Details

#cursor ⇒ Object

Last fully processed position (block number for log watchers), set by the tick.



102
103
104
 
# File 'lib/block_given/poller.rb', line 102

def cursor
  @cursor
end

#id ⇒ Object (readonly)

Returns the value of attribute id.



100
101
102
 
# File 'lib/block_given/poller.rb', line 100

def id
  @id
end

#interval ⇒ Object (readonly)

Returns the value of attribute interval.



100
101
102
 
# File 'lib/block_given/poller.rb', line 100

def interval
  @interval
end

#last_error ⇒ Object (readonly)

Returns the value of attribute last_error.



100
101
102
 
# File 'lib/block_given/poller.rb', line 100

def last_error
  @last_error
end

#last_error_at ⇒ Object (readonly)

Returns the value of attribute last_error_at.



100
101
102
 
# File 'lib/block_given/poller.rb', line 100

def last_error_at
  @last_error_at
end

#last_tick_at ⇒ Object (readonly)

Returns the value of attribute last_tick_at.



100
101
102
 
# File 'lib/block_given/poller.rb', line 100

def last_tick_at
  @last_tick_at
end

#name ⇒ Object (readonly)

Returns the value of attribute name.



100
101
102
 
# File 'lib/block_given/poller.rb', line 100

def name
  @name
end

#started_at ⇒ Object (readonly)

Returns the value of attribute started_at.



100
101
102
 
# File 'lib/block_given/poller.rb', line 100

def started_at
  @started_at
end

#ticks ⇒ Object (readonly)

Returns the value of attribute ticks.



100
101
102
 
# File 'lib/block_given/poller.rb', line 100

def ticks
  @ticks
end

Class Method Details

.all ⇒ Object

Running (or stopping) watchers, oldest first.



50
51
52
 
# File 'lib/block_given/poller.rb', line 50

def all
  @registry_mutex.synchronize { @registry.values.dup }
end

.find(id) ⇒ Object 



54
55
56
 
# File 'lib/block_given/poller.rb', line 54

def find(id)
  @registry_mutex.synchronize { @registry[id.to_s] }
end

.find!(id) ⇒ Object 



58
59
60
 
# File 'lib/block_given/poller.rb', line 58

def find!(id)
  find(id) || raise(InvalidArgumentError, "no running watcher with id #{id.inspect} (running: #{ids.join(', ')})")
end

.generate_id(name) ⇒ Object 



97
 
# File 'lib/block_given/poller.rb', line 97

def generate_id(name) = "#{name.to_s.gsub(/[^a-zA-Z0-9_.:@-]/, '_')}-#{SecureRandom.hex(3)}"

.ids ⇒ Object 



62
 
# File 'lib/block_given/poller.rb', line 62

def ids = all.map(&:id)

.kill(id) ⇒ Object 



70
 
# File 'lib/block_given/poller.rb', line 70

def kill(id) = find(id)&.kill

.kill_all ⇒ Object 



79
 
# File 'lib/block_given/poller.rb', line 79

def kill_all = all.each(&:kill)

.register(watcher) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



82
83
84
85
86
87
88
89
90
 
# File 'lib/block_given/poller.rb', line 82

def register(watcher)
  @registry_mutex.synchronize do
    if (existing = @registry[watcher.id]) && !existing.equal?(watcher)
      raise InvalidArgumentError, "a watcher with id #{watcher.id.inspect} is already running"
    end

    @registry[watcher.id] = watcher
  end
end

.stop(id, join: nil) ⇒ Object

Graceful stop by id. Returns the watcher, or nil if unknown.



65
66
67
68
 
# File 'lib/block_given/poller.rb', line 65

def stop(id, join: nil)
  watcher = find(id)
  watcher&.stop&.join(join)
end

.stop_all(join: nil) ⇒ Object 



72
73
74
75
76
77
 
# File 'lib/block_given/poller.rb', line 72

def stop_all(join: nil)
  watchers = all
  watchers.each(&:stop)
  watchers.each { |w| w.join(join) }
  watchers
end

.unregister(watcher) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



93
94
95
 
# File 'lib/block_given/poller.rb', line 93

def unregister(watcher)
  @registry_mutex.synchronize { @registry.delete(watcher.id) if @registry[watcher.id].equal?(watcher) }
end

Instance Method Details

#inspect ⇒ Object 



187
188
189
190
 
# File 'lib/block_given/poller.rb', line 187

def inspect
  error = last_error ? " last_error=#{last_error.class}" : ""
  "#<BlockGiven::Watcher #{id} #{name} #{status} cursor=#{cursor.inspect} ticks=#{ticks}#{error}>"
end

#join(timeout = nil) ⇒ Object 



165
166
167
168
 
# File 'lib/block_given/poller.rb', line 165

def join(timeout = nil)
  @thread&.join(timeout)
  self
end

#kill ⇒ Object

Forceful: kills the thread even in the middle of a tick (use when stop does not return).



147
148
149
150
151
152
 
# File 'lib/block_given/poller.rb', line 147

def kill
  stop
  @thread&.kill
  self.class.unregister(self)
  self
end

#on_error(&handler) ⇒ Object

Replace the error handler. Without a handler errors are logged and polling continues.



171
172
173
174
 
# File 'lib/block_given/poller.rb', line 171

def on_error(&handler)
  @on_error = handler
  self
end

#running? ⇒ Boolean

Returns:

  • (Boolean) 


154
 
# File 'lib/block_given/poller.rb', line 154

def running? = !!@thread&.alive?

#start ⇒ Object 



124
125
126
127
128
129
130
131
132
133
134
 
# File 'lib/block_given/poller.rb', line 124

def start
  return self if running?

  self.class.register(self)
  @stopped = false
  @started_at = Time.now
  @thread = Thread.new { run }
  @thread.name = "block_given:#{id}"
  @thread.report_on_exception = false
  self
end

#status ⇒ Object 



157
158
159
160
161
162
163
 
# File 'lib/block_given/poller.rb', line 157

def status
  return :idle if @thread.nil?
  return :running if running? && !stopped?
  return :stopping if running?

  :stopped
end

#stop ⇒ Object Also known as: unwatch

Graceful: the current tick finishes, then the thread exits.



137
138
139
140
141
142
143
 
# File 'lib/block_given/poller.rb', line 137

def stop
  @mutex.synchronize do
    @stopped = true
    @cond.broadcast
  end
  self
end

#stopped? ⇒ Boolean

Returns:

  • (Boolean) 


155
 
# File 'lib/block_given/poller.rb', line 155

def stopped? = @stopped

#to_h ⇒ Object 



178
179
180
181
182
183
184
185
 
# File 'lib/block_given/poller.rb', line 178

def to_h
  {
    id: id, name: name, status: status, interval: interval, cursor: cursor, ticks: ticks,
    started_at: started_at, last_tick_at: last_tick_at,
    last_error: last_error && "#{last_error.class}: #{last_error.message}", last_error_at: last_error_at,
    thread: @thread&.name
  }
end

#uptime ⇒ Object 



176
 
# File 'lib/block_given/poller.rb', line 176

def uptime = started_at ? Time.now - started_at : 0