Tutorial Alur Kerja Langganan Bagian 2: Menerapkan Alur Kerja

Sampai sekarang, kode kami masih cukup generik. Ini adalah bagian di mana kita mulai benar-benar menentukan apa yang alur kerja kita lakukan, dan aktivitas apa yang kita perlukan untuk menerapkannya.

Merancang Alur Kerja

Jika Anda mengingat kembali, gagasan awal untuk alur kerja ini terdiri dari langkah-langkah berikut:

1. Dapatkan alamat berlangganan (email atau SMS) dari pengguna.

2. Buat topik SNS dan berlangganan titik akhir yang disediakan untuk topik tersebut.

3. Tunggu pengguna mengonfirmasi langganan.

4. Jika pengguna mengonfirmasi, publikasikan pesan ucapan selamat ke topik tersebut.

Kita dapat menganggap setiap langkah dalam alur kerja kita sebagai aktivitas yang harus dijalankan oleh alur kerja. Alur Kerja kita bertanggung jawab untuk menjadwalkan setiap aktivitas pada waktu yang tepat, dan mengoordinasikan transfer data antar aktivitas.

Untuk alur kerja ini, kita akan membuat aktivitas terpisah untuk setiap langkah berikut, menamainya secara deskriptif:

1. get_contact_activity

2. subscribe_topic_activity

3. wait_for_confirmation_activity

4. send_result_activity

Aktivitas ini akan dijalankan secara berurutan, dan data dari setiap langkah akan digunakan pada langkah berikutnya.

Kita bisa merancang aplikasi kita sehingga semua kode berada dalam satu file sumber, tapi ini berjalan bertentangan dengan cara yang dirancang Amazon SWF. Cara ini dirancang untuk alur kerja yang dapat menjangkau seluruh Internet dalam ruang lingkup, jadi mari kita setidaknya membagi aplikasi menjadi dua executable terpisah:

• swf_sns_workflow.rb - Berisi alur kerja dan starter alur kerja.

• swf_sns_activities.rb - Berisi aktivitas dan starter aktivitas.

Penerapan alur kerja dan aktivitas dapat dijalankan di jendela terpisah, komputer terpisah, atau bahkan bagian dunia yang berbeda. Karena Amazon SWF melacak detail alur kerja dan aktivitas Anda, alur kerja Anda dapat mengkoordinasikan penjadwalan dan transfer data aktivitas Anda di mana pun mereka berjalan.

Menyiapkan Kode Alur Kerja

Kita akan mulai dengan membuat sebuah file bernama swf_sns_workflow.rb. Dalam file ini, nyatakan kelas yang disebut SampleWorkflow. Berikut adalah deklarasi kelas dan konstruktornya, metode initialize.

require_relative 'utils.rb'

# SampleWorkflow - the main workflow for the SWF/SNS Sample
#
# See the file called `README.md` for a description of what this file does.
class SampleWorkflow

  attr_accessor :name

  def initialize(workflowId)

    # the domain to look for decision tasks in.
    @domain = init_domain

    # the task list is used to poll for decision tasks.
    @workflowId = workflowId

    # The list of activities to run, in order. These name/version hashes can be
    # passed directly to AWS::SimpleWorkflow::DecisionTask#schedule_activity_task.
    @activity_list = [
      { :name => 'get_contact_activity', :version => 'v1' },
      { :name => 'subscribe_topic_activity', :version => 'v1' },
      { :name => 'wait_for_confirmation_activity', :version => 'v1' },
      { :name => 'send_result_activity', :version => 'v1' },
    ].reverse! # reverse the order... we're treating this like a stack.

    register_workflow
  end

Seperti yang Anda lihat, kita menyimpan data instans kelas berikut:

• domain - Nama domain yang diambil dari init_domain di utils.rb.

• workflowId - Daftar tugas diteruskan ke initialize.

• activity_list - Daftar aktivitas, yang memiliki nama dan versi dari aktivitas yang akan kita jalankan.

Nama domain, nama aktivitas, dan versi aktivitas sudah mencukupi bagi Amazon SWF untuk mengidentifikasi tipe aktivitas secara positif, jadi itu semua merupakan data tentang aktivitas kita yang perlu disimpan untuk menjadwalkan aktivitas.

Daftar tugas akan digunakan oleh kode decider alur kerja untuk melakukan polling tugas keputusan dan aktivitas jadwal.

Pada akhir fungsi ini, kita memanggil metode yang belum kita tentukan: register_workflow. Kita akan menentukan metode ini selanjutnya.

Mendaftarkan Alur Kerja

