Abonnement-Workflow, Tutorial Teil 3: Implementieren der Aktivitäten

Wir implementieren nun die einzelnen Aktivitäten in unserem Workflow. Dazu beginnen wir mit einer Basisklasse, die einige gängige Funktionen für den Aktivitätscode bereitstellt.

Definieren eines Basis-Aktivitätstyps

Beim Entwerfen des Workflows haben wir die folgenden Aktivitäten identifiziert:

• get_contact_activity

• subscribe_topic_activity

• wait_for_confirmation_activity

• send_result_activity

Jede dieser Aktivitäten werden wir nun implementieren. Da unsere Aktivitäten teilweise die gleichen Funktionen haben, leisten wir zunächst ein wenig Vorarbeit und erstellen Code, den sie gemeinsam nutzen können. Wir nennen diesen Code BasicActivity und definieren ihn in einer neuen Datei namens basic_activity.rb.

Wie bei den anderen Quelldateien fügen wir utils.rb hinzu, um auf die Funktion init_domain zuzugreifen und die Beispieldomäne einzurichten.

   require_relative 'utils.rb' 

Als Nächstes deklarieren wir die Basis-Aktivitätsklasse und einige gemeinsame Daten für die Aktivitäten. Wir speichern die Instance AWS::SimpleWorkflow::ActivityType, den Namen und die Ergebnisse der Aktivität in den Attributen der Klasse.

   class BasicActivity

     attr_accessor :activity_type
     attr_accessor :name
     attr_accessor :results 

Diese Attribute greifen auf Instanzdaten zu, die in der Klasse definiert sindinitializeMethode, die eine Aktivität annimmtNameund ein optionalesAusführungund Karte vonOptionendie bei der Registrierung der Aktivität bei Amazon SWF verwendet werden soll.

     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 

Wie bei der Registrierung des Workflow-Typs können wir einen bereits registrierten Aktivitätstyp abrufen, indem wir uns die Sammlung activity_types der Domäne anschauen. Wird die Aktivität nicht gefunden, wird sie registriert.

Ebenso wie bei den Workflow-Typen können Sie Standardoptionen festlegen, die mit Ihrem Aktivitätstyp bei dessen Registrierung gespeichert werden.

Im letzten Schritt legen wir eine konsistente Ausführungsmethode für unsere Basisaktivität fest. Wir definieren eine do_activity-Methode, die eine Aktivitätsaufgabe verwendet. Wie oben dargestellt können wir die übergebene Aktivitätsaufgabe verwenden, um Daten über ihr Instance-Attribut input zu empfangen.

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

Dadurch wird die Klasse BasicActivity zusammengefasst. Wir können sie nun verwenden, um unsere Aktivitäten auf einfache und konsistente Weise zu definieren.

Definieren von „GetContactActivity“

Die erste Aktivität, die bei einer Workflow-Ausführung ausgeführt wird, istget_contact_activity, das die Amazon-SNS-Themenabonnement-Informationen des Benutzers abruft.

Erstellen Sie eine neue Datei mit dem Namenget_contact_activity.rb, und benötigen beidesyaml, die wir verwenden, um eine Zeichenfolge für die Weitergabe an Amazon SWF vorzubereiten, undbasic_activity.rb, die wir als Grundlage dafür verwenden werdenGetContactActivity-Klasse.

   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

Da wir den Code für die Registrierung der Aktivität in BasicActivity definiert haben, ist die initialize-Methode für GetContactActivity recht einfach. Wir rufen einfach den Basisklassenkonstruktor mit dem Aktivitätsnamen get_contact_activity auf. Dies ist alles, was für die Registrierung unserer Aktivität erforderlich ist.

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

Wir definieren nun die do_activity-Methode, die zur Eingabe der E-Mail-Adresse und/oder Telefonnummer des Benutzers auffordert.

     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

Am Ende von do_activity platzieren wir die vom Benutzer erhaltene E-Mail-Adresse und Telefonnummer in eine Map und konvertieren diese mithilfe von to_yaml in eine YAML-Zeichenfolge. Dafür gibt es einen wichtigen Grund: Alle Ergebnisse, die Sie an Amazon SWF übergeben, wenn Sie eine Aktivität abschließen, müssenNur Zeichenfolgendatenaus. Die Fähigkeit von Ruby, Objekte einfach in YAML-Zeichenfolgen und dann wieder zurück in Objekte zu konvertieren, ist für diesen Zweck sehr gut geeignet.

