class EventHub::Message

Constants

REQUIRED_HEADERS

Headers that are required (value can be nil) in order to pass valid?

VERSION

Attributes

body[RW]
header[RW]
raw[RW]
routing_key[RW]
vhost[RW]

Public Class Methods

from_json(raw) click to toggle source
# File lib/eventhub/message.rb, line 39
def self.from_json(raw)
  data = JSON.parse(raw)
  Message.new(data.get('header'), data.get('body'),raw)
rescue => e
  Message.new({ "status" =>  { "code" => STATUS_INVALID, "message" => "JSON parse error: #{e}" }} ,{ "original_message_base64_encoded" => Base64.encode64(raw)},raw)
end
new(header = nil, body = nil, raw = nil) click to toggle source
# File lib/eventhub/message.rb, line 46
def initialize(header = nil, body = nil, raw = nil)

  @header = header || {}
  @body   = body || {}
  @raw    = raw

  # set message defaults, that we have required headers
  @header.set('message_id', UUIDTools::UUID.timestamp_create.to_s, false)
  @header.set('version', VERSION, false)
  @header.set('created_at', now_stamp, false)

  @header.set('origin.module_id', 'undefined', false)
  @header.set('origin.type', 'undefined', false)
  @header.set('origin.site_id', 'undefined', false)

  @header.set('process.name', 'undefined', false)
  @header.set('process.execution_id', UUIDTools::UUID.timestamp_create.to_s, false)
  @header.set('process.step_position', 0, false)

  @header.set('status.retried_count', 0, false)
  @header.set('status.code', STATUS_INITIAL, false)
  @header.set('status.message', '', false)

end
translate_status_code(code) click to toggle source
# File lib/eventhub/message.rb, line 137
def self.translate_status_code(code)
  case code
    when EventHub::STATUS_INITIAL            then return 'STATUS_INITIAL'
    when EventHub::STATUS_SUCCESS            then return 'STATUS_SUCCESS'
    when EventHub::STATUS_RETRY              then return 'STATUS_RETRY'
    when EventHub::STATUS_RETRY_PENDING      then return 'STATUS_RETRY_PENDING'
    when EventHub::STATUS_INVALID            then return 'STATUS_INVALID'
    when EventHub::STATUS_DEADLETTER         then return 'STATUS_DEADLETTER'
    when EventHub::STATUS_SCHEDULE           then return 'STATUS_SCHEDULE'
    when EventHub::STATUS_SCHEDULE_RETRY     then return 'STATUS_SCHEDULE_RETRY'
    when EventHub::STATUS_SCHEDULE_PENDING   then return 'STATUS_SCHEDULE_PENDING'
  end
end

Public Instance Methods

append_to_execution_history(processor_name) click to toggle source
# File lib/eventhub/message.rb, line 130
def append_to_execution_history(processor_name)
  unless header.get('execution_history')
    header.set('execution_history', [])
  end
  header.get('execution_history') << {'processor' => processor_name, 'timestamp' => now_stamp}
end
copy(status_code = STATUS_SUCCESS) click to toggle source

copies the message and set's provided status code (default: success), actual stamp, and a new message id

# File lib/eventhub/message.rb, line 117
def copy(status_code = STATUS_SUCCESS)

  # use Marshal dump and load to make a deep object copy
  copied_header = Marshal.load( Marshal.dump(header))
  copied_body   = Marshal.load( Marshal.dump(body))

  copied_header.set("message_id",UUIDTools::UUID.timestamp_create.to_s)
  copied_header.set("created_at",now_stamp)
  copied_header.set("status.code",status_code)

  Message.new(copied_header, copied_body)
end
initial?() click to toggle source
# File lib/eventhub/message.rb, line 84
def initial?
  self.status_code == STATUS_INITIAL
end
invalid?() click to toggle source
# File lib/eventhub/message.rb, line 92
def invalid?
  self.status_code == STATUS_INVALID
end
retry?() click to toggle source
# File lib/eventhub/message.rb, line 80
def retry?
  self.status_code == STATUS_RETRY
end
retry_pending?() click to toggle source
# File lib/eventhub/message.rb, line 88
def retry_pending?
  self.status_code == STATUS_RETRY_PENDING
end
schedule?() click to toggle source
# File lib/eventhub/message.rb, line 96
def schedule?
  self.status_code == STATUS_SCHEDULE
end
schedule_pending?() click to toggle source
# File lib/eventhub/message.rb, line 104
def schedule_pending?
  self.status_code == STATUS_SCHEDULE_PENDING
end
schedule_retry?() click to toggle source
# File lib/eventhub/message.rb, line 100
def schedule_retry?
  self.status_code == STATUS_SCHEDULE_RETRY
end
success?() click to toggle source
# File lib/eventhub/message.rb, line 76
def success?
  self.status_code == STATUS_SUCCESS
end
to_json() click to toggle source
# File lib/eventhub/message.rb, line 108
def to_json
  {'header' => self.header, 'body' => self.body}.to_json
end
to_s() click to toggle source
# File lib/eventhub/message.rb, line 112
def to_s
  "Msg: process [#{self.process_name},#{self.process_step_position},#{self.process_execution_id}], status [#{self.status_code},#{self.status_message},#{self.status_retried_count}]"
end
valid?() click to toggle source
# File lib/eventhub/message.rb, line 71
def valid?
  # check for existence and defined value
  REQUIRED_HEADERS.all? { |key| @header.all_keys_with_path.include?(key) && !!self.send(key.gsub(/\./,"_").to_sym)}
end