diff options
author | rhenium <re4k@re4k.info> | 2013-05-11 14:34:41 +0900 |
---|---|---|
committer | rhenium <re4k@re4k.info> | 2013-05-11 14:34:41 +0900 |
commit | 4103946380a5badf209b659867d826cf6e6cd381 (patch) | |
tree | 469d27d2bde73f55dc597b5a4837fb6bafaad7c6 /lib | |
parent | e5accc8e74bf2a6f66638b1085ee7cc85abd281c (diff) | |
download | aclog-4103946380a5badf209b659867d826cf6e6cd381.tar.gz |
clean receiver
Diffstat (limited to 'lib')
-rw-r--r-- | lib/aclog/receiver/collector_connection.rb (renamed from lib/aclog/receiver/collector_server.rb) | 2 | ||||
-rw-r--r-- | lib/aclog/receiver/register_server.rb | 27 | ||||
-rw-r--r-- | lib/aclog/receiver/worker.rb | 30 |
3 files changed, 33 insertions, 26 deletions
diff --git a/lib/aclog/receiver/collector_server.rb b/lib/aclog/receiver/collector_connection.rb index 1130a25..ca37d55 100644 --- a/lib/aclog/receiver/collector_server.rb +++ b/lib/aclog/receiver/collector_connection.rb @@ -3,7 +3,7 @@ require "time" module Aclog module Receiver - class CollectorServer < EM::Connection + class CollectorConnection < EM::Connection def initialize(connections) @connections = connections diff --git a/lib/aclog/receiver/register_server.rb b/lib/aclog/receiver/register_server.rb new file mode 100644 index 0000000..0358c23 --- /dev/null +++ b/lib/aclog/receiver/register_server.rb @@ -0,0 +1,27 @@ +# -*- coding: utf-8 -*- +module Aclog + module Receiver + class RegisterServer + def initialize(connections) + @connections = connections + end + + def register(account_) + account = Marshal.load(account_) + con_num = account.id % Settings.worker_count + con = @connections[con_num] + if con + con.send_account(account) + Rails.logger.info("Sent account: connection_number: #{con_num} / account_id: #{account.id}") + else + Rails.logger.info("Connection not found: connection_number: #{con_num} / account_id: #{account.id}") + end + end + + def unregister(account) + # TODO + end + end + end +end + diff --git a/lib/aclog/receiver/worker.rb b/lib/aclog/receiver/worker.rb index fe383e2..330d872 100644 --- a/lib/aclog/receiver/worker.rb +++ b/lib/aclog/receiver/worker.rb @@ -4,28 +4,6 @@ require "msgpack/rpc/transport/unix" module Aclog module Receiver class Worker < DaemonSpawn::Base - class RegisterServer - def initialize(connections) - @connections = connections - end - - def register(account_) - account = Marshal.load(account_) - con_num = account.id % Settings.worker_count - con = @connections[con_num] - if con - con.send_account(account) - Rails.logger.info("Sent account: connection_number: #{con_num} / account_id: #{account.id}") - else - Rails.logger.info("Connection not found: connection_number: #{con_num} / account_id: #{account.id}") - end - end - - def unregister(account) - # TODO - end - end - def initialize(opts = {}) super(opts) unless opts.empty? _logger = Logger.new(STDOUT) @@ -34,13 +12,16 @@ module Aclog end def start(args) + _sock_path = File.join(Rails.root, "tmp", "sockets", "receiver.sock") + Rails.logger.info("Receiver started") + File.delete(_sock_path) if File.exists?(_sock_path) EM.run do connections = {} - collector_server = EM.start_server("0.0.0.0", Settings.listen_port, Aclog::Receiver::CollectorServer, connections) + collector_server = EM.start_server("0.0.0.0", Settings.listen_port, CollectorConnection, connections) - reg_svr_listener = MessagePack::RPC::UNIXServerTransport.new(File.join(Rails.root, "tmp", "sockets", "receiver.sock")) + reg_svr_listener = MessagePack::RPC::UNIXServerTransport.new(_sock_path) register_server = MessagePack::RPC::Server.new register_server.listen(reg_svr_listener, RegisterServer.new(connections)) EM.defer { register_server.run } @@ -48,7 +29,6 @@ module Aclog stop = Proc.new do EM.stop_server(collector_server) register_server.close - File.delete(File.join(Rails.root, "tmp", "sockets", "receiver.sock")) EM.stop end Signal.trap(:INT, &stop) |