Untuk menggunakan tipe alur kerja, pertama-tama kita harus mendaftarkannya. Seperti tipe aktivitas, tipe alur kerja dapat diidentifikasi berdasarkan domain, nama, dan versinya. Selain itu, seperti domain dan tipe aktivitas, Anda tidak dapat mendaftarkan ulang tipe alur kerja yang ada. Jika Anda perlu mengubah apa pun tentang tipe alur kerja, Anda harus menyediakannya dengan versi baru, yang pada dasarnya menciptakan tipe baru.

Berikut adalah kode untuk register_workflow, yang digunakan untuk mengambil tipe alur kerja yang ada yang telah kita daftarkan pada eksekusi sebelumnya atau untuk mendaftarkan alur kerja jika belum terdaftar.

  # Registers the workflow
  def register_workflow
    workflow_name = 'swf-sns-workflow'
    @workflow_type = nil

    # a default value...
    workflow_version = '1'

    # Check to see if this workflow type already exists. If so, use it.
    @domain.workflow_types.each do | a |
      if (a.name == workflow_name) && (a.version == workflow_version)
        @workflow_type = a
      end
    end

    if @workflow_type.nil?
      options =  {
        :default_child_policy => :terminate,
        :default_task_start_to_close_timeout => 3600,
        :default_execution_start_to_close_timeout => 24 * 3600 }

      puts "registering workflow: #{workflow_name}, #{workflow_version}, #{options.inspect}"
      @workflow_type = @domain.workflow_types.register(workflow_name, workflow_version, options)
    end

    puts "** registered workflow: #{workflow_name}"
  end 

Pertama, kita memeriksa untuk melihat apakah nama dan versi alur kerja sudah terdaftar dengan mengiterasi melalui koleksi workflow_types domain. Jika kita menemukan kecocokan, kita akan menggunakan tipe alur kerja yang sudah terdaftar.

Jika kita tidak menemukan kecocokan, maka tipe alur kerja baru didaftarkan (dengan memanggil register (daftarkan) pada koleksi workflow_types yang sama tempat kita mencari alur kerja) dengan nama 'swf-sns-workflow', versi '1', dan opsi berikut.

      options =  {
        :default_child_policy => :terminate,
        :default_task_start_to_close_timeout => 3600,
        :default_execution_start_to_close_timeout => 24 * 3600 } 

Opsi yang diteruskan selama pendaftaran digunakan untuk mengatur perilaku default untuk tipe alur kerja kita, jadi kita tidak perlu mengatur nilai-nilai ini setiap kali kita mulai mengeksekusi alur kerja baru.

Di sini, kita hanya menetapkan beberapa nilai batas waktu: waktu maksimum yang dapat diambil dari saat tugas mulai hingga menutup (satu jam), dan waktu maksimum yang dapat digunakan untuk menyelesaikan eksekusi alur kerja (24 jam). Jika salah satu dari waktu tersebut terlampaui, tugas atau alur kerja akan mencapai batas waktu.

Untuk informasi lebih lanjut tentang nilai-nilai batas waktu tersebut, lihat Tipe Batas Waktu Amazon SWF .

Polling untuk Keputusan

Pada inti dari setiap eksekusi alur kerja terdapat decider. Tanggung jawab decider adalah untuk mengelola eksekusi alur kerja itu sendiri. Decider menerima tugas keputusan dan meresponsnya, baik dengan menjadwalkan aktivitas baru, membatalkan dan memulai ulang aktivitas, atau dengan menetapkan status eksekusi alur kerja sebagai selesai, dibatalkan, atau gagal.

Decider menggunakan nama daftar tugas eksekusi alur kerja untuk menerima tugas keputusan untuk direspons. Untuk melakukan polling untuk tugas keputusan, panggil poll pada koleksi decision_tasks domain untuk melakukan loop pada tugas keputusan yang tersedia. Anda kemudian dapat memeriksa kejadian baru dalam tugas keputusan dengan mengiterasi pada koleksi new_events.

Peristiwa yang dikembalikan adalah objek AWS::SimpleWorkflow::HistoryEvent, dan Anda bisa mendapatkan tipe kejadian dengan menggunakan anggota event_type yang dikembalikan. Untuk daftar dan keterangan tipe kejadian riwayat, lihat HistoryEvent dalam Referensi API Amazon Simple Workflow Service.

Berikut adalah awal dari logika poller tugas keputusan ini. Sebuah metode baru di kelas alur kerja kita disebut sebagai poll_for_decisions.

  def poll_for_decisions
    # first, poll for decision tasks...
    @domain.decision_tasks.poll(@workflowId) do | task |
      task.new_events.each do | event |
        case event.event_type 

Sekarang kita akan membagi eksekusi decider kita berdasarkan event_type yang diterima. Yang pertama yang mungkin akan kita terima adalah WorkflowExecutionStarted. Ketika kejadian ini diterima, artinya Amazon SWF memberi sinyal untuk decider Anda bahwa decider harus memulai eksekusi alur kerja. Kita akan mulai dengan menjadwalkan aktivitas pertama dengan memanggil schedule_activity_task pada tugas yang kita terima saat melakukan polling.

