Support communication with AWS SNS SDK.

TreyE · TreyE · commit 6a287dcb63f0 · 2025-09-15T19:02:01.000-04:00
diff --git a/lib/event_source.rb b/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
diff --git a/lib/event_source/async_api/contracts/channel_bindings_contract.rb b/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
diff --git a/lib/event_source/async_api/contracts/channel_item_contract.rb b/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
diff --git a/lib/event_source/connection_manager.rb b/lib/event_source/connection_manager.rb
@@ -90,6 +90,10 @@ def find_connection(params)
         end
       else
         connections.detect do |connection|
+          if params[:protocol] == :arn
+            logger.info params[:subscribe_operation_name].inspect
+            logger.info connection.subscribe_operations.keys.inspect
+          end
           connection.subscribe_operation_exists?(
             params[:subscribe_operation_name]
           )
diff --git a/lib/event_source/paginator.rb b/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
diff --git a/lib/event_source/protocols/arn/arn_channel_proxy.rb b/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.
diff --git a/lib/event_source/protocols/arn/contracts/sns_channel_bindings_contract.rb b/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
diff --git a/lib/event_source/protocols/arn/sns_publisher_proxy.rb b/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
diff --git a/lib/event_source/protocols/arn/sns_sdk_publisher_proxy.rb b/lib/event_source/protocols/arn/sns_sdk_publisher_proxy.rb
@@ -0,0 +1,35 @@
+# 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)
+          executable = EventSource::Protocols::Arn.sns_sdk_worker_registry.resolve_worker_for_event(@channel_item.publish.operationId)
+          worker_pool = EventSource::WorkerPool.instance
+          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
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/lib/event_source/protocols/arn/sns_sdk_subscriber_proxy.rb b/lib/event_source/protocols/arn/sns_sdk_subscriber_proxy.rb
@@ -0,0 +1,20 @@
+# frozen_string_literal: true
+
+module EventSource
+  module Protocols
+    module Arn
+      # Create and manage an SNS SDK Proxy response handler
+      class SnsSdkSubscriberProxy
+        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"]
+        end
+
+        def subscribe(subscriber_klass, _bindings)
+          EventSource::Protocols::Arn.sns_sdk_worker_registry.register_subscriber(@channel_item.subscribe.operationId, subscriber_klass)
+        end
+      end
+    end
+  end
+end
diff --git a/lib/event_source/protocols/arn/sns_sdk_worker_registry.rb b/lib/event_source/protocols/arn/sns_sdk_worker_registry.rb
@@ -0,0 +1,35 @@
+# frozen_string_literal: true
+
+require 'concurrent/map'
+
+module EventSource
+  module Protocols
+    module Arn
+      # Create and manage a SNS SDK Proxy
+      class SnsSdkWorkerRegistry
+        def initialize
+          @subscribers = Concurrent::Map.new
+        end
+
+        def register_subscriber(operation_id, klass)
+          @subscribers[operation_id] = klass
+        end
+
+        def resolve_worker_for_event(operation_id)
+          sub_klass = subscriber_klass_for(operation_id)
+          action_name = operation_id.split(".").last
+          sub_exe_key = sub_klass.subscriber_executable_key_for("on_#{action_name}")
+          EventSource::Subscriber.executable_container[sub_exe_key]
+        end
+
+        protected
+
+        def subscriber_klass_for(operation_id)
+          sub_id = operation_id.split(".")[0..-2].join(".")
+          sub_key = EventSource::Protocols::Arn.subscribe_operation_name_for(EventSource.app_name, sub_id)
+          @subscribers[sub_key]
+        end
+      end
+    end
+  end
+end
diff --git a/lib/event_source/protocols/arn_protocol.rb b/lib/event_source/protocols/arn_protocol.rb
@@ -3,6 +3,9 @@
 require_relative "arn/arn_connection_proxy"
 require_relative "arn/arn_channel_proxy"
 require_relative "arn/sns_publisher_proxy"
+require_relative "arn/sns_sdk_publisher_proxy"
+require_relative "arn/sns_sdk_subscriber_proxy"
+require_relative "arn/sns_sdk_worker_registry"
 
 module EventSource
   module Protocols
@@ -11,6 +14,10 @@ module Protocols
     #
     # @since 0.7.0
     module Arn
