Class: C3s::Republisher

Inherits:
Component
  • Object
show all
Defined in:
lib/republisher.rb

Instance Attribute Summary

Attributes inherited from Component

#config, #model

Instance Method Summary collapse

Methods inherited from Component

#add_callbacks, #handle_disco_info, #handle_set, #initialize, #publish_to_pubsub, #send!, #send_error, #send_ok

Constructor Details

This class inherits a constructor from C3s::Component

Instance Method Details

#build_object(klass, item) ⇒ Object

Builds an object information based on pubsub item content. Each field of the published item is maped to an object attribute.

klass
Class

the object class

item
REXML::Element

the published item



90
91
92
93
94
95
96
97
98
99
 
# File 'lib/republisher.rb', line 90

def build_object(klass, item)
  return nil if item.nil?
  @object = klass.new
  klass.attributes.each do |att|
    @object.send("#{att}=", item.first_element(att).text)
  end
  @object
rescue Exception => e
  $LOG.error "Error on build_object(): #{e.inspect} #{e.backtrace.join("\n")}"
end

#connect ⇒ Object

When connects our republisher must subscribe nodes from which will receive context.



18
19
20
21
22
23
 
# File 'lib/republisher.rb', line 18

def connect
  super
  if config['subscribe_nodes']
    subscribe_nodes()
  end
end

#handle_iq(iq) ⇒ Object

Block the set iqs because this provider doesn’t receive iq’s directly to set the information.



28
29
30
31
32
33
34
35
36
 
# File 'lib/republisher.rb', line 28

def handle_iq(iq)
  if iq.type == :set
    puts "WARNING: IQ of type 'set' received from #{iq.from}!"
    $LOG.warn("IQ of type 'set' received from #{iq.from}")
    send_error(iq, ['unexpected-request', nil])
    return
  end
  super
end

#handle_items_publication(c3snode, items) ⇒ Object

Handles the publicated items and the republishing on new node(s)



67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
 
# File 'lib/republisher.rb', line 67

def handle_items_publication(c3snode, items)
  items.each_element("//item/") do |item|
    begin
      c3snode.provider = config['name'] # TODO - assuming that provider_name.eql?(pubsub_root_nodes)
      t = Thread.new do
        # This must be implemented on the republisher
        # because the strategy may vary from component to component.
        republish(c3snode, item)
      end
      t.join
    rescue Exception => e
      $LOG.error "Error on published item: #{e}"
    end
  end
rescue Exception => e
  $LOG.error "Error on handle_msg(): #{e.inspect} #{e.backtrace.join("\n")}"
end

#handle_msg(msg) ⇒ Object

Handles messages. I only deal with pubsub messages (publication notify) and pubsub events (most likely).

msg

The message received



55
56
57
58
59
60
61
62
63
 
# File 'lib/republisher.rb', line 55

def handle_msg(msg)
  event = msg.first_element("event")
  return if !event.kind_of?(Jabber::PubSub::Event)
  event.each_element("//items/") do |items|
    c3snode = C3s::Node.new(items.attributes['node'])
    next unless c3snode
    handle_items_publication(c3snode, items)
  end
end

#start(subscribing_nodes) ⇒ Object

Starts the component and sets the subscribing nodes.

subscribing_nodes
String or Array

the subscribing node(s)



9
10
11
12
13
 
# File 'lib/republisher.rb', line 9

def start(subscribing_nodes)
  @subscribing_nodes = subscribing_nodes.to_a
  @subscriber = C3s::Subscriber.new(self, config['pubsub'])
  super()
end

#subscribe_nodes ⇒ Object

Subscribes to all nodes



40
41
42
43
44
45
46
47
48
49
 
# File 'lib/republisher.rb', line 40

def subscribe_nodes
  @subscribing_nodes.each do |node|
    begin
      @subscriber.subscribe_collection(node)
    rescue Exception => e
      $LOG.error "Could not subscribe to node #{node} (#{e.inspect})"
      next
    end
  end
end