Class: Droonga::Client::Connection::DroongaProtocol::Thread
- Inherits:
-
Object
- Object
- Droonga::Client::Connection::DroongaProtocol::Thread
show all
- Defined in:
- lib/droonga/client/connection/droonga-protocol/thread.rb
Defined Under Namespace
Classes: NilMessage, Receiver, ReceiverError, Request
Constant Summary
collapse
- DEFAULT_TIMEOUT_SECONDS =
10
Instance Attribute Summary collapse
Instance Method Summary
collapse
Constructor Details
#initialize(host, port, tag, options = {}) ⇒ Thread
Returns a new instance of Thread.
49
50
51
52
53
54
55
56
57
58
59
|
# File 'lib/droonga/client/connection/droonga-protocol/thread.rb', line 49
def initialize(host, port, tag, options={})
@host = host
@port = port
@tag = tag
default_options = {
:timeout => 1,
}
@options = default_options.merge(options)
@logger = Fluent::Logger::FluentLogger.new(@tag, @options)
@timeout = @options[:timeout]
end
|
Instance Attribute Details
#on_error=(value) ⇒ Object
Sets the attribute on_error
28
29
30
|
# File 'lib/droonga/client/connection/droonga-protocol/thread.rb', line 28
def on_error=(value)
@on_error = value
end
|
Instance Method Details
#close ⇒ Object
152
153
154
|
# File 'lib/droonga/client/connection/droonga-protocol/thread.rb', line 152
def close
@logger.close
end
|
#request(message, options = {}, &block) ⇒ Object
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
|
# File 'lib/droonga/client/connection/droonga-protocol/thread.rb', line 61
def request(message, options={}, &block)
receiver = create_receiver
receiver.on_error = lambda do |error|
on_error(ReceiverError.new(error))
end
message = message.dup
message["replyTo"] = "#{receiver.host}:#{receiver.port}/droonga"
send(message, options)
sync = block.nil?
if sync
responses = []
receive(receiver, options) do |response|
responses << response
end
if responses.size > 1
responses
else
responses.first
end
else
thread = ::Thread.new do
receive(receiver, options, &block)
end
Request.new(thread)
end
end
|
#send(message, options = {}, &block) ⇒ Object
148
149
150
|
# File 'lib/droonga/client/connection/droonga-protocol/thread.rb', line 148
def send(message, options={}, &block)
@logger.post("message", message)
end
|
#subscribe(message, options = {}, &block) ⇒ Object
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
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
|
# File 'lib/droonga/client/connection/droonga-protocol/thread.rb', line 89
def subscribe(message, options={}, &block)
receiver = create_receiver
receive_end_point = "#{receiver.host}:#{receiver.port}/droonga"
message = message.dup
message["replyTo"] = receive_end_point
message["from"] = receive_end_point
send(message, options)
subscription_timeout = options[:subscription_timeout]
max_messages = options[:max_messages]
start = Time.now
receive_options = {
:timeout => options[:timeout] || DEFAULT_TIMEOUT_SECONDS,
}
n_messages = 0
sync = block.nil?
if sync
Enumerator.new do |yielder|
loop do
receiver.receive(receive_options) do |object|
yielder << object
n_messages += 1
end
if max_messages and
n_messages >= max_messages
break
end
if subscription_timeout
elapsed_seconds = Time.now - start
break if elapsed_seconds >= subscription_timeout
end
end
receiver.close
end
else
thread = ::Thread.new do
begin
loop do
receiver.receive(receive_options) do |message|
block.call(message)
n_messages += 1
end
if max_messages and
n_messages >= max_messages
break
end
if subscription_timeout
elapsed_seconds = Time.now - start
break if elapsed_seconds >= subscription_timeout
end
end
ensure
receiver.close
end
end
Request.new(thread)
end
end
|