Support communication with AWS SNS SDK. (#132)

* Support communication with AWS SNS SDK.
* Remove files we shouldn't install and make logging better.

Author: TreyE

event_source.gemspec

@@ -27,7 +27,7 @@ Gem::Specification.new do |spec|
   spec.files =
     Dir.chdir(File.expand_path('..', __FILE__)) do
       `git ls-files -z`.split("\x0").reject do |f|
-        f.match(%r{^(test|spec|features)/})
+        f.match(%r{^(bin|test|spec|features|hugo|docs_assets|log)/})
       end
     end
 

lib/event_source.rb

@@ -45,6 +45,8 @@
 require 'event_source/operations/build_message_options'
 require 'event_source/operations/build_message'
 require 'event_source/boot_registry'
+require 'event_source/paginator'
+require 'event_source/worker_pool'
 
 # Event source provides ability to compose, publish and subscribe to events
 module EventSource

lib/event_source/async_api/contracts/channel_bindings_contract.rb

@@ -0,0 +1,28 @@
+# frozen_string_literal: true
+
+require "event_source/protocols/arn/contracts/sns_channel_bindings_contract"
+
+module EventSource
+  module AsyncApi
+    module Contracts
+      # Schema and validation rules for protocol bindings to a channel
+      class ChannelBindingsContract < Contract
+        params do
+          optional(:sns).maybe(:hash)
+          optional(:amqp).maybe(:hash)
+        end
+
+        rule(:sns) do
+          if key? && value
+            validation_result = ::EventSource::Protocols::Arn::Contracts::SnsChannelBindingsContract.new.call(value)
+            if validation_result&.failure?
+              key.failure(text: 'invalid channel bindings', error: validation_result.errors.to_h)
+            else
+              values.data.merge(sns: validation_result.values.to_h)
+            end
+          end
+        end
+      end
+    end
+  end
+end

lib/event_source/async_api/contracts/channel_item_contract.rb

@@ -1,5 +1,7 @@
 # frozen_string_literal: true
 
+require_relative "channel_bindings_contract"
+
 module EventSource
   module AsyncApi
     module Contracts
@@ -21,7 +23,7 @@ class ChannelItemContract < Contract
           optional(:publish).value(:hash)
           optional(:description).value(:string)
           optional(:parameters).value(Types::HashOrNil)
-          optional(:bindings).hash { optional(:amqp).maybe(:hash) }
+          optional(:bindings).value(:hash)
         end
 
         rule(:subscribe) do
@@ -45,6 +47,17 @@ class ChannelItemContract < Contract
             end
           end
         end
+
+        rule(:bindings) do
+          if key? && value
+            validation_result = ChannelBindingsContract.new.call(value)
+            if validation_result&.failure?
+              key.failure(text: 'invalid channel bindings', error: validation_result.errors.to_h)
+            else
+              values.data.merge(bindings: validation_result.values.to_h)
+            end
+          end
+        end
       end
     end
   end

lib/event_source/logging.rb

@@ -17,21 +17,36 @@ def logger
           )
 
         # # only show "info" or higher messages on STDOUT using the Basic layout
-        ::Logging.appenders.stdout(level: :info, layout: layout)
+        stdout_logger = ::Logging.appenders.stdout(level: :info, layout: layout)
 
         Dir.mkdir('log') unless File.directory?('log')
         # # send all log events to the development log (including debug) as JSON
