Module: RSMP::Proxy::Modules::Tasks

Included in:
RSMP::Proxy
Defined in:
lib/rsmp/proxy/modules/tasks.rb

Overview

Reader and timer tasks are siblings in one supervised session barrier.

Instance Method Summary collapse

Instance Method Details

#connection_read_failure(reason, text, error) ⇒ Object 



45
46
47
48
49
50
51
52
53
54
 
# File 'lib/rsmp/proxy/modules/tasks.rb', line 45

def connection_read_failure(reason, text, error)
  log(text, level: :warning)
  Result.failure(
    :disconnected,
    message: text,
    source: reason == :peer_closed ? :peer : :transport,
    context: { reason: reason, session_id: @session_id },
    cause: error
  )
end

#log_processing_statistics(result, beginning) ⇒ Object 



81
82
83
84
85
86
87
 
# File 'lib/rsmp/proxy/modules/tasks.rb', line 81

def log_processing_statistics(result, beginning)
  message = result.success? ? result.value : result.failure.context[:message]
  duration = Time.now - beginning
  type = message.respond_to?(:type) ? message.type : 'Unknown'
  m_id = message.respond_to?(:m_id) ? Logger.shorten_message_id(message.m_id) : nil
  log([type, m_id, processing_speed(duration)].compact.join(' '), level: :statistics)
end

#process_received_line(json) ⇒ Object 



74
75
76
77
78
79
 
# File 'lib/rsmp/proxy/modules/tasks.rb', line 74

def process_received_line(json)
  beginning = Time.now
  result = process_packet(json)
  log_processing_statistics(result, beginning)
  result
end

#processing_speed(duration) ⇒ Object 



89
90
91
92
93
 
# File 'lib/rsmp/proxy/modules/tasks.rb', line 89

def processing_speed(duration)
  milliseconds = (duration * 1000).round(4)
  per_second = duration.positive? ? (1.0 / duration).round : Float::INFINITY
  "processed in #{milliseconds}ms, #{per_second}req/s"
end

#read_protocol_line ⇒ Object 



34
35
36
37
38
39
40
41
42
43
 
# File 'lib/rsmp/proxy/modules/tasks.rb', line 34

def read_protocol_line
  json = @protocol.read_line
  raise EOFError, 'Connection closed by peer' unless json

  Result.success(json)
rescue EOFError => e
  connection_read_failure(:peer_closed, 'Connection closed by peer', e)
rescue IOError, Errno::ECONNRESET, Errno::EPIPE => e
  connection_read_failure(:transport_failure, e.message, e)
end

#run_reader(id) ⇒ Object 



22
23
24
25
26
27
28
29
30
31
32
 
# File 'lib/rsmp/proxy/modules/tasks.rb', line 22

def run_reader(id)
  @stream ||= IO::Stream::Buffered.new(@socket)
  @protocol ||= RSMP::Protocol.new(@stream)
  while session_active?(id)
    line = read_protocol_line
    return line if line.failure?

    process_received_line(line.value)
  end
  Result.success(:closed)
end

#run_timer(task, interval, id) ⇒ Object 



108
109
110
111
112
113
114
115
116
117
118
 
# File 'lib/rsmp/proxy/modules/tasks.rb', line 108

def run_timer(task, interval, id)
  next_time = Time.now.to_f
  while session_active?(id)
    result = timer(Clock.now)
    return result if result.failure?

    next_time += interval
    task.sleep([next_time - Time.now.to_f, 0].max) if session_active?(id)
  end
  Result.success(:closed)
end

#session_active?(id = @session_id) ⇒ Boolean 



6
7
8
 
# File 'lib/rsmp/proxy/modules/tasks.rb', line 6

def session_active?(id = @session_id)
  id == @session_id && !@session_closed
end

#session_tasks ⇒ Object 



10
11
12
 
# File 'lib/rsmp/proxy/modules/tasks.rb', line 10

def session_tasks
  @session_tasks ||= Async::Barrier.new(parent: @task)
end

#start_reader ⇒ Object 



14
15
16
17
18
19
20
 
# File 'lib/rsmp/proxy/modules/tasks.rb', line 14

def start_reader
  id = @session_id
  @reader = session_tasks.async do |task|
    task.annotate "reader session #{id}"
    run_reader(id)
  end
end

#start_timer ⇒ Object 



95
96
97
98
99
100
101
102
103
104
105
106
 
# File 'lib/rsmp/proxy/modules/tasks.rb', line 95

def start_timer
  return if @timer&.running?

  id = @session_id
  interval = @site_settings['intervals']['timer'] || 1
  log "Starting timer with interval #{interval} seconds", level: :debug
  @latest_watchdog_received = Clock.now
  @timer = session_tasks.async do |task|
    task.annotate "timer session #{id}"
    run_timer(task, interval, id)
  end
end

#timer(now) ⇒ Object 



120
121
122
123
124
125
126
127
 
# File 'lib/rsmp/proxy/modules/tasks.rb', line 120

def timer(now)
  watchdog_send_timer(now)
  acknowledgement = check_ack_timeout(now)
  return acknowledgement if acknowledgement.failure?

  check_watchdog_timeout(now)
  Result.success
end

#wait_for_session ⇒ Object

Wait for the reader while observing every sibling. A failed timer task raises here with its original exception and backtrace.



58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
 
# File 'lib/rsmp/proxy/modules/tasks.rb', line 58

def wait_for_session
  reader = @reader
  observed = session_tasks.wait do |finished|
    result = finished.wait
    break result if finished.equal?(reader)
    break result if result.is_a?(Result::Failure)
    next if finished.cancelled? || !session_active?

    raise "#{finished.annotation} ended while its connection session was active"
  end
  observed.is_a?(Result::Success) || observed.is_a?(Result::Failure) ? observed : Result.success(:closed)
ensure
  @session_tasks.cancel if @session_tasks && !@session_tasks.empty?
  @session_tasks = nil
end