サブスクリプションワークフローのチュートリアルのパート 3: アクティビティの実装

アクティビティをワークフローに実装します。まず、アクティビティコードの一般的な機能を提供する基本クラスから開始します。

基本アクティビティタイプの定義

ワークフローの設計時に確認したアクティビティは以下のとおりです。

• get_contact_activity

• subscribe_topic_activity

• wait_for_confirmation_activity

• send_result_activity

ここで、以上の各アクティビティを実装します。アクティビティ間で共有される機能があるため、準備作業として、共通コードを作成して共有できるようにします。これを BasicActivity と呼び、basic_activity.rb という新しいファイルに定義します。

他のソースファイルのように、utils.rb を含めて init_domain 関数にアクセスし、サンプルドメインをセットアップします。

   require_relative 'utils.rb' 

次に、基本的なアクティビティクラスと、各アクティビティで関心があるいくつかの一般的なデータを宣言します。クラスの属性に、アクティビティの AWS::SimpleWorkflow::ActivityType インスタンス、名前、および結果を保存します。

   class BasicActivity

     attr_accessor :activity_type
     attr_accessor :name
     attr_accessor :results 

これらの属性は、アクティビティ名を取得するクラス initialize メソッドで定義されたインスタンスデータ、およびアクティビティを Amazon SWF に登録するときに使用されるオプションのバージョンとマップにアクセスします。

     def initialize(name, version = 'v1', options = nil)

       @activity_type = nil
       @name = name
       @results = nil

       # get the domain to use for activity tasks.
       @domain = init_domain

       # Check to see if this activity type already exists.
       @domain.activity_types.each do | a |
         if (a.name == @name) && (a.version == version)
           @activity_type = a
         end
       end

       if @activity_type.nil?
         # If no options were specified, use some reasonable defaults.
         if options.nil?
           options = {
             # All timeouts are in seconds.
             :default_task_heartbeat_timeout => 900,
             :default_task_schedule_to_start_timeout => 120,
             :default_task_schedule_to_close_timeout => 3800,
             :default_task_start_to_close_timeout => 3600 }
         end
         @activity_type = @domain.activity_types.register(@name, version, options)
       end
     end 

ワークフロータイプの登録と同様に、アクティビティタイプが既に登録されている場合、ドメインの activity_types コレクションを表示してそれを取得できます。アクティビティが見つからない場合は、登録されます。

また、ワークフロータイプと同様に、登録時にアクティビティタイプとともに保存されるデフォルトオプションを設定できます。

基本アクティビティで最後に取得するのは、それを実行するための一貫した方法です。アクティビティタスクを受け取る do_activity メソッドを定義します。示されているように、渡されたアクティビティタスクを使用して、input インスタンス属性経由でデータを受け取ることができます。

     def do_activity(task)
       @results = task.input # may be nil
       return true
     end
   end 

これで BasicActivity クラスの説明を終わります。これを使用して、アクティビティの定義を簡単で一貫性のあるものにできます。

GetContactActivity の定義

ワークフロー実行で実行される最初のアクティビティは get_contact_activity です。このアクティビティでは、ユーザーの Amazon SNS トピックのサブスクリプション情報を取得します。

get_contact_activity.rb という名前の新しいファイルを作成し、Amazon SWF に渡す文字列を準備するために使用する yaml と、この GetContactActivity クラスの基礎として使用する basic_activity.rb の両方を必要とします。

   require 'yaml'
   require_relative 'basic_activity.rb'

   # **GetContactActivity** provides a prompt for the user to enter contact
   # information. When the user successfully enters contact information, the
   # activity is complete.
   class GetContactActivity < BasicActivity

アクティビティ登録コードを入れてベーシックアクティビティとすると、GetContactActivity の initialize メソッドは非常に単純です。基本クラスのコンストラクタを単純にアクティビティ名 get_contact_activity で呼びます。これで、アクティビティを登録するために必要なすべての操作を実行しました。

     # initialize the activity
     def initialize
       super('get_contact_activity')
     end 

