Parte 3 del tutorial acerca del flujo de trabajo de suscripción: implementación de las actividades

Ahora implementaremos cada una de las actividades en nuestro flujo de trabajo, empezando con una clase base que proporciona algunas características comunes para el código de actividad.

Definición de un tipo de actividad básica

Al diseñar el flujo de trabajo, identificamos las siguientes actividades:

• get_contact_activity

• subscribe_topic_activity

• wait_for_confirmation_activity

• send_result_activity

Implementaremos cada una de estas actividades ahora. Como nuestras actividades compartirán algunas características, realicemos un poco de trabajo preliminar y creemos código común que puedan compartir. Lo llamaremos BasicActivity y lo definiremos en un nuevo archivo denominado basic_activity.rb.

Al igual que con los demás archivos de código fuente, incluiremos utils.rb para obtener acceso a la función init_domain para configurar el dominio de ejemplo.

   require_relative 'utils.rb' 

A continuación, declararemos la clase de actividad básica y algunos datos comunes que serán de nuestro interés para cada actividad. Guardaremos la instancia de AWS::SimpleWorkflow::ActivityType de la actividad, el nombre y los resultados en los atributos de la clase.

   class BasicActivity

     attr_accessor :activity_type
     attr_accessor :name
     attr_accessor :results 

Estos atributos obtienen acceso a los datos de la instancia que se definen en el método initialize de la clase, que toma un nombre de actividad, una versión opcional y un mapa de opciones que se van a utilizar al registrar la actividad en 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 

Al igual que ocurre con el registro del tipo de flujo de trabajo, si ya se ha registrado un tipo de actividad, podemos recuperarla consultando la colección de activity_types del dominio. Si no se encuentra la actividad, se registrará.

También, al igual que ocurre con los tipos de flujo de trabajo, puede establecer opciones predeterminadas que se almacenen con su tipo de actividad al registrarla.

Lo último que obtiene nuestra actividad básica es una forma coherente de ejecutarla. Definiremos un método do_activity que tome una tarea de actividad. Tal como se indica, podemos usar la tarea de actividad transmitida para recibir datos a través de su atributo de instancia de input.

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

De ese modo, concluye la clase BasicActivity. Ahora la usaremos para hacer que la definición de nuestras actividades sea sencilla y coherente.

Definición de GetContactActivity

La primera actividad que se ejecuta durante una ejecución de flujo de trabajo es get_contact_activity, que recupera la información de suscripción a un tema de Amazon SNS del usuario.

Cree un nuevo archivo llamado get_contact_activity.rb y exija tanto yaml, que usaremos a fin de preparar una cadena para transferirla a Amazon SWF, como basic_activity.rb, que usaremos como base para esta clase de GetContactActivity.

   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

Como ponemos el código de registro de la actividad en BasicActivity, el método initialize para GetContactActivity es muy sencillo. Nos limitamos a llamar al constructor de clases base con el nombre de actividad, get_contact_activity. Esto es todo lo que hace falta para registrar nuestra actividad.

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

Ahora definiremos el método do_activity, que solicita el correo electrónico o el número de teléfono del usuario.

     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

Al final de do_activity, tomamos el correo electrónico y el número de teléfono recuperados del usuario, los colocamos en un mapa y, a continuación, usamos to_yaml para convertir todo el mapa a una cadena YAML. Existe un motivo importante para esto: los resultados que pase a Amazon SWF al completar una actividad deben ser datos de cadena únicamente. La capacidad de Ruby de convertir fácilmente objetos a cadenas YAML y, posteriormente, a objetos una vez más es, por suerte, adecuada para este fin.

Ese es el final de la implementación get_contact_activity. Estos datos se usarán a continuación en la implementación de subscribe_topic_activity.

Definición de SubscribeTopicActivity

Ahora profundizaremos en Amazon SNS y crearemos una actividad que utilice la información que genera get_contact_activity para suscribir al usuario a un tema de Amazon SNS.

Cree un nuevo archivo llamado subscribe_topic_activity.rb, añada los mismos requisitos que usamos para get_contact_activity, declare su clase y proporcione su método 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 

Ahora que tenemos el código para que la actividad se configure y registre, añadiremos código para crear un tema de Amazon SNS. Para ello, utilizaremos el método create_topic del objeto AWS::SNS::Client.

Añada el método create_topic a la clase, que toma un objeto de cliente de Amazon SNS transmitido.

     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 

Una vez que tengamos el nombre de recurso de Amazon (ARN) del tema, podemos usarlo con el método set_topic_attributes del cliente de Amazon SNS para establecer el DisplayName del tema, que es necesario para enviar mensajes SMS con Amazon SNS.

Por último, definiremos el método do_activity. Comenzaremos recopilando los datos que se han pasado mediante la opción input al programarse la actividad. Tal como se ha mencionado anteriormente, estos deben transferirse como cadena, que creamos mediante to_yaml. Una vez recuperados, usaremos YAML.load para convertir los datos en objetos Ruby.

Este es el principio de do_activity, donde recuperamos los datos de entrada.

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