Kita akan meneruskan aktivitas pertama yang kita nyatakan dalam daftar aktivitas kita, yang, karena kita membalik daftar sehingga kita dapat menggunakannya seperti tumpukan, menempati posisi last pada daftar. “Aktivitas” yang kita tentukan hanyalah berupa peta yang terdiri dari nama dan nomor versi, tapi inilah yang dibutuhkan Amazon SWF untuk mengidentifikasi aktivitas penjadwalan, dengan asumsi bahwa aktivitas tersebut telah terdaftar.

          when 'WorkflowExecutionStarted'
            # schedule the last activity on the (reversed, remember?) list to
            # begin the workflow.
            puts "** scheduling activity task: #{@activity_list.last[:name]}"

            task.schedule_activity_task( @activity_list.last,
              { :workflowId => "#{@workflowId}-activities" } ) 

Ketika kita menjadwalkan suatu aktivitas, Amazon SWF mengirimkan tugas aktivitas ke daftar tugas aktivitas yang kita teruskan saat menjadwalkannya, memberi sinyal pada tugas untuk mulai. Kita akan menangani tugas aktivitas di Tutorial Alur Kerja Bagian 3: Menerapkan Aktivitas, tetapi perlu dicatat bahwa kita tidak mengeksekusi tugas di sini. Kita hanya memberitahu Amazon SWF bahwa tugas perlu dijadwalkan.

Aktivitas berikutnya yang perlu kita tangani adalah kejadian ActivityTaskCompleted, yang terjadi ketika Amazon SWF telah menerima respon bahwa aktivitas selesai dari tugas aktivitas.

          when 'ActivityTaskCompleted'
            # we are running the activities in strict sequential order, and
            # using the results of the previous activity as input for the next
            # activity.
            last_activity = @activity_list.pop

            if(@activity_list.empty?)
              puts "!! All activities complete! Sending complete_workflow_execution..."
              task.complete_workflow_execution
              return true;
            else
              # schedule the next activity, passing any results from the
              # previous activity. Results will be received in the activity
              # task.
              puts "** scheduling activity task: #{@activity_list.last[:name]}"
              if event.attributes.has_key?('result')
                task.schedule_activity_task(
                  @activity_list.last,
                  { :input => event.attributes[:result],
                    :workflowId => "#{@workflowId}-activities" } )
              else
                task.schedule_activity_task(
                  @activity_list.last, { :workflowId => "#{@workflowId}-activities" } )
              end
            end 

Karena kita mengeksekusi tugas kita secara linear, dan hanya ada satu aktivitas yang dieksekusi pada satu waktu, kita akan mengambil kesempatan ini untuk memunculkan tugas yang telah diselesaikan dari tumpukan activity_list. Jika hal ini menghasilkan daftar kosong, maka kita tahu bahwa alur kerja kita sudah selesai. Dalam hal ini, kita mengirim sinyal kepada Amazon SWF bahwa alur kerja kita sudah selesai dengan memanggil complete_workflow_execution pada tugas.

Jika daftar masih memiliki entri, kita akan menjadwalkan aktivitas berikutnya dalam daftar (sekali lagi, di posisi terakhir). Namun, kali ini, kita akan melihat untuk memeriksa apakah aktivitas sebelumnya mengembalikan data hasil apa pun ke Amazon SWF setelah diselesaikan, yang disediakan untuk alur kerja dalam atribut kejadian, di kunci result. Jika aktivitas menghasilkan hasil, kita akan meneruskannya sebagai opsi input ke aktivitas terjadwal berikutnya bersama dengan daftar tugas aktivitas.

Dengan mengambil nilai-nilai result aktivitas yang sudah selesai, dan dengan menetapkan nilai-nilai input aktivitas yang dijadwalkan, kita dapat meneruskan data dari satu aktivitas ke aktivitas berikutnya, atau kita dapat menggunakan data dari suatu aktivitas untuk mengubah perilaku dalam decider kita berdasarkan hasil dari suatu aktivitas.

Untuk tujuan tutorial ini, dua tipe kejadian ini adalah yang paling penting dalam menentukan perilaku alur kerja kita. Namun, suatu aktivitas dapat menghasilkan kejadian selain ActivityTaskCompleted. Kita akan menyelesaikan kode decider kita dengan menyediakan kode pengendali demonstrasi untuk kejadian ActivityTaskTimedOut dan ActivityTaskFailed, dan untuk kejadian WorkflowExecutionCompleted, yang akan dihasilkan ketika Amazon SWF memproses panggilan complete_workflow_execution yang kita buat ketika kita kehabisan kegiatan untuk dijalankan.

          when 'ActivityTaskTimedOut'
            puts "!! Failing workflow execution! (timed out activity)"
            task.fail_workflow_execution
            return false

          when 'ActivityTaskFailed'
            puts "!! Failing workflow execution! (failed activity)"
            task.fail_workflow_execution
            return false

          when 'WorkflowExecutionCompleted'
            puts "## Yesss, workflow execution completed!"
            task.workflow_execution.terminate
            return false
        end
      end
    end
  end 

