Sha256: 1f570de2a0c641c70d5cebd054ddff83cbbafbd14781de127a6c14d76fa9a750
Contents?: true
Size: 1.52 KB
Versions: 29
Compression:
Stored size: 1.52 KB
Contents
require_relative 'compression'
module Pheme
class TopicPublisher
include Compression
#
# Constant with message size limit.
# The message size also includes some metadata: 'name' and 'type'.
# We give ourselves a buffer for this metadata.
#
# Source: https://docs.aws.amazon.com/sns/latest/dg/SNSMessageAttributes.html#SNSMessageAttributesNTV
#
SNS_SIZE_LIMIT = 256.kilobytes
EXPECTED_METADATA_SIZE = 1.kilobyte
MESSAGE_SIZE_LIMIT = SNS_SIZE_LIMIT - EXPECTED_METADATA_SIZE
class << self
attr_reader :_topic_arn
def topic_arn(topic_arn)
@_topic_arn = topic_arn
end
end
def initialize(topic_arn: self.class._topic_arn)
raise ArgumentError, "must specify non-nil topic_arn" if topic_arn.blank?
@topic_arn = topic_arn
end
attr_accessor :topic_arn
def publish_events
raise NotImplementedError
end
def publish(message, sns_client: Pheme.configuration.sns_client, message_attributes: nil)
payload = {
message: "#{self.class} publishing message to #{topic_arn}",
body: message,
publisher: self.class.to_s,
topic_arn: topic_arn,
}
Pheme.logger.info(payload.to_json)
sns_client.publish(topic_arn: topic_arn, message: serialize(message), message_attributes: message_attributes)
end
def serialize(message)
message = message.to_json unless message.is_a? String
return compress(message) if message.bytesize > MESSAGE_SIZE_LIMIT
message
end
end
end
Version data entries
29 entries across 29 versions & 1 rubygems