-        ::Logging.appenders.rolling_file(
+        rf = ::Logging.appenders.rolling_file(
           'log/event_source.log',
           age: 'daily',
           level: EventSource.config.log_level,
           keep: 7,
           layout: ::Logging.layouts.json
         )
 
-        logger_instance.add_appenders 'stdout', 'log/event_source.log'
+        logger_instance.add_appenders stdout_logger, rf
       end
       logger_instance
     end
   end
+
+  # Logger that pulls the base settings from the event source log.
+  class InheritedLogger
+    extend EventSource::Logging
+
+    def self.logger_for(klass)
+      es_base_logger = logger
+      klass_logger = ::Logging.logger[klass]
+      if klass_logger.appenders.empty?
+        klass_logger.additive = false
+        klass_logger.appenders = es_base_logger.appenders
+      end
+      klass_logger
+    end
+  end
 end

lib/event_source/paginator.rb

@@ -0,0 +1,26 @@
+# frozen_string_literal: true
+
+module EventSource
+  # Abstract paginator for protocols which support it.
+  class Paginator
+    attr_reader :state, :continuation_data
+
+    def initialize(more, state, continue_data)
+      @more = more
+      @state = state
+      @continuation_data = continue_data
+    end
+
+    def more?
+      @more
+    end
+
+    def self.continue(state, continue_data)
+      self.new(true, state, continue_data)
+    end
+
+    def self.finished
+      self.new(false, nil, nil)
+    end
+  end
+end

lib/event_source/protocols/arn/arn_channel_proxy.rb

@@ -21,11 +21,24 @@ def initialize(channel_name, channel_item, client, connection_arn)
         #                                      settings and configuration
         # @return [Object] a protocol-specific proxy for publishing
         def add_publish_operation(async_api_channel_item)
-          raise NotImplementedError, "Service not yet implemented for ARN: #{@connection_arn.service}" unless @connection_arn.sns_service?
+          if async_api_channel_item.bindings&.key?("sns")
+            sns_bindings = async_api_channel_item.bindings["sns"]
+            return SnsSdkPublisherProxy.new(client, async_api_channel_item) if sns_bindings.key?("x-aws-sdk")
+          end
+          raise NotImplementedError, "Publish Service not yet implemented for ARN: #{@connection_arn.service}" unless @connection_arn.sns_service?
 
           SnsPublisherProxy.new(client, async_api_channel_item)
         end
 
+        # Add a subscriber operation.
+        def add_subscribe_operation(async_api_channel_item)
+          if async_api_channel_item.bindings&.key?("sns")
+            sns_bindings = async_api_channel_item.bindings["sns"]
+            return SnsSdkSubscriberProxy.new(client, async_api_channel_item) if sns_bindings.key?("x-aws-sdk")
+          end
+          raise NotImplementedError, "Subscriber Service not yet implemented for ARN: #{@connection_arn.service}"
+        end
+
         # Called after a publish proxy created.  Normally does nothing.
         # It's used for example by AMQP to create any exchange to exchange
         # bindings.

lib/event_source/protocols/arn/contracts/sns_channel_bindings_contract.rb

@@ -0,0 +1,20 @@
+# frozen_string_literal: true
+
+module EventSource
+  module Protocols
+    module Arn
+      module Contracts
+        # Validate parameters for AsyncAPI SNS channel bindings.
+        class SnsChannelBindingsContract < Dry::Validation::Contract
+          params do
+            # At the channel level, we currently only support sdk bindings.
+            optional(:"x-aws-sdk").maybe(:hash) do
+              required(:sdk_action).maybe(:string, :included_in? => ["ListPhoneNumbersOptedOut"])
+              optional(:paginate).maybe(:bool)
+            end
+          end
+        end
+      end
+    end
+  end
+end

lib/event_source/protocols/arn/sns_publisher_proxy.rb

@@ -27,10 +27,28 @@ def publish(event)
             raise Errors::MissingSmsPhoneError, @publish_info.operationId unless phone
             message = symbolized_payload[:message]
             raise Errors::MissingSmsMessageError.new(@publish_info.operationId, phone) unless message
-            @client.publish(
-              phone_number: phone,
-              message: message
-            )
+            remaining_keys = symbolized_payload.keys - [:message, phone_key.to_sym]
+            extra_attributes = {}
+            remaining_keys.each do |k|
+              v = symbolized_payload[k]
+              next unless v.is_a?(String)
+              extra_attributes[k.to_s] = {
+                data_type: "String",
+                string_value: v
+              }
+            end
+            if extra_attributes.any?
+              @client.publish(
+                phone_number: phone,
+                message: message,
+                message_attributes: extra_attributes
+              )
+            else
+              @client.publish(
+                phone_number: phone,
+                message: message
+              )
+            end
           end
         end
       end

lib/event_source/protocols/arn/sns_sdk_publisher_proxy.rb

@@ -0,0 +1,40 @@
+# frozen_string_literal: true
+
+module EventSource
+  module Protocols
+    module Arn
+      # Create and manage a SNS SDK Proxy
+      class SnsSdkPublisherProxy
+        def initialize(client, async_api_channel_item)
+          @client = client
+          @channel_item = async_api_channel_item
+          @sdk_bindings = async_api_channel_item.bindings["sns"]["x-aws-sdk"]
+          @sdk_m = @sdk_bindings["sdk_action"].underscore
+          @paginate = [true, "true"].include?(@sdk_bindings["paginate"])
+        end
+
+        def publish(_event)
+          sub_klass, executable = EventSource::Protocols::Arn.sns_sdk_worker_registry.resolve_worker_for_event(@channel_item.publish.operationId)
+          worker_pool = EventSource::WorkerPool.instance
+          # rubocop:disable Lint/RescueException
+          worker_pool.post do
+            if @paginate
+              result = executable.call(nil, @client.send(@sdk_m))
+              # rubocop:disable Style/WhileUntilModifier
+              while result.more?
+                result = executable.call(result.state, @client.send(@sdk_m, result.continuation_data))
+              end
+              # rubocop:enable Style/WhileUntilModifier
+            else
+              executable.call(@client.send(@sdk_m))
+            end
+          rescue Exception => e
+            worker_logger = ::EventSource::InheritedLogger.logger_for(sub_klass)
+            worker_logger.error e
+          end
+          # rubocop:enable Lint/RescueException
+        end
+      end
+    end
+  end
+end