Memulai Eksekusi Alur Kerja

Sebelum tugas keputusan akan dibuat untuk alur kerja untuk di-polling, kita perlu memulai eksekusi alur kerja.

Untuk memulai eksekusi alur kerja, panggil start_execution pada tipe alur kerja terdaftar Anda (AWS::SimpleWorkflow::WorkflowType). Kita akan menentukan pembungkus kecil di sekitarnya untuk memanfaatkan anggota instans workflow_type yang telah kita dapatkan dalam konstruktor kelas.

  def start_execution
    workflow_execution = @workflow_type.start_execution( {
      :workflowId => @workflowId } )
    poll_for_decisions
  end
end 

Setelah alur kerja dieksekusi, kejadian keputusan akan mulai muncul di daftar tugas alur kerja, yang diteruskan sebagai opsi eksekusi alur kerja di start_execution.

Tidak seperti opsi yang disediakan ketika tipe alur kerja didaftarkan, opsi yang diteruskan ke start_execution tidak dianggap sebagai bagian dari tipe alur kerja. Anda bebas untuk mengubahnya per eksekusi alur kerja tanpa mengubah versi alur kerja.

Karena kita menginginkan alur kerja untuk mulai mengeksekusi ketika kita menjalankan file, tambahkan beberapa kode yang menginstansiasi kelas dan kemudian panggil metode start_execution yang baru saja kita tentukan.

if __FILE__ == $0
  require 'securerandom'

  # Use a different task list name every time we start a new workflow execution.
  #
  # This avoids issues if our pollers re-start before SWF considers them closed,
  # causing the pollers to get events from previously-run executions.
  workflowId = SecureRandom.uuid

  # Let the user start the activity worker first...

  puts ""
  puts "Amazon SWF Example"
  puts "------------------"
  puts ""
  puts "Start the activity worker, preferably in a separate command-line window, with"
  puts "the following command:"
  puts ""
  puts "> ruby swf_sns_activities.rb #{workflowId}-activities"
  puts ""
  puts "You can copy & paste it if you like, just don't copy the '>' character."
  puts ""
  puts "Press return when you're ready..."

  i = gets

  # Now, start the workflow.

  puts "Starting workflow execution."
  sample_workflow = SampleWorkflow.new(workflowId)
  sample_workflow.start_execution
end 

Untuk menghindari konflik penamaan daftar tugas, kita akan menggunakan SecureRandom.uuid untuk menghasilkan UUID acak yang dapat kita gunakan sebagai nama daftar tugas, yang menjamin bahwa nama daftar tugas yang berbeda digunakan untuk setiap eksekusi alur kerja.

catatan

Daftar tugas digunakan untuk merekam kejadian tentang eksekusi alur kerja, jadi jika Anda menggunakan daftar tugas yang sama untuk beberapa eksekusi dari tipe alur kerja yang sama, Anda mungkin mendapatkan kejadian yang dihasilkan selama eksekusi sebelumnya, terutama jika Anda menjalankannya dalam suksesi yang dekat satu sama lain, yang sering terjadi ketika mencoba kode baru atau menjalankan tes.

Agar terhindar dari keharusan untuk berurusan dengan artefak dari eksekusi sebelumnya, kita dapat menggunakan daftar tugas baru untuk setiap eksekusi, menentukannya ketika kita memulai eksekusi alur kerja.

Terdapat pula sedikit kode di sini untuk menyediakan instruksi bagi orang yang menjalankannya (mungkin Anda), dan untuk menyediakan versi “aktivitas” dari daftar tugas. Decider menggunakan nama daftar tugas ini untuk menjadwalkan aktivitas untuk alur kerja, dan penerapan aktivitas akan mendengarkan kejadian aktivtas pada nama daftar tugas ini untuk mengetahui kapan harus memulai aktivitas terjadwal dan untuk menyediakan pembaruan tentang eksekusi aktivitas.

Kode juga menunggu pengguna untuk mulai menjalankan starter aktivitas sebelum memulai eksekusi alur kerja, sehingga starter aktivitas akan siap merespons ketika tugas aktivitas mulai muncul di daftar tugas yang disediakan.

Langkah selanjutnya

Anda telah menerapkan alur kerja. Selanjutnya, Anda akan menentukan aktivitas dan starter aktivitas, di Tutorial Alur Kerja Bagian 3: Menerapkan Aktivitas.