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
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
|
module Collector
class EventQueue
def initialize(dalli)
@dalli = dalli
@queue_mutex = Mutex.new
@queue_user = []
@queue_tweet = []
@queue_favorite = []
@queue_retweet = []
@queue_unfavorite = []
@queue_delete = []
@queue_unauthorized = []
end
def self.start(dalli)
instance = self.new(dalli)
EM.add_periodic_timer(Settings.collector.flush_interval) do
instance.flush
end
instance
end
def flush
@queue_mutex.lock
begin
users, @queue_user = @queue_user, []
tweets, @queue_tweet = @queue_tweet, []
favorites, @queue_favorite = @queue_favorite, []
retweets, @queue_retweet = @queue_retweet, []
unfavorites, @queue_unfavorite = @queue_unfavorite, []
deletes, @queue_delete = @queue_delete, []
unauthorizeds, @queue_unauthorized = @queue_unauthorized, []
ensure
@queue_mutex.unlock
end
users.reverse!.uniq! { |i| i[:id] }
tweets.reverse!.uniq! { |i| i[:id] }
User.create_or_update_bulk_from_json(users)
Tweet.create_bulk_from_json(tweets)
Favorite.create_bulk_from_json(favorites)
Favorite.delete_bulk_from_json(unfavorites) # TODO: race?
Retweet.create_bulk_from_json(retweets)
if deletes.size > 0
Tweet.destroy_bulk_from_json(deletes)
Retweet.delete_bulk_from_json(deletes)
end
if Settings.notification.enabled
NotificationQueue.push(favorites.map { |f| f.dig(:target_object, :id) })
end
if unauthorizeds.size > 0
AccountTokenVerificationJob.perform_later(unauthorizeds.map { |u| u[:id] })
end
end
def push_user(msg)
cache(msg) do
@queue_user << msg[:data]
end
end
def push_tweet(msg)
cache(msg) do
@queue_tweet << msg[:data]
end
end
def push_retweet(msg)
cache(msg) do
@queue_retweet << msg[:data]
end
end
def push_favorite(msg)
cache(msg) do
@queue_favorite << msg[:data]
end
end
def push_unfavorite(msg)
cache(msg) do
@queue_unfavorite << msg[:data]
end
end
def push_delete(msg)
cache(msg) do
@queue_delete << msg[:data]
end
end
def push_unauthorized(unauthorized)
@queue_mutex.synchronize { @queue_unauthorized << unauthorized[:data] }
end
private
def cache(object)
if id = object[:identifier]
if version = object[:version]
cur = @dalli.get(id)
if cur && cur >= version
Rails.logger.debug("EventQueue") { "dup: #{id}/#{cur} <=> #{version}" } if $VERBOSE
return
else
@dalli.set(id, version)
end
end
end
@queue_mutex.synchronize { yield }
end
end
end
|
