Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Browse files

Retries for jabber errors of type 'retry'

could use some configuration in outcall to specify whether we want to retry, for what conditions etc.
  • Loading branch information...
commit fb2ca84d6efd027f64d595e30c7c20064813fc4d 1 parent 84a8b3e
@outerim outerim authored
View
25 lib/pelvis/advertiser.rb
@@ -1,8 +1,9 @@
module Pelvis
class Advertisement
extend Callbacks
+ include Logging
- callbacks :advertised
+ callbacks :completed, :failed
def initialize(advertiser, actor)
@advertiser, @actor = advertiser, actor
@@ -13,7 +14,12 @@ def start
request = agent.request(:direct, "/security/advertise", options_for_advertisement,
:identities => [agent.herault])
request.on_completed do |event|
- advertised
+ logger.debug "Advertisement Successful"
+ completed
+ end
+ request.on_failed do |event|
+ logger.debug "Advertisement Failed"
+ failed
end
end
@@ -34,7 +40,7 @@ class Advertiser
include Logging
extend Callbacks
- callbacks :completed
+ callbacks :succeeded, :failed
def initialize(agent, actors)
@agent, @actors = agent, actors
@@ -48,7 +54,8 @@ def advertise
actors.each do |actor|
a = Advertisement.start(self, actor)
pending_advertisements << a
- a.on_advertised { advertisement_complete(a) }
+ a.on_completed { advertisement_complete(a) }
+ a.on_failed { failed_advertisements << a; advertisement_complete(a) }
end
check_complete
end
@@ -57,6 +64,10 @@ def pending_advertisements
@pending_advertisements ||= []
end
+ def failed_advertisements
+ @failed_advertisements ||= []
+ end
+
def advertisement_complete(which)
pending_advertisements.delete(which)
@finished_advertisements += 1
@@ -67,7 +78,11 @@ def check_complete
return if @completed
if @finished_advertisements >= actors.size
- completed
+ if failed_advertisements.empty?
+ succeeded
+ else
+ failed
+ end
@completed = true
end
end
View
5 lib/pelvis/agent.rb
@@ -92,7 +92,7 @@ def add_actor(klass, &block)
end
def advertise
- unless @protocol.advertise?
+ if !@protocol.advertise? || actors.empty?
logger.debug "Not advertising"
advertised
return
@@ -100,7 +100,8 @@ def advertise
# initial advertisement
a = Advertiser.new(self, actors)
- a.on_completed { advertised }
+ a.on_succeeded { advertised }
+ a.on_failed { raise "unable to perform advertisement" }
end
def evoke(identity, job)
View
31 lib/pelvis/outcall.rb
@@ -7,9 +7,9 @@ class Outcall
def initialize(agent, job)
@agent, @job = agent, job
- @started_at, @finished_at = nil, nil
+ @started_at, @finished_at, @retried_times = nil, nil, 0
end
- attr_reader :agent, :job
+ attr_reader :agent, :job, :retried_times
def start
logger.debug "starting outcall on #{@agent.identity}: #{inspect}"
@@ -72,11 +72,30 @@ def evoke(identity)
check_complete
end
e.on_failed do |error|
- logger.debug "outcall failed: #{identity}: #{error.inspect}"
- evocations[e] = true
- finish
- failed(error)
+ logger.warn "Outcall to #{identity} failed: #{error.inspect}"
+ if can_retry?(error)
+ evocations.delete(e)
+ time = 2 ** (@retried_times + 1)
+ logger.debug "Waiting #{time} seconds before retrying"
+ EM.add_timer(time) { @retried_times += 1; evoke(identity) }
+ else
+ evocations[e] = true
+ finish
+ failed(error)
+ end
+ end
+ end
+
+ def can_retry?(error)
+ if error[:type] == 'wait' and retried_times < max_retries
+ logger.warn "Can Retry #{max_retries - retried_times} times"
+ return true
end
+ false
+ end
+
+ def max_retries
+ 5
end
def check_complete
View
4 lib/pelvis/protocols/xmpp/incall.rb
@@ -14,8 +14,8 @@ def initialize(agent, job)
def start
@agent.send_job_init(job.token, job.scope, job.operation, job.args) do |reply|
- if reply["type"] == 'error'
- failed(:message => "Got jabber error", :data => reply)
+ if reply.error?
+ failed(reply)
else
initialized
end
View
21 lib/pelvis/protocols/xmpp/remote_agent.rb
@@ -1,6 +1,18 @@
module Pelvis
module Protocols
class XMPP
+ class Error < Hash
+ def self.create(attrs)
+ a = new
+ a.update(attrs)
+ a
+ end
+
+ def error?
+ true
+ end
+ end
+
class RemoteAgent
include Logging
@@ -76,7 +88,14 @@ def handle_result(stanza, node, token)
def handle_error(stanza, node, token)
block, original = outbounds.delete(stanza.id)
if block
- block.call(stanza)
+ # TODO: this should send the extra stuff too.
+ # EX:
+ # <error code="404" type="wait">
+ # <recipient-unavailable/>
+ # </error>
+ e = Blather::Stanza::Iq.import(stanza).find_first('error')
+
+ block.call( Error.create(:code => e['code'], :type => e['type'], :message => "Jabber Error") )
else
puts "Ignoring error: #{stanza.inspect}"
end
Please sign in to comment.
Something went wrong with that request. Please try again.