+      def self.sns_sdk_worker_registry
+        @sns_sdk_worker_registry ||= SnsSdkWorkerRegistry.new
+      end
+
       def self.scheme_patterns
         [:arn]
       end
diff --git a/lib/event_source/subscriber.rb b/lib/event_source/subscriber.rb
@@ -122,18 +122,13 @@ def app_name
       end
 
       def create_subscription
-        subscribe_operation_name =
-          if protocol == :http
-            "/on#{publisher_key}"
-          else
-            "on_#{app_name}.#{publisher_key}"
-          end
+        subscribe_operation_name = ::EventSource::Protocols::Registry.subscribe_operation_name_for(protocol, app_name, publisher_key)
 
         connection_params = {
           protocol: protocol,
           subscribe_operation_name: subscribe_operation_name
         }
-        logger.debug "Subscriber#create_subscription find subscribe operation for #{connection_params}"
+        logger.info "Subscriber#create_subscription find subscribe operation for #{connection_params}"
         subscribe_operation =
           connection_manager.find_subscribe_operation(connection_params)
 
diff --git a/lib/event_source/worker_pool.rb b/lib/event_source/worker_pool.rb
@@ -0,0 +1,26 @@
+# frozen_string_literal: true
+
+module EventSource
+  # A fixed size, cached worker pool.
+  #
+  # Uses `concurrent-ruby` to manage workloads asynchronously.
+  class WorkerPool
+    include Singleton
+
+    attr_reader :pool
+
+    def initialize
+      @pool = ::Concurrent::FixedThreadPool.new(
+        5,
+        max_queue: 40,
+        fallback_policy: :caller_runs
+      )
+    end
+
+    # rubocop:disable Style/ArgumentsForwarding
+    def post(*args, &blk)
+      @pool.post(*args, &blk)
+    end
+    # rubocop:enable Style/ArgumentsForwarding
+  end
+end
diff --git a/spec/rails_app/app/event_source/events/sms_blocklist/update/update.rb b/spec/rails_app/app/event_source/events/sms_blocklist/update/update.rb
@@ -0,0 +1,11 @@
+# frozen_string_literal: true
+
+module Events
+  module SmsBlocklist
+    module Update
+      class Requested < EventSource::Event
+        publisher_path 'publishers.sms_blocklist_publisher'
+      end
+    end
+  end
+end
diff --git a/spec/rails_app/app/event_source/publishers/sms_blocklist_publisher.rb b/spec/rails_app/app/event_source/publishers/sms_blocklist_publisher.rb
@@ -0,0 +1,8 @@
+# frozen_string_literal: true
+
+module Publishers
+  class SmsBlocklistPublisher
+    include ::EventSource::Publisher[arn: 'sms_blocklist.update']
+    register_event 'requested'
+  end
+end
diff --git a/spec/rails_app/app/event_source/subscribers/sms_blocklist_subscriber.rb b/spec/rails_app/app/event_source/subscribers/sms_blocklist_subscriber.rb
@@ -0,0 +1,20 @@
+# frozen_string_literal: true
+
+module Subscribers
+  # Subscriber will receive Enterprise requests like date change
+  class SmsBlocklistSubscriber
+    include ::EventSource::Subscriber[arn: "sms_blocklist.update"]
+
+    subscribe(:on_requested) do |state, event|
+      if event.next_token
+        current_state = state || []
+        EventSource::Paginator.continue(current_state + [event.next_token], {next_token: event.next_token})
+      else
+        # rubocop:disable Style/GlobalVars
+        $a_blocklist_subscriber_testing_future.set(state)
+        # rubocop:enable Style/GlobalVars
+        EventSource::Paginator.finished
+      end
+    end
+  end
+end
diff --git a/spec/rails_app/spec/sms_blocklist_updater_spec.rb b/spec/rails_app/spec/sms_blocklist_updater_spec.rb
diff --git a/spec/rails_app/spec/sms_messages_publisher_spec.rb b/spec/rails_app/spec/sms_messages_publisher_spec.rb
diff --git a/spec/support/async_api_files/arn_sms_blocklist_request_publisher.yml b/spec/support/async_api_files/arn_sms_blocklist_request_publisher.yml
