-
Notifications
You must be signed in to change notification settings - Fork 2
/
alephant.rb
96 lines (78 loc) · 1.94 KB
/
alephant.rb
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
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
94
95
96
$: << File.dirname(__FILE__)
require 'aws-sdk'
require 'json'
require_relative 'env'
require 'alephant/models/logger'
require 'alephant/models/queue'
require 'alephant/models/cache'
require 'alephant/models/renderer'
require 'alephant/models/sequencer'
require 'alephant/errors'
require 'alephant/views'
module Alephant
class Alephant
attr_reader :sequencer, :queue, :cache, :renderer
VALID_OPTS = [
:s3_bucket_id,
:s3_object_path,
:s3_object_id,
:table_name,
:sqs_queue_id,
:view_id,
:view_path,
:sequential_proc,
:set_last_seen_proc
]
def initialize(opts = {}, logger = nil)
set_logger(logger)
set_opts(opts)
@logger = ::Alephant.logger
@sequencer = Sequencer.new(
{
:table_name => @table_name
},
@sqs_queue_id
)
@queue = Queue.new(@sqs_queue_id)
@cache = Cache.new(@s3_bucket_id, @s3_object_path)
@renderer = Renderer.new(@view_id, @view_path)
end
def set_logger(logger)
::Alephant.logger = logger
end
def parse(msg)
JSON.parse(msg)
end
def write(data)
@cache.put(
@s3_object_id,
@renderer.render(data)
)
end
def receive(msg)
data = parse(msg.body)
@logger.info("Alephant.receive: with id #{msg.id} and body digest: #{msg.md5}")
if @sequencer.sequential?(data, &@sequential_proc)
write data
@sequencer.set_last_seen(data, &@set_last_seen_proc)
else
@logger.warn("Alephant.receive: out of sequence message received #{msg.id} (discarded)")
end
end
def run!
Thread.new do
@queue.poll { |msg| receive(msg) }
end
end
private
def set_opts(opts)
VALID_OPTS.each do | k |
v = opts.has_key?(k) ? opts[k] : nil
singleton_class.class_eval do
attr_accessor k
end
send("#{k}=", v)
end
end
end
end