Module: RSMP::Proxy::Modules::Send

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

Overview

Message sending functionality Handles sending messages, validation, and buffering

Instance Method Summary collapse

Instance Method Details

#apply_nts_message_attributes(message) ⇒ Object 



79
80
81
82
83
84
 
# File 'lib/rsmp/proxy/modules/send.rb', line 79

def apply_nts_message_attributes(message)
  return if core_3_3?

  message.attributes['ntsOId'] = main && main.ntsoid ? main.ntsoid : ''
  message.attributes['xNId'] = main && main.xnid ? main.xnid : ''
end

#buffer_message(message, error = nil) ⇒ Object 



48
49
50
51
52
 
# File 'lib/rsmp/proxy/modules/send.rb', line 48

def buffer_message(message, error = nil)
  str = "Cannot send #{message.type} because the connection is closed."
  log str, message: message, level: :error
  raise error if error
end

#handle_send_schema_error(message, error) ⇒ Object 



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

def handle_send_schema_error(message, error)
  schemas_string = error.schemas.map { |schema| "#{schema.first}: #{schema.last}" }.join(', ')
  str = "Could not send #{message.type} because schema validation failed (#{schemas_string}): #{error.message}"
  log str, message: message, level: :error
  distribute_error error.exception("#{str} #{message.json}")
end

#log_send(message, reason = nil) ⇒ Object 



54
55
56
57
58
59
60
61
62
63
64
65
66
 
# File 'lib/rsmp/proxy/modules/send.rb', line 54

def log_send(message, reason = nil)
  str = if reason
          "Sent #{message.type} #{reason}"
        else
          "Sent #{message.type}"
        end

  if message.type == 'MessageNotAck'
    log str, message: message, level: :warning
  else
    log str, message: message, level: :log
  end
end

#send_message(message, reason = nil, validate: true, force: false, buffer: true) ⇒ Object 



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

def send_message(message, reason = nil, validate: true, force: false, buffer: true)
  with_send_ready(message, force: force, buffer: buffer) do
    write_message(message, validate: validate)
    expect_acknowledgement message
    distribute message
    log_send message, reason
  end
rescue NotReady, IOError => e
  raise e unless buffer

  buffer_message message, e
rescue SchemaError, RSMP::Schema::Error => e
  handle_send_schema_error(message, e)
end

#send_message_and_collect(message, collector, validate: true) ⇒ Object 



68
69
70
71
72
73
74
75
76
77
 
# File 'lib/rsmp/proxy/modules/send.rb', line 68

def send_message_and_collect(message, collector, validate: true)
  task = @task.async do |t|
    t.annotate 'send_message_and_collect'
    collector.collect
    collector
  end
  send_message message, validate: validate
  collector = task.wait
  { sent: message, collector: collector }
end

#with_send_ready(message, force:, buffer:) ⇒ Object 



29
30
31
32
33
34
35
36
 
# File 'lib/rsmp/proxy/modules/send.rb', line 29

def with_send_ready(message, force:, buffer:)
  return yield if force || connected?

  error = NotReady.new
  raise error unless buffer

  buffer_message(message, error)
end

#write_message(message, validate:) ⇒ Object

Raises:

  • (IOError) 


38
39
40
41
42
43
44
45
46
 
# File 'lib/rsmp/proxy/modules/send.rb', line 38

def write_message(message, validate:)
  raise IOError unless @protocol

  message.direction = :out
  message.encode_for(schemas) unless validate == false
  message.generate_json
  message.validate schemas unless validate == false
  @protocol.write_lines message.json
end