Damit ist die Implementierung von get_contact_activity abgeschlossen. Diese Daten werden bei der subscribe_topic_activity-Implementierung als Nächstes verwendet.

Definieren von „SubscribeTopicActivity“

Wir beschäftigen uns nun mit Amazon SNS und erstellen eine Aktivität, die die von generierten Informationen verwendetget_contact_activityum den Benutzer für ein Amazon SNS SNS-Thema zu abonnieren.

Erstellen Sie eine neue Datei mit dem Namen subscribe_topic_activity.rb. Fügen Sie die gleichen Anforderungen wie bei get_contact_activity hinzu, deklarieren Sie Ihre Klasse und stellen Sie ihre initialize-Methode bereit.

   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 

Nachdem wir nun den Code für die Einrichtung und Registrierung der Aktivität erstellt haben, fügen wir etwas Code hinzu, um ein Amazon SNS SNS-Thema zu erstellen. Hierfür benutzen wir dieAWS:: SNS:: -Client-Objektecreate_topic-Methode.

Hinzufügen dercreate_topic-Methode zu Ihrer -Klasse, die ein übergebendes Amazon SNS SNS-Client-Objekt verwendet.

     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 

Sobald wir den Amazon-Ressourcennamen (ARN) des Themas haben, können wir ihn mit den Amazon SNS Themas verwendenset_topic_attributesMethode zum Festlegen des ThemasDisplayName, das für den Versand von SMS-Nachrichten mit Amazon SNS erforderlich ist.

Schließlich definieren wir die do_activity-Methode. Zunächst sammeln wir alle Daten, die bei Terminierung der Aktivität über die input-Option übergeben wurden. Wie zuvor erwähnt muss diese als Zeichenfolge, erstellt mit to_yaml, übergeben werden. Beim Abrufen verwenden wir YAML.load, um die Daten in Ruby-Objekte umzuwandeln.

Hier ist der Anfang von do_activity, in den wir die Eingabedaten abrufen.

     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)) 

Wenn wir keine Eingabedaten empfangen haben, gibt es nichts zu verarbeiten, also schlägt die Aktivität fehl.

Davon ausgehend, dass alles okay ist, füllen wir weiter unseredo_activity, holen Sie sich einen Amazon SNS SNS-Client mit demAWS SDK for Ruby, und geben Sie es an unserecreate_topic-Methode zum Erstellen des Amazon-SNS-Themas.

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

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

An dieser Stelle sei auf Folgendes hingewiesen:

• Wir benutzenAWS.config.withum die Region für unseren Amazon SNS SNS-Client festzulegen. Da wir SMS-Nachrichten versenden wollen, verwenden wir die SMS-fähige Region, die wir in utils.rb deklariert haben.

• Wir speichern den ARN des Themas in unserer Map activity_data. Dies ist der Teil der Daten, die an die nächste Aktivität in unserem Workflow übergeben werden.

Schließlich hat diese Aktivität den Benutzer für das Amazon-SNS-Thema unter Verwendung der übergebenen Endpunkte (E-Mail und SMS) für den Amazon SNS-Thema abonniert. Wir fordern nicht vom Benutzer, dass er beide Endpunkte angibt, wir benötigen aber mindestens einen.

       # 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 verwendet den ARN des Themas, das Protokoll (das für den entsprechenden Endpunkt als activity_data-Zuweisungsschlüssel dargestellt ist).

Schließlich verpacken wir die Informationen erneut im YAML-Format, damit wir sie an Amazon SWF zurücksenden können.

       # 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 

Damit ist die Implementierung von subscribe_topic_activity abgeschlossen. Als Nächstes definieren wir wait_for_confirmation_activity.

Definieren von „WaitForConfirmationActivity“

Sobald ein Benutzer für ein Amazon SNS SNS-Thema abonniert wurde, muss er die Abonnementanfrage weiterhin bestätigen. In diesem Fall warten wir, bis der Benutzer entweder per E-Mail oder SMS bestätigt.

