Module: RSMP::Proxy::Modules::Acknowledgements

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

Overview

Message acknowledgement handling Manages sending/receiving acks and nacks, and tracking acknowledged messages

Instance Method Summary collapse

Instance Method Details

#acknowledge(original) ⇒ Object

Raises:



7
8
9
10
11
12
13
14
 
# File 'lib/rsmp/proxy/modules/acknowledgements.rb', line 7

def acknowledge(original)
  raise ArgumentError, 'original message is required' unless original

  ack = MessageAck.build_from(original)
  ack.original = original.clone
  send_generated_message ack, "for #{ack.original.type} #{original.m_id_short}"
  check_ingoing_acknowledged original
end

#acknowledged_first_ingoing(message) ⇒ Object 



90
 
# File 'lib/rsmp/proxy/modules/acknowledgements.rb', line 90

def acknowledged_first_ingoing(message); end

#acknowledged_first_outgoing(message) ⇒ Object 



88
 
# File 'lib/rsmp/proxy/modules/acknowledgements.rb', line 88

def acknowledged_first_outgoing(message); end

#check_ack_timeout(now) ⇒ Object 



39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
 
# File 'lib/rsmp/proxy/modules/acknowledgements.rb', line 39

def check_ack_timeout(now)
  timeout = @site_settings['timeouts']['acknowledgement']
  # hash cannot be modify during iteration, so clone it
  @awaiting_acknowledgement.clone.each_pair do |_m_id, message|
    latest = message.timestamp + timeout
    next unless now > latest

    str = "No acknowledgements for #{message.type} #{message.m_id_short} within #{timeout} seconds"
    log str, level: :error
    failure = Failure.new(
      code: :missing_acknowledgement,
      message: str,
      source: :peer,
      context: { message: message, reason: :missing_acknowledgement, session_id: session_id }
    )
    distribute_event(
      Event.new(
        type: :missing_acknowledgement,
        source: self,
        session_id: session_id,
        message: message,
        failure: failure
      )
    )
    return Result.failure(failure: failure)
  end

  Result.success
end

#check_ingoing_acknowledged(message) ⇒ Object 



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

def check_ingoing_acknowledged(message)
  return if @ingoing_acknowledged[message.type]

  @ingoing_acknowledged[message.type] = true
  acknowledged_first_ingoing message
end

#check_outgoing_acknowledged(message) ⇒ Object

TODO: this might be better handled by a proper event machine using e.g. the EventMachine gem



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

def check_outgoing_acknowledged(message)
  return if @outgoing_acknowledged[message.type]

  @outgoing_acknowledged[message.type] = true
  acknowledged_first_outgoing message
end

#dont_acknowledge(original, prefix = nil, reason = nil, force: true) ⇒ Object

Raises:



16
17
18
19
20
21
22
23
24
25
26
27
 
# File 'lib/rsmp/proxy/modules/acknowledgements.rb', line 16

def dont_acknowledge(original, prefix = nil, reason = nil, force: true)
  raise ArgumentError, 'original message is required' unless original

  str = [prefix, reason].join(' ')
  log str, message: original, level: :warning if reason
  message = MessageNotAck.new({
                                'oMId' => original.m_id,
                                'rea' => reason || 'Unknown reason'
                              })
  message.original = original.clone
  send_generated_message message, "for #{original.type} #{original.m_id_short}", force: force
end

#dont_expect_acknowledgement(message) ⇒ Object 



35
36
37
 
# File 'lib/rsmp/proxy/modules/acknowledgements.rb', line 35

def dont_expect_acknowledgement(message)
  @awaiting_acknowledgement.delete message.attribute('oMId')
end

#expect_acknowledgement(message) ⇒ Object 



29
30
31
32
33
 
# File 'lib/rsmp/proxy/modules/acknowledgements.rb', line 29

def expect_acknowledgement(message)
  return if message.is_a?(MessageAck) || message.is_a?(MessageNotAck)

  @awaiting_acknowledgement[message.m_id] = message
end

#find_original_for_message(message) ⇒ Object 



69
70
71
 
# File 'lib/rsmp/proxy/modules/acknowledgements.rb', line 69

def find_original_for_message(message)
  @awaiting_acknowledgement[message.attribute('oMId')]
end

#log_acknowledgement_for_original(message, original) ⇒ Object 



130
131
132
133
134
135
136
137
138
139
 
# File 'lib/rsmp/proxy/modules/acknowledgements.rb', line 130

def log_acknowledgement_for_original(message, original)
  str = "Received #{message.type} for #{original.type} #{message.attribute('oMId')[0..3]}"
  if message.type == 'MessageNotAck'
    reason = message.attributes['rea']
    str = "#{str}: #{reason}" if reason
    log str, message: message, level: :warning
  else
    log str, message: message, level: :log
  end
end

#log_acknowledgement_for_unknown(message) ⇒ Object 



141
142
143
144
 
# File 'lib/rsmp/proxy/modules/acknowledgements.rb', line 141

def log_acknowledgement_for_unknown(message)
  log "Received #{message.type} for unknown message #{message.attribute('oMId')[0..3]}", message: message,
                                                                                         level: :warning
end

#process_ack(message) ⇒ Object 



92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
 
# File 'lib/rsmp/proxy/modules/acknowledgements.rb', line 92

def process_ack(message)
  original = find_original_for_message message
  if original
    dont_expect_acknowledgement message
    message.original = original
    log_acknowledgement_for_original message, original

    case original.type
    when 'Version'
      version_acknowledged
    when 'ComponentList'
      component_list_acknowledged
    when 'StatusSubscribe'
      status_subscribe_acknowledged original
    end

    check_outgoing_acknowledged original

    @acknowledgements[original.m_id] = message
    @acknowledgement_condition.signal message
  else
    log_acknowledgement_for_unknown message
  end
end

#process_not_ack(message) ⇒ Object 



117
118
119
120
121
122
123
124
125
126
127
128
 
# File 'lib/rsmp/proxy/modules/acknowledgements.rb', line 117

def process_not_ack(message)
  original = find_original_for_message message
  if original
    dont_expect_acknowledgement message
    message.original = original
    log_acknowledgement_for_original message, original
    @acknowledgements[original.m_id] = message
    @acknowledgement_condition.signal message
  else
    log_acknowledgement_for_unknown message
  end
end

#status_subscribe_acknowledged(original) ⇒ Object 



146
147
148
149
150
151
152
153
154
155
 
# File 'lib/rsmp/proxy/modules/acknowledgements.rb', line 146

def status_subscribe_acknowledged(original)
  component = find_component original.attribute('cId')
  return unless component

  short = Message.shorten_m_id original.m_id
  subscribe_list = original.attributes['sS']
  log "StatusSubscribe #{short} acknowledged, allowing repeated status values for #{subscribe_list}",
      level: :info
  component.allow_repeat_updates subscribe_list
end