次に、do_activity メソッドを定義します。このメソッドはユーザーの E メール、電話番号、またはその両方の入力を求めます。

     def do_activity(task)
       puts ""
       puts "Please enter either an email address or SMS message (mobile phone) number to"
       puts "receive SNS notifications. You can also enter both to use both address types."
       puts ""
       puts "If you enter a phone number, it must be able to receive SMS messages, and must"
       puts "be 11 digits (such as 12065550101 to represent the number 1-206-555-0101)."

       input_confirmed = false
       while !input_confirmed
         puts ""
         print "Email: "
         email = $stdin.gets.strip

         print "Phone: "
         phone = $stdin.gets.strip

         puts ""
         if (email == '') && (phone == '')
           print "You provided no subscription information. Quit? (y/n)"
            confirmation = $stdin.gets.strip.downcase
            if confirmation == 'y'
              return false
            end
         else
            puts "You entered:"
            puts "  email: #{email}"
            puts "  phone: #{phone}"
            print "\nIs this correct? (y/n): "
            confirmation = $stdin.gets.strip.downcase
            if confirmation == 'y'
              input_confirmed = true
            end
         end
       end

       # make sure that @results is a single string. YAML makes this easy.
       @results = { :email => email, :sms => phone }.to_yaml
       return true
     end
   end

do_activity の最後に、ユーザーから取得した E メールと電話番号を使用し、それをマップに置き、to_yaml を使用してマップ全体を YAML 文字列に変換します。これには重要な理由があります。アクティビティの完了時に Amazon SWF に渡す結果は、文字列データのみ である必要があります。オブジェクトを簡単に YAML 文字列に変換し、再びオブジェクトに戻す Ruby の機能は、この目的に最適です。

これで get_contact_activity の実装が終わりました。このデータは、次に subscribe_topic_activity の実装で使用します。

SubscribeTopicActivity の定義

Amazon SNS の詳細を説明し、ユーザーを Amazon SNS トピックにサブスクライブするために get_contact_activity によって生成された情報を使用するアクティビティを作成します。

subscribe_topic_activity.rb という名前の新しいファイルを作成します。get_contact_activity に使用したのと同じ要件を追加し、クラスを宣言して、その initialize メソッドを提供します。

   require 'yaml'
   require_relative 'basic_activity.rb'

   # **SubscribeTopicActivity** sends an SMS / email message to the user, asking for
   # confirmation.  When this action has been taken, the activity is complete.
   class SubscribeTopicActivity < BasicActivity

     def initialize
       super('subscribe_topic_activity')
     end 

これでアクティビティをセットアップおよび登録するためにコードを配置したので、コードを追加して Amazon SNS トピックを作成します。そのためには、AWS::SNS::Client オブジェクトの create_topic メソッドを使用します。

create_topic メソッドをクラスに追加します。このメソッドは渡された Amazon SNS クライアントオブジェクトを受け取ります。

     def create_topic(sns_client)
       topic_arn = sns_client.create_topic(:name => 'SWF_Sample_Topic')[:topic_arn]

       if topic_arn != nil
         # For an SMS notification, setting `DisplayName` is *required*. Note that
         # only the *first 10 characters* of the DisplayName will be shown on the
         # SMS message sent to the user, so choose your DisplayName wisely!
         sns_client.set_topic_attributes( {
           :topic_arn => topic_arn,
           :attribute_name => 'DisplayName',
           :attribute_value => 'SWFSample' } )
       else
         @results = {
           :reason => "Couldn't create SNS topic", :detail => "" }.to_yaml
         return nil
       end

       return topic_arn
     end 

トピックの Amazon リソースネーム (ARN) を設定したら、これを Amazon SNS クライアントの set_topic_attributes メソッドで使用して、トピックの DisplayName を設定します。これは、Amazon SNS で SMS メッセージを送信するために必要です。

最後に、do_activity メソッドを定義します。アクティビティがスケジュールされたときに input オプション経由で渡されたデータの収集から開始します。前に説明したように、これは文字列として渡す必要があります。この文字列は「to_yaml」を使用して作成しました。文字列を取得するときは、YAML.load を使用してデータを Ruby オブジェクトに変換します。

これは do_activity の開始であり、ここでは入力データを取得します。

     def do_activity(task)
       activity_data = {
         :topic_arn => nil,
         :email => { :endpoint => nil, :subscription_arn => nil },
         :sms => { :endpoint => nil, :subscription_arn => nil },
       }

       if task.input != nil
         input = YAML.load(task.input)
         activity_data[:email][:endpoint] = input[:email]
         activity_data[:sms][:endpoint] = input[:sms]
       else
         @results = { :reason => "Didn't receive any input!", :detail => "" }.to_yaml
         puts("  #{@results.inspect}")
         return false
       end

       # Create an SNS client. This is used to interact with the service. Set the
       # region to $SMS_REGION, which is a region that supports SMS notifications
       # (defined in the file `utils.rb`).
       sns_client = AWS::SNS::Client.new(
         :config => AWS.config.with(:region => $SMS_REGION)) 

