Class: Apollo::Agent::FetcherAgent

Inherits:
BaseAgent
  • Object
show all
Defined in:
lib/apollo_crawler/agent/fetcher_agent.rb

Constant Summary collapse

THREAD_POOL_SIZE = 
1

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods inherited from BaseAgent

#run

Constructor Details

#initialize(amqp, opts = {}) ⇒ FetcherAgent

Returns a new instance of FetcherAgent.



41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
 
# File 'lib/apollo_crawler/agent/fetcher_agent.rb', line 41

def initialize(amqp, opts={})
  self.fetcher = Apollo::Fetcher::SmartFetcher.new

  if(opts[:verbose])
    puts "Initializing fetcher agent..."
  end

  # Declarations
  channel = amqp.create_channel
  channel.prefetch(THREAD_POOL_SIZE)
  
  # Binding (Default)
  self.declarations = Apollo::Agent.declare_entities(channel, opts)
  queue = declarations[:queues]["fetcher.queue"]

  # AMQP contexts for threads
  contexts = []
  (0...THREAD_POOL_SIZE).each do |i|
    puts "FetcherAgent::initialize() - Creating context #{i}" if opts[:verbose]
  end

  # AMQP contexts mutex/lock
  self.mutex = Mutex.new()

  exchange = self.declarations[:exchanges]["fetcher"]

  queue.bind(exchange).subscribe(:ack => true) do |delivery_info, metadata, payload|   
    # There can be troubles with concurency, please see https://groups.google.com/forum/?fromgroups=#!topic/ruby-amqp/aO9GPu-jxuE
    queued_url = JSON.parse(payload)
    url = queued_url["url"]

    puts "FetcherAgent: Received - '#{url}', metadata #{metadata.inspect}" if opts[:verbose]
    self.mutex.synchronize {
      puts "FetcherAgent: Acking - '#{delivery_info.delivery_tag}'" if opts[:verbose]
      channel.basic_ack(delivery_info.delivery_tag, true)
    }

    begin
      doc = Apollo::Fetcher::SmartFetcher::fetch(url)
      doc = process_fetched_doc(queued_url, doc, metadata, opts)
      
      if(metadata && metadata[:reply_to])
        puts "Replying to '#{metadata[:reply_to]}'"
        send_response_msg(metadata[:reply_to], queued_url, doc)
      end

    rescue Exception => e
      puts "EXCEPTION: FetcherAgent::initialize() - Unable to fetch '#{url}', reason: '#{e.to_s}'"
    end

    doc
  end
end

Instance Attribute Details

#declarations ⇒ Object

Returns the value of attribute declarations.



38
39
40
 
# File 'lib/apollo_crawler/agent/fetcher_agent.rb', line 38

def declarations
  @declarations
end

#fetcher ⇒ Object

Returns the value of attribute fetcher.



37
38
39
 
# File 'lib/apollo_crawler/agent/fetcher_agent.rb', line 37

def fetcher
  @fetcher
end

#mutex ⇒ Object

Returns the value of attribute mutex.



39
40
41
 
# File 'lib/apollo_crawler/agent/fetcher_agent.rb', line 39

def mutex
  @mutex
end

Instance Method Details

#format_response_msg(queued_url, doc) ⇒ Object 



108
109
110
111
112
113
 
# File 'lib/apollo_crawler/agent/fetcher_agent.rb', line 108

def format_response_msg(queued_url, doc)
  return {
    :request => queued_url,
    :response => doc
  }
end

#process_fetched_doc(queued_url, doc, metadata, opts = {}) ⇒ Object 



95
96
97
98
99
100
101
102
103
104
105
106
 
# File 'lib/apollo_crawler/agent/fetcher_agent.rb', line 95

def process_fetched_doc(queued_url, doc, metadata, opts={})
  url = queued_url["url"]

  res = Apollo::Model::RawDocument.new
  res.headers = doc.headers
  res.body = doc.body
  res.sha_hash = Digest::SHA1.hexdigest(doc.body)
  res.status = doc.status
  res.url = url

  return res
end

#send_response_msg(dest, queued_url, doc) ⇒ Object 



115
116
117
118
119
120
121
122
123
124
 
# File 'lib/apollo_crawler/agent/fetcher_agent.rb', line 115

def send_response_msg(dest, queued_url, doc)
  if(dest != nil)              
    msg = format_response_msg(queued_url, doc)

    self.mutex.synchronize {
      exchange = self.declarations[:exchanges][dest]
      exchange.publish(msg.to_json)
    }
  end
end