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