Si no hemos recibido ninguna entrada, no hay mucho que se pueda hacer, de modo que produciremos un error en la actividad.

Sin embargo, en el supuesto caso de que todo funcione, seguiremos rellenando nuestro método do_activity, obtendremos un cliente de Amazon SNS con el AWS SDK for Ruby y lo pasaremos a nuestro método create_topic para crear el tema de 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 

Hay un par de aspectos que merece la pena señalar aquí:

• Utilizamos AWS.config.with para establecer la región de nuestro cliente de Amazon SNS. Dado que deseamos enviar mensajes SMS, usaremos la región habilitada para SMS que declaramos en utils.rb.

• Guardamos el ARN del tema en nuestro mapa activity_data. Este forma parte de los datos que se pasarán a la siguiente actividad de nuestro flujo de trabajo.

Por último, esta actividad suscribe al usuario al tema de Amazon SNS mediante los puntos de conexión transmitidos (correo electrónico y SMS). No exigimos que el usuario escriba ambos puntos de conexión, pero necesitamos al menos uno.

       # 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 toma el ARN del tema, el protocolo (que, ingeniosamente, camuflamos como clave de asignación de activity_data para el punto de conexión correspondiente).

Por último, podemos volver a empaquetar la información para la siguiente actividad en formato YAML, de modo que podamos devolverla a 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 

De ese modo se completa la implementación de subscribe_topic_activity. A continuación, definiremos wait_for_confirmation_activity.

Definición de WaitForConfirmationActivity

Una vez que un usuario se suscriba a un tema de Amazon SNS, este deberá confirmar la solicitud de suscripción. En este caso, esperaremos la confirmación del usuario por correo electrónico o mediante un mensaje SMS.

La actividad que espera a que el usuario confirme la suscripción se llama wait_for_confirmation_activity y la definiremos aquí. Para empezar, cree un nuevo archivo llamado wait_for_confirmation_activity.rb y configúrelo cuando hayamos configurado las actividades anteriores.

   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 

A continuación, empezaremos a definir el método do_activity y recuperaremos los datos de entrada de una variable local llamada 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) 

Ahora que tenemos el ARN del tema, podemos recuperar el tema creando una nueva instancia de AWS::SNS::Topic y pasarle el 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 

Ahora, comprobaremos el tema para ver si el usuario ha confirmado la suscripción mediante uno de los puntos de conexión. Solo exigiremos que se haya confirmado un punto de conexión para considerar la actividad un éxito.

Cualquier tema de Amazon SNS mantiene una lista de las suscripciones a dicho tema, y podemos verificar si el usuario ha confirmado o no una suscripción determinada al comprobar si el ARN de la suscripción está establecido en otra opción distinta a 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 

Si obtenemos un ARN para la suscripción, lo guardaremos en los datos de resultados de la actividad, lo convertiremos a YAML y devolveremos true de do_activity, que indica que la actividad se ha completado correctamente.

Como esperar la confirmación de una suscripción puede llevar su tiempo, llamaremos ocasionalmente a record_heartbeat en la tarea de actividad. Esto indica a Amazon SWF que la actividad se sigue procesando y que puede utilizarse para proporcionar actualizaciones acerca del progreso de la actividad (si realiza alguna tarea, como el procesamiento de archivos, de cuyo progreso pueda informar).

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

Esto pone fin a nuestro bucle while. Si de algún modo salimos del bucle while sin éxito, notificaremos el error y pondremos fin al método 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 

Eso pone fin a la implementación de wait_for_confirmation_activity. Solo tenemos una actividad más que definir: send_result_activity.

Definición de SendResultActivity

Si el flujo de trabajo ha progresado hasta este punto, quiere decir que hemos suscrito correctamente al usuario a un tema de Amazon SNS y que el usuario ha confirmado la suscripción.

Nuestra última actividad, send_result_activity, envía al usuario una confirmación de la suscripción correcta al tema, mediante el tema al que se ha suscrito al usuario y el punto de conexión con el que el usuario ha confirmado la suscripción.

Cree un nuevo archivo llamado send_result_activity.rb y configúrelo cuando hayamos configurado todas las actividades hasta el momento.

   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 

Nuestro método do_activity empieza también de forma similar, obteniendo los datos de entrada del flujo de trabajo, convirtiéndolos de YAML y usando después el ARN del tema para crear una instancia de 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 

Una vez que tengamos el tema, publicaremos un mensaje en él (y también lo repetiremos en la pantalla).

       @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 

Al publicar en un tema de Amazon SNS, se envía el mensaje que el usuario haya proporcionado a todos los puntos de conexión suscritos y confirmados que existan para ese tema. Así pues, si el usuario ha confirmado con ambos un correo electrónico y un número de SMS, él o ella recibirá dos mensajes de confirmación, uno en cada punto de conexión.

Pasos siguientes

De este modo se completa la implementación de send_result_activity. Ahora vinculará todas estas actividades en una aplicación de actividad que controla las tareas de actividad y puede lanzar actividades como respuesta, en Parte 4 del tutorial acerca del flujo de trabajo de suscripción: implementación del sondeador de tareas de actividades.