入力がなかった場合、多くの操作はないため、アクティビティが失敗します。

ただし、すべてが正しい場合、do_activity メソッドに入力し、AWS SDK for Ruby で Amazon SNS クライアントを取得して、それを create_topic メソッドに渡して Amazon SNS トピックを作成します。

       # Create the topic and get the ARN
       activity_data[:topic_arn] = create_topic(sns_client)

       if activity_data[:topic_arn].nil?
         return false
       end 

ここでいくつか注意が必要なことがあります。

• AWS.config.with を使用して、Amazon SNS クライアントのリージョンを設定します。SMS メッセージを送信するため、utils.rb で宣言した SMS 対応リージョンを使用します。

• トピックの ARN を activity_data マップに保存します。これはワークフローの次のアクティビティに渡されるデータの部分です。

最後に、このアクティビティは、渡されたエンドポイント (E メールと SMS) を使用してユーザーを Amazon SNS トピックにサブスクライブします。ユーザーが両方のエンドポイントを入力する必要はありませんが、少なくとも 1 つが必要です。

       # Subscribe the user to the topic, using either or both endpoints.
       [:email, :sms].each do | x |
         ep = activity_data[x][:endpoint]
         # don't try to subscribe an empty endpoint
         if (ep != nil && ep != "")
           response = sns_client.subscribe( {
             :topic_arn => activity_data[:topic_arn],
             :protocol => x.to_s, :endpoint => ep } )
           activity_data[x][:subscription_arn] = response[:subscription_arn]
         end
       end 

AWS::SNS::Client.subscribe はトピック ARN、プロトコル (これは明らかに対応エンドポイントの activity_data マップキーとして見せかけたものです)。

最後に、YAML 形式で次のアクティビティの情報を再パッケージし、Amazon SWF に送信できるようにします。

       # if at least one subscription arn is set, consider this a success.
       if (activity_data[:email][:subscription_arn] != nil) or (activity_data[:sms][:subscription_arn] != nil)
         @results = activity_data.to_yaml
       else
         @results = { :reason => "Couldn't subscribe to SNS topic", :detail => "" }.to_yaml
         puts("  #{@results.inspect}")
         return false
       end
       return true
     end
   end 

これで subscribe_topic_activity の実装が完了します。次に、wait_for_confirmation_activity を定義します。

WaitForConfirmationActivity の定義

ユーザーは、Amazon SNS トピックに登録したら、サブスクリプションリクエストを確認する必要があります。この場合は、ユーザーが E メールまたは SMS メッセージによってサブスクリプションを確認するのを待ちます。

ユーザーによるサブスクリプションの確認を待機するアクティビティは wait_for_confirmation_activity と呼ばれ、ここで定義します。開始するには、wait_for_confirmation_activity.rb と呼ばれる新しいファイルを作成し、前のアクティビティをセットアップしたようにセットアップします。

   require 'yaml'
   require_relative 'basic_activity.rb'

   # **WaitForConfirmationActivity** waits for the user to confirm the SNS
   # subscription.  When this action has been taken, the activity is complete. It
   # might also time out...
   class WaitForConfirmationActivity < BasicActivity

     # Initialize the class
     def initialize
       super('wait_for_confirmation_activity')
     end 

次に、do_activity メソッドの定義を開始し、subscription_data というローカル変数に入力データを取得します。

     def do_activity(task)
       if task.input.nil?
         @results = { :reason => "Didn't receive any input!", :detail => "" }.to_yaml
         return false
       end

       subscription_data = YAML.load(task.input) 

これでトピック ARN を用意できたので、AWS::SNS::Topic の新しいインスタンスを作成し、これを ARN に渡します。

       topic = AWS::SNS::Topic.new(subscription_data[:topic_arn])

       if topic.nil?
         @results = {
           :reason => "Couldn't get SWF topic ARN",
           :detail => "Topic ARN: #{topic.arn}" }.to_yaml
         return false
       end 

ここで、トピックをチェックし、ユーザーがいずれかのエンドポイントを使用して、サブスクリプションを確認したかどうか調べます。アクティビティを成功と見なすためには、1 つのエンドポイントの確認のみが必要です。

