Kouhei Sutou
null+****@clear*****
Sun Nov 24 17:34:34 JST 2013
Kouhei Sutou 2013-11-24 17:34:34 +0900 (Sun, 24 Nov 2013) New Revision: 9f0ba94a103d114eb71e3df319dd7da81e4de151 https://github.com/droonga/fluent-plugin-droonga/commit/9f0ba94a103d114eb71e3df319dd7da81e4de151 Message: Add Processor that processes requests for a partition Added files: lib/droonga/processor.rb Modified files: lib/droonga/dispatcher.rb lib/droonga/farm.rb lib/droonga/partition.rb Modified: lib/droonga/dispatcher.rb (+1 -1) =================================================================== --- lib/droonga/dispatcher.rb 2013-11-24 17:24:01 +0900 (284e45a) +++ lib/droonga/dispatcher.rb 2013-11-24 17:34:34 +0900 (33082ac) @@ -91,7 +91,7 @@ module Droonga post(message, "type" => type, "synchronous"=> synchronous) else envelope =****@worke*****("body" => message, "type" => type) - @farm.emit(route, envelope, synchronous) + @farm.process(route, envelope, synchronous) end end Modified: lib/droonga/farm.rb (+2 -3) =================================================================== --- lib/droonga/farm.rb 2013-11-24 17:24:01 +0900 (2e6206c) +++ lib/droonga/farm.rb 2013-11-24 17:34:34 +0900 (5749476) @@ -40,9 +40,8 @@ module Droonga end end - # TODO: fix method name - def emit(partition_name, envelope, synchronous) - @partitions[partition_name].emit('', Time.now.to_f, envelope, synchronous) + def process(partition_name, envelope, synchronous) + @partitions[partition_name].process(envelope, synchronous) end end end Modified: lib/droonga/partition.rb (+12 -17) =================================================================== --- lib/droonga/partition.rb 2013-11-24 17:24:01 +0900 (2eb62fe) +++ lib/droonga/partition.rb 2013-11-24 17:34:34 +0900 (e54581d) @@ -19,38 +19,33 @@ require "serverengine" require "droonga/server" require "droonga/worker" -require "droonga/executor" +require "droonga/processor" module Droonga class Partition - DEFAULT_OPTIONS = { - :queue_name => "DroongaQueue", - :n_workers => 0, - } - def initialize(options={}) - @options = DEFAULT_OPTIONS.merge(options) + @options = options + @n_workers = @options[:n_workers] || 0 + @processor = Processor.new(@options) + @supervisor = nil end def start - if @options[:database] && !@options[:database].empty? - Droonga::JobQueue.ensure_schema(@options[:database], - @options[:queue_name]) - end - start_supervisor if @options[:n_workers] > 0 - @executor = Executor.new(@options.merge(:standalone => true)) + start_supervisor if @n_workers > 0 + @processor.start end def shutdown $log.trace("partition: shutdown: start") - @executor.shutdown if @executor + @processor.shutdown shutdown_supervisor if @supervisor $log.trace("partition: shutdown: done") end - def emit(tag, time, record, synchronous=nil) - $log.trace("[#{Process.pid}] tag: <#{tag}> caller: <#{caller.first}>") - @executor.dispatch(tag, time, record, synchronous) + def process(envelope, synchronous=nil) + $log.trace("partition: process: start") + @processor.process(envelope, synchronous) + $log.trace("partition: process: done") end private Added: lib/droonga/processor.rb (+65 -0) 100644 =================================================================== --- /dev/null +++ lib/droonga/processor.rb 2013-11-24 17:34:34 +0900 (01275e5) @@ -0,0 +1,65 @@ +# -*- coding: utf-8 -*- +# +# Copyright (C) 2013 Droonga Project +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License version 2.1 as published by the Free Software Foundation. +# +# This library is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public +# License along with this library; if not, write to the Free Software +# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA + +require "droonga/job_queue" +require "droonga/handler" + +module Droonga + class Processor + def initialize(options={}) + @options = options + @database_name = @options[:database] + @queue_name = @options[:options] || "DroongaQueue" + @n_workers = @options[:n_workers] || 0 + end + + def start + Droonga::JobQueue.ensure_schema(@database_name, + @queue_name) + @job_queue = JobQueue.open(@database_name, @queue_name) + @handler = Handler.new(@options) + end + + def shutdown + $log.trace("processor: shutdown: start") + @handler.shutdown + @job_queue.close + $log.trace("processor: shutdown: done") + end + + def process(envelope, synchronous=nil) + $log.trace("proessor: process: start") + reply_to = envelope["replyTo"] + command = envelope["type"] + if****@handl*****?(command) + $log.trace("proessor: process: handlable: #{command}") + if synchronous.nil? + synchronous =****@handl*****_synchronous?(command) + end + message = ["", Time.now.to_f, envelope] + if @n_workers.zero? or synchronous + @handler.handle(message) + else + @job_queue.push_message(message) + end + else + $log.trace("proessor: process: ignore #{command}") + end + $log.trace("proessor: process: done") + end + end +end -------------- next part -------------- HTML����������������������������...Download