Die Aktivität, mit der der Benutzer das Abonnement bestätigt, heißt wait_for_confirmation_activity. Sie wird hier definiert. Erstellen Sie zunächst eine neue Datei namens wait_for_confirmation_activity.rb. Richten Sie sie so wie die vorherigen Aktivitäten ein.

   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 

Als Nächstes beginnen wir mit der Definition der do_activity-Methode und rufen alle Eingabedaten in eine lokale Variable namens subscription_data ab.

     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) 

Nun, da wir über den ARN des Themas verfügen, können wir das Thema abrufen, indem wir eine neue Instance von AWS::SNS::Topic erstellen und sie an den ARN übergeben.

       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 

Als Nächstes überprüfen wir das Thema, um zu schauen, ob der Benutzer das Abonnement über einen der Endpunkte bestätigt hat. Wir verlangen nur, dass ein Endpunkt bestätigt wurde, um die Aktivität als erfolgreich zu betrachten.

Ein Amazon-SNS-Thema pflegt eine Liste derAbonnementsFür dieses Thema können wir prüfen, ob der Benutzer ein Abonnement bestätigt hat oder nicht, indem wir prüfen, ob der ARN des Abonnements auf etwas anderes festgelegt ist alsPendingConfirmationaus.

       # 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 

Wenn wir für das Abonnement einen ARN erhalten, speichern wir diesen in den Ergebnisdaten der Aktivität, konvertieren ihn in YAML und geben von do_activity „true“ zurück. Dies weist darauf hin, dass die Aktivität erfolgreich abgeschlossen wurde.

Da es eine Weile dauern kann, bis ein Abonnement bestätigt wird, rufen wir gelegentlich record_heartbeat für die Aktivitätsaufgabe auf. Dies zeigt Amazon SWF an, dass die Aktivität noch verarbeitet wird. Es kann auch verwendet werden, um Aktualisierungen zum Fortschritt der Aktivität zu vermelden (wenn Sie etwas tun, z. B. Dateien verarbeiten, für das ein Fortschritt angezeigt werden kann).

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

Damit endet unsere while-Schleife. Wenn wir irgendwie aus der While-Schleife geraten, melden wir einen Fehler und beenden die do_activity-Methode.

       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 

Damit ist die Implementierung von wait_for_confirmation_activity abgeschlossen. Wir müssen nur noch eine Aktivität definieren: send_result_activity.

Definieren von „SendResultActivity“

Wenn der Workflow bis hierhin gelangt ist, haben wir den Benutzer erfolgreich für ein Amazon SNS SNS-Thema abonniert und er hat das Abonnement bestätigt.

Unsere letzte Aktivität, send_result_activity, sendet dem Benutzer eine Bestätigung, dass er das Thema erfolgreichen abonniert hat. Dazu verwendet sie dieses Thema sowie den Endpunkt, über das der Benutzer das Abonnement bestätigt hat.

Erstellen Sie eine neue Datei namens send_result_activity.rb. Richten Sie sie so wie alle anderen bisherigen Aktivitäten ein.

   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 

Unsere do_activity-Methode beginnt ähnlich, das heißt, die Eingabedaten werden vom Workflow abgerufen, von YAML konvertiert und dann wird mithilfe des ARN des Themas eine AWS::SNS::Topic-Instance erstellt.

     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 

Wenn wir das Thema haben, veröffentlichen wir eine Nachricht dafür (und stellen sie auf dem Bildschirm dar).

       @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 

Durch Veröffentlichung in einem Amazon SNS SNS-Thema wird die Nachricht gesendet, die Sie anallesder abonnierten und bestätigten Endpunkte, die für dieses Thema existieren. Wenn der Benutzer also mit einer E-Mail-Adresse und einer Rufnummer für SMS bestätigt hat, erhält er zwei Bestätigungsmeldungen (eine an jedem Endpunkt).

Nächste Schritte

Damit ist die Implementierung von send_result_activity abgeschlossen. Nun verbinden Sie alle diese Aktivitäten in einer Aktivitätsanwendung, die die Aktivitätsaufgaben verarbeitet und als Reaktion darauf Aktivitäten starten kann (siehe Abonnement-Workflow, Anleitung Teil 4: Implementieren des Pollers Aktivitätsaufgaben).