From 7d82c76ebdd3bba326dd001844ad95682372bc1a Mon Sep 17 00:00:00 2001
From: Rhenium <rhenium@rhe.jp>
Date: Mon, 20 Jan 2014 14:01:16 +0900
Subject: _(┐「ε:)_
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 app/models/account.rb                      |  5 +---
 lib/aclog/receiver/collector_connection.rb | 37 +++++++++++++++---------------
 2 files changed, 20 insertions(+), 22 deletions(-)

diff --git a/app/models/account.rb b/app/models/account.rb
index 9b55b5f..5ab8181 100644
--- a/app/models/account.rb
+++ b/app/models/account.rb
@@ -5,6 +5,7 @@ class Account < ActiveRecord::Base
 
   belongs_to :user
   scope :active, -> { where(status: Account::ACTIVE) }
+  scope :set_of_collector, -> { active.where("id % ? = ?", Settings.collector.count, collector_id) }
 
   def notification?; notification end
   def private?; private end
@@ -19,10 +20,6 @@ class Account < ActiveRecord::Base
     account
   end
 
-  def self.set_of_collector(collector_id)
-    self.active.where("id % ? = ?", Settings.collector.count, collector_id)
-  end
-
   def deactivate!
     self.status = Account::INACTIVE
     self.save! if self.changed?
diff --git a/lib/aclog/receiver/collector_connection.rb b/lib/aclog/receiver/collector_connection.rb
index 6ba7175..c1258de 100644
--- a/lib/aclog/receiver/collector_connection.rb
+++ b/lib/aclog/receiver/collector_connection.rb
@@ -6,7 +6,7 @@ module Aclog
         @connections = connections
 
         @worker_number = nil
-        @unpacker = MessagePack::Unpacker.new
+        @unpacker = MessagePack::Unpacker.new(symbolize_keys: true)
       end
 
       def send_account(account)
@@ -36,7 +36,7 @@ module Aclog
 
       def receive_data(data)
         @unpacker.feed_each(data) do |msg|
-          unless msg.is_a?(Hash) && msg["type"]
+          unless msg.is_a?(Hash) && msg[:type]
             log(:error, "unknown data: #{msg}")
             send_object(type: "fatal", message: "unknown data")
             close_connection_after_writing
@@ -44,7 +44,7 @@ module Aclog
           end
 
           unless @authorized
-            if msg["type"] == "auth"
+            if msg[:type] == "auth"
               auth(msg)
             else
               log(:warn, "not authorized client: #{msg}")
@@ -54,44 +54,45 @@ module Aclog
             return
           end
 
-          case msg["type"]
+          case msg[:type]
           when "unauthorized"
             @channel << -> {
-              log(:warn, "unauthorized: ##{msg["id"]}/#{msg["user_id"]}")
+              log(:warn, "unauthorized: ##{msg[:id]}/#{msg[:user_id]}")
             }
           when "tweet"
             @channel << -> {
-              log(:debug, "receive tweet: #{msg["id"]}")
-              Tweet.from_receiver(msg)
+              log(:debug, "receive tweet: #{msg[:id]}")
+              Tweet.from_json(msg)
             }
           when "favorite"
             @channel << -> {
-              log(:debug, "receive favorite: #{msg["source"]["id"]} => #{msg["target_object"]["id"]}")
-              if f = Favorite.from_receiver(msg)
+              log(:debug, "receive favorite: #{msg[:source][:id]} => #{msg[:target_object][:id]}")
+              if f = Favorite.from_json(msg)
                 f.tweet.notify_favorite
               end
             }
           when "unfavorite"
             @channel << -> {
-              log(:debug, "receive unfavorite: #{msg["source"]["id"]} => #{msg["target_object"]["id"]}")
-              Favorite.delete_from_receiver(msg)
+              log(:debug, "receive unfavorite: #{msg[:source][:id]} => #{msg[:target_object][:id]}")
+              Favorite.where(user_id: msg[:source][:id], tweet_id: msg[:target_object][:id]).destroy_all
             }
           when "retweet"
             @channel << -> {
-              log(:debug, "receive retweet: #{msg["user"]["id"]} => #{msg["retweeted_status"]["id"]}")
-              Retweet.from_receiver(msg)
+              log(:debug, "receive retweet: #{msg[:user][:id]} => #{msg[:retweeted_status][:id]}")
+              Retweet.from_json(msg)
             }
           when "delete"
             @channel << -> {
-              log(:debug, "receive delete: #{msg["id"]}")
-              Tweet.delete_from_receiver(msg)
+              log(:debug, "receive delete: #{msg[:id]}")
+              Tweet.where(id: msg[:id]).destroy_all
+              Retweet.where(id: msg[:id]).destroy_all
             }
           when "quit"
-            log(:info, "receive quit: #{msg["reason"]}")
+            log(:info, "receive quit: #{msg[:reason]}")
             send_data(type: "quit", message: "Bye")
             close_connection_after_writing
           else
-            log(:warn, "unknown message: #{msg["type"]}")
+            log(:warn, "unknown message: #{msg[:type]}")
             send_object(type: "error", message: "Unknown message type")
           end
         end
@@ -110,7 +111,7 @@ module Aclog
       end
 
       def auth(msg)
-        secret_key = msg["secret_key"]
+        secret_key = msg[:secret_key]
         unless secret_key == Settings.collector.secret_key
           log(:warn, "Invalid secret_key: \"#{secret_key}\"")
           send_object(type: "fatal", message: "invalid secret_key")
-- 
cgit v1.2.3