Sha256: 400477699626b9480a0a2bb851ffbf5b77e3d3de3124d7385c9a31a1624384c3
Contents?: true
Size: 1.79 KB
Versions: 11
Compression:
Stored size: 1.79 KB
Contents
require 'socket'
module Propono
class PublisherError < ProponoError
end
class Publisher
include Sns
def self.publish(topic_id, message, options = {})
new(topic_id, message, options).publish
end
attr_reader :topic_id, :message, :protocol, :id
def initialize(topic_id, message, options = {})
raise PublisherError.new("Topic is nil") if topic_id.nil?
raise PublisherError.new("Message is nil") if message.nil?
options = options.symbolize_keys
@topic_id = topic_id
@message = message
@protocol = options.fetch(:protocol, :sns).to_sym
@id = SecureRandom.hex(3)
@id = "#{options[:id]}-#{@id}" if options[:id]
end
def publish
Propono.config.logger.info "Propono [#{id}]: Publishing #{message} to #{topic_id} via #{protocol}"
send("publish_via_#{protocol}")
end
private
def publish_via_sns
topic = TopicCreator.find_or_create(topic_id)
Thread.new do
begin
sns.publish(topic.arn, body.to_json)
rescue => e
Propono.config.logger.error "Propono [#{id}]: Failed to send via sns : #{e}"
end
end
end
def publish_via_udp
payload = body.merge(topic: topic_id).to_json
UDPSocket.new.send(payload, 0, Propono.config.udp_host, Propono.config.udp_port)
rescue => e
Propono.config.logger.error "Propono [#{id}]: Failed to send : #{e}"
end
def publish_via_tcp
payload = body.merge(topic: topic_id).to_json
socket = TCPSocket.new(Propono.config.tcp_host, Propono.config.tcp_port)
socket.write payload
socket.close
rescue => e
Propono.config.logger.error "Propono [#{id}]: Failed to send : #{e}"
end
def body
{
id: id,
message: message
}
end
end
end
Version data entries
11 entries across 11 versions & 1 rubygems