receiver.rb

# Copyright 2011© MaestroDev.  All rights reserved.

module Maestro
  
  #
  # An error class for error emitted by the "remote side" and received here.
  #
  class ReceiveError < RuntimeError

    attr_reader :fei

    def initialize(fei, message)
      @fei = fei
      @message = message
      super("for #{Ruote::FlowExpressionId.to_storage_id(fei)}")
    end
    
    def to_s
      @message
    end
  end
  
  class Receiver < RuoteStomp::Receiver
    include Maestro::MaestroParticipantMixin
    
    def initialize(e_or_s, opts={})
      @context = e_or_s.context
      @options = opts
      @launchitems = opts[:launchitems]

      @queue =
        opts[:queue] ||
        (@launchitems == :only ? '/queue/ruote_launchitems' : '/queue/ruote_workitems')

      begin
        RuoteStomp.start!

        if opts[:unsubscribe]
          $stomp.unsubscribe(@queue)
          sleep 0.300
        end
       
        $stomp.subscribe(@queue, :ack => 'client') do |message|
          Maestro.log.debug "Received #{message}"
          unless $stomp.connected?
            # do nothing, we're going down
          else
            $stomp.ack message
            handle(message)
          end
        end
      rescue
        $stomp.disconnect
      end
    end
    
    def close
      $stomp.unsubscribe(@queue)
      $stomp.disconnect
    end
    
    def handle(msg)

      workitem = decode_workitem(msg.body)
      
      Maestro.log.debug "Received message from agent:  #{msg.body}"
      
      return unless workitem.is_a? Hash
      
      persist_end(workitem)
      
      if (!workitem['fields']["__error__"].nil? and workitem['fields']["__error__"].andand.size > 0)
        #completed not successfully
        #insert errors if they exist
        @context.error_handler.action_handle(
          'error', workitem["fei"], ReceiveError.new(workitem["fei"], workitem['fields']["__error__"])
        )
        Maestro.log.error("Error executing process - #{workitem["__error__"]}")
      else
        #completed successfully
        #persist results
        receive(workitem)
      end
    end
        
  end
end