Class: WSDirector::Protocols::Base

Inherits:
Object
  • Object
show all
Includes:
Utils
Defined in:
lib/wsdirector/protocols/base.rb

Overview

Base protocol describes basic actions

Direct Known Subclasses

ActionCable, Phoenix

Defined Under Namespace

Classes: ReceiveTimeoutError

Constant Summary

Constants included from Utils

Utils::MULTIPLIER_FORMAT

Instance Attribute Summary

Attributes included from Utils

#scale

Instance Method Summary collapse

Methods included from Utils

#parse_multiplier

Constructor Details

#initialize(task, scale: 1, logger: nil, id: nil, color: nil) ⇒ Base

Returns a new instance of Base.



107
108
109
110
111
112
113
 
# File 'lib/wsdirector/protocols/base.rb', line 107

def initialize(task, scale: 1, logger: nil, id: nil, color: nil)
  @task = task
  @scale = scale
  @logger = logger
  @id = id
  @color = color
end

Instance Method Details

#debug(step) ⇒ Object

Prints provided message



152
153
154
155
156
 
# File 'lib/wsdirector/protocols/base.rb', line 152

def debug(step)
  with_logger do
    log(nil) { step.fetch("message") }
  end
end

#handle_step(step) ⇒ Object

Raises:



123
124
125
126
127
128
129
130
 
# File 'lib/wsdirector/protocols/base.rb', line 123

def handle_step(step)
  type = step.delete("type")
  raise Error, "Unknown step: #{type}" unless respond_to?(type)

  return unless task.sampled?(step)

  public_send(type, step)
end

#init_client ⇒ Object 



115
116
117
118
119
120
121
 
# File 'lib/wsdirector/protocols/base.rb', line 115

def init_client(...)
  log { "Connecting" }

  @client = build_client(...)

  log(:done) { "Connected" }
end

#receive(step) ⇒ Object 



158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
 
# File 'lib/wsdirector/protocols/base.rb', line 158

def receive(step)
  expected = step["data"] || PartialMatcher.new(step["data>"])
  ordered = step["ordered"]
  timeout = step.fetch("timeout", 5).to_f

  log { "Receive a message in #{timeout}s: #{expected.inspect.truncate(100)}" }

  start = Time.now.to_f
  received = nil

  client.each_message do |msg, id|
    received = msg
    if expected.matches?(msg)
      client.consumed(id)
      break
    end

    if ordered
      raise UnmatchedExpectationError, prepare_receive_error(expected, received)
    end

    if Time.now.to_f - start > timeout
      raise ReceiveTimeoutError
    end
  end

  if step["print"]
    debug({"message" => received})
  end

  log(:done) { "Received a message: #{received&.truncate(100)}" }
rescue ThreadError, ReceiveTimeoutError
  if received
    raise UnmatchedExpectationError, prepare_receive_error(expected, received)
  else
    raise NoMessageError, "Expected to receive #{expected.inspect} but nothing has been received"
  end
end

#receive_all(step) ⇒ Object 



197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
 
# File 'lib/wsdirector/protocols/base.rb', line 197

def receive_all(step)
  messages = step.delete("messages")
  raise ArgumentError, "Messages array must be specified" if
    messages.nil? || messages.empty?

  expected =
    messages.map do |msg|
      multiplier = parse_multiplier(msg.delete("multiplier") || "1")
      [msg["data"] || PartialMatcher.new(msg["data>"]), multiplier]
    end.to_h

  total_expected = expected.values.sum
  total_received = 0

  log { "Receive #{total_expected} messages" }

  total_expected.times do
    received = client.receive

    total_received += 1

    match = expected.find { |k, _| k.matches?(received) }

    raise UnexpectedMessageError, "Unexpected message received: #{received}" if
      match.nil?

    expected[match.first] -= 1
    expected.delete(match.first) if expected[match.first].zero?
  end

  log(:done) { "Received #{total_expected} messages" }
rescue ThreadError
  raise NoMessageError,
    "Expected to receive #{total_expected} messages " \
    "but received only #{total_received}"
end

#send(step) ⇒ Object

rubocop: enable Metrics/CyclomaticComplexity



235
236
237
238
239
240
241
242
 
# File 'lib/wsdirector/protocols/base.rb', line 235

def send(step)
  data = step.fetch("data")
  data = JSON.generate(data) if data.is_a?(Hash)

  client.send(data)

  log(nil) { "Sent message: #{data.truncate(50)}" }
end

#sleep(step) ⇒ Object

Sleeps for a specified number of seconds.

If “shift” is provided than the initial value is shifted by random number from (-shift, shift).

Set “debug” to true to print the delay time.



138
139
140
141
142
143
144
145
146
147
148
149
 
# File 'lib/wsdirector/protocols/base.rb', line 138

def sleep(step)
  delay = step.fetch("time").to_f
  shift = step.fetch("shift", 0).to_f

  delay = delay - shift * rand + shift * rand

  log { "Sleep for #{delay}s" }

  Kernel.sleep delay if delay > 0

  log(:done) { "Slept for #{delay}s" }
end

#to_proc ⇒ Object 



250
251
252
 
# File 'lib/wsdirector/protocols/base.rb', line 250

def to_proc
  proc { |step| handle_step(step) }
end

#wait_all(_step) ⇒ Object 



244
245
246
247
248
 
# File 'lib/wsdirector/protocols/base.rb', line 244

def wait_all(_step)
  log { "Wait all clients" }
  task.global_holder.wait_all
  log { "All clients" }
end