Navigation Menu

Skip to content

Commit

Permalink
Add ability to detect nil message error
Browse files Browse the repository at this point in the history
  • Loading branch information
piroor committed Apr 15, 2015
1 parent 143b7bb commit 332369a
Show file tree
Hide file tree
Showing 2 changed files with 34 additions and 0 deletions.
24 changes: 24 additions & 0 deletions lib/droonga/client/connection/droonga-protocol/coolio.rb
Expand Up @@ -21,6 +21,11 @@ class Client
module Connection
class DroongaProtocol
class Coolio
attr_writer :on_error

class NilMessage < StandardError
end

class Request
def initialize(receiver, id, loop)
@receiver = receiver
Expand Down Expand Up @@ -103,6 +108,7 @@ def on_connect_failed

class Receiver < ::Coolio::TCPServer
attr_accessor :max_messages
attr_writer :on_error

def initialize(*args)
super(*args) do |engine|
Expand Down Expand Up @@ -159,7 +165,14 @@ def handle_engine(engine)
n_messages = 0
on_read = lambda do |data|
unpacker.feed_each(data) do |fluent_message|
unless fluent_message
on_error(NilMessage.new("unpacker.feed_each"))
end
tag, time, droonga_message = fluent_message
unless droonga_message
on_error(NilMessage.new("unpacker.feed_each",
:fluent_message => fluent_message.inspect))
end
id = droonga_message["inReplyTo"]
request = @requests[id]
n_messages += 1
Expand All @@ -184,6 +197,10 @@ def handle_engine(engine)
on_close.call
end
end

def on_error(error)
@on_error.call(error) if @on_error
end
end

def initialize(host, port, tag, options={})
Expand All @@ -200,6 +217,9 @@ def initialize(host, port, tag, options={})
@receiver_host = @options[:receiver_host] || Socket.gethostname
@receiver_port = @options[:receiver_port] || 0
@receiver = Receiver.new(@receiver_host, @receiver_port)
@receiver.on_error = lambda do |error|
on_error(error)
end
@receiver.attach(@loop)
end

Expand Down Expand Up @@ -284,6 +304,10 @@ def close
@sender.close
@receiver.close
end

def on_error(error)
@on_error.call(error) if @on_error
end
end
end
end
Expand Down
10 changes: 10 additions & 0 deletions lib/droonga/client/connection/droonga-protocol/thread.rb
Expand Up @@ -25,6 +25,11 @@ class DroongaProtocol
class Thread
DEFAULT_TIMEOUT_SECONDS = 10

attr_writer :on_error

class NilMessage < StandardError
end

class Request
def initialize(thread)
@thread = thread
Expand Down Expand Up @@ -153,13 +158,18 @@ def receive(receiver, options)
}
begin
receiver.receive(receive_options) do |response|
on_error(NilMessage.new("receiver.receive"))
yield(response)
end
ensure
receiver.close
end
end

def on_error(error)
@on_error.call(error) if @on_error
end

class Receiver
def initialize(options={})
host = options[:host] || Socket.gethostname
Expand Down

0 comments on commit 332369a

Please sign in to comment.