Amazon SNS トピックはそのトピックのサブスクリプションのリストを保持します。サブスクリプションの ARN が PendingConfirmation 以外に設定されているかどうか確認して、ユーザーが特定のサブスクリプションを確認したかどうかをチェックできます。

       # loop until we get some indication that a subscription was confirmed.
       subscription_confirmed = false
       while(!subscription_confirmed)
         topic.subscriptions.each do | sub |
           if subscription_data[sub.protocol.to_sym][:endpoint] == sub.endpoint
             # this is one of the endpoints we're interested in. Is it subscribed?
             if sub.arn != 'PendingConfirmation'
               subscription_data[sub.protocol.to_sym][:subscription_arn] = sub.arn
               puts "Topic subscription confirmed for (#{sub.protocol}: #{sub.endpoint})"
               @results = subscription_data.to_yaml
               return true
             else
               puts "Topic subscription still pending for (#{sub.protocol}: #{sub.endpoint})"
             end
           end
         end 

サブスクリプションの ARN を取得する場合、これをアクティビティの結果データに保存し、YAML に変換して、do_activity から true を返します。これにより、アクティビティが正常に完了したことを伝えます。

サブスクリプションの確認の待機にはしばらく時間がかかるため、アクティビティタスクで record_heartbeat をときどき呼び出します。これにより、アクティビティがまだ処理中であることが Amazon SWF に伝えられ、これを使用してアクティビティの進行状況の更新を提供できます (ファイルの処理など何かを実行中は、その進捗状況を報告できます)。

         task.record_heartbeat!(
           { :details => "#{topic.num_subscriptions_confirmed} confirmed, #{topic.num_subscriptions_pending} pending" })
         # sleep a bit.
         sleep(4.0)
       end 

これで while ループを終了します。成功せずにループを終了した場合、失敗を報告して do_activity メソッドを終了します。

       if (subscription_confirmed == false)
         @results = {
           :reason => "No subscriptions could be confirmed",
           :detail => "#{topic.num_subscriptions_confirmed} confirmed, #{topic.num_subscriptions_pending} pending" }.to_yaml
         return false
       end
     end
   end 

これで wait_for_confirmation_activity の実装が終わりました。定義するアクティビティは 後 1 つのみ (send_result_activity) です。

SendResultActivity の定義

ワークフローがここまで進んだ場合、ユーザーを Amazon SNS トピックに正しくサブスクライブし、ユーザーはサブスクリプションを確認しました。

最後のアクティビティ send_result_activity では、ユーザーがサブスクライブしたトピックと、ユーザーがサブスクリプションを確認したエンドポイントを使用して、ユーザーに成功したトピックサブスクリプションの確認を送信します。

send_result_activity.rb と呼ばれる新しいファイルを作成し、これまでのすべてのアクティビティをセットアップしたようにセットアップします。

   require 'yaml'
   require_relative 'basic_activity.rb'

   # **SendResultActivity** sends the result of the activity to the screen, and, if
   # the user successfully registered using SNS, to the user using the SNS contact
   # information collected.
   class SendResultActivity < BasicActivity

     def initialize
       super('send_result_activity')
     end 

do_activity メソッドは同様に開始し、ワークフローから入力データを取得し、YAML から変換して、トピック ARN を使用して AWS::SNS::Topic インスタンスを作成します。

     def do_activity(task)
       if task.input.nil?
         @results = { :reason => "Didn't receive any input!", :detail => "" }
         return false
       end

       input = YAML.load(task.input)

       # get the topic, so we publish a message to it.
       topic = AWS::SNS::Topic.new(input[:topic_arn])

       if topic.nil?
         @results = {
           :reason => "Couldn't get SWF topic",
           :detail => "Topic ARN: #{topic.arn}" }
         return false
       end 

トピックの準備ができたら、そのトピックにメッセージを発行します (また、画面にエコーします)。

       @results = "Thanks, you've successfully confirmed registration, and your workflow is complete!"

       # send the message via SNS, and also print it on the screen.
       topic.publish(@results)
       puts(@results)

       return true
     end
   end 

Amazon SNS トピックへの発行では、そのトピックに対して存在するサブスクライブおよび確認済みのすべてのエンドポイントに提供するメッセージを送信します。したがって、ユーザーが E メールと SMS 番号の両方で確認された場合、ユーザーはエンドポイントごとに 2 つの確認メッセージを受信します。

次のステップ

これで send_result_activity の実装が完了します。ここでアクティビティタスクを処理し、サブスクリプションワークフローのチュートリアルのパート 4: アクティビティタスクポーラーの実装 レスポンスでアクティビティタスクを起動できるアクティビティアプリケーションで、これらのアクティビティをまとめます。