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

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.



104
105
106
107
108
109
110
 
# File 'lib/wsdirector/protocols/base.rb', line 104

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



149
150
151
152
153
 
# File 'lib/wsdirector/protocols/base.rb', line 149

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

#handle_step(step) ⇒ Object

Raises:



120
121
122
123
124
125
126
127
 
# File 'lib/wsdirector/protocols/base.rb', line 120

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 



112
113
114
115
116
117
118
 
# File 'lib/wsdirector/protocols/base.rb', line 112

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

  @client = build_client(...)

  log(:done) { "Connected" }
end

#receive(step) ⇒ Object 



155
156
157
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
 
# File 'lib/wsdirector/protocols/base.rb', line 155

def receive(step)
  expected = step["data"] || PartialMatcher.new(step["data>"])
  ordered = step["ordered"]

  log { "Receive a message: #{expected.truncate(50)}" }

  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
  end

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

#receive_all(step) ⇒ Object 



184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
 
# File 'lib/wsdirector/protocols/base.rb', line 184

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



222
223
224
225
226
227
228
229
 
# File 'lib/wsdirector/protocols/base.rb', line 222

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.



135
136
137
138
139
140
141
142
143
144
145
146
 
# File 'lib/wsdirector/protocols/base.rb', line 135

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 



237
238
239
 
# File 'lib/wsdirector/protocols/base.rb', line 237

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

#wait_all(_step) ⇒ Object 



231
232
233
234
235
 
# File 'lib/wsdirector/protocols/base.rb', line 231

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