class CustomersController < ApplicationController
  def send_invoice
    @customer = Customer.find(params[:id])
    Email.send_email(:pdf_invoice,
                     :to => @customer.email,
                     :customer => @customer)
  end
end

class CustomersController < ApplicationController
  def send_invoice
    @customer = Customer.find(params[:id])
    Process.fork do
      Email.send_email(:pdf_invoice,
                       :to = @customer.email,
                       :customer => @customer)
    end
  end
end

class CustomersController < ApplicationController
  def send_invoice
    @customer = Customer.find(params[:id])
    EmailWork.create(:template_name => :pdf_invoice,
                     :options => {
                       :to => @customer.email,
                       :customer => @customer,
                     })
  end
end

class CustomersController < ApplicationController
  def send_invoice
    @customer = Customer.find(params[:id])
    queue_service.put(:queue => :Email,
                      :action => :send_email,
                      :template_name => :pdf_invoice,
                      :options => {
                        :to => @customer.email,
                        :customer => @customer,
                      })
  end
end

module Email
  def send_email template_name, options
    $stderr.puts "*** #{$$}: Email.send_mail #{template_name.inspect} #{options.inspect}"
    :ok
  end
  def do_raise msg
    raise msg
  end
  extend self
end

Email.send_email(:pdf_invoice,
                 :to => "user@email.com",
                 :customer => @customer)
# ->
Email.asir.
      send_email(:pdf_invoice,
                 :to => "user@email.com",
                 :customer => @customer)

Email.asir.send_email(:pdf_invoice,
                        :to => "user@email.com",
                        :customer => @customer)
# ->
message = Message.new(...)
message.receiver_class == ::Module
message.receiver == ::Email
message.selector == :send_email
message.arguments == [ :pdf_invoice,
                       { :to => "user@email.com",
                         :customer => ... } ]

  module Client
    # Client Mixin
    def self.included target; ...; end 
    # Client Proxy
    class Proxy; ...; end 
  end

  module Client
    class Proxy
      attr_accessor :receiver, :receiver_class

      # Accept messages as a proxy for the receiver.
      # Blocks are used represent a "continuation" for the Result.
      def send selector, *arguments, &block
        message = Message.new(@receiver, selector, arguments, block, self)
        message = @before_send_message.call(message) if @before_send_message
        @__configure.call(message, self) if @__configure
        result = transport.send_message(message)
        result
      end
      # Accept all other messages to be encoded and transported to a service.
      alias :method_missing :send

      # Client Transport
      def transport; ...; end 
      # Proxy Configuration
      def _configure &blk; ...; end 
      # Configuration Callbacks
      def initialize rcvr, rcvr_class; ...; end 
    end
  end

  class Message
    include AdditionalData, Identity, CodeMore
    attr_accessor :receiver, :receiver_class, :selector, :arguments, :block
    attr_accessor :result, :one_way

    def initialize r, s, a, b, p
      @receiver, @selector, @arguments = r, s, a
      @block = b if b
      @receiver_class = @receiver.class
      @one_way = p._one_way if p
    end

    def invoke!
      @result = Result.new(self, @receiver.__send__(@selector, *@arguments))
    rescue *Error::Unforwardable.unforwardable => exc
      @result = Result.new(self, nil, Error::Unforwardable.new(exc))
    rescue ::Exception => exc
      @result = Result.new(self, nil, exc)
    end

    # Optional: Specifies the Numeric seconds or absolute Time for the Transport to delay the Message until actual invocation.
    attr_accessor :delay
  end

  class Result
    include AdditionalData, Identity, CodeMore::Result
    attr_accessor :message, :result, :exception
    # Optional: Opaque data about the server that processed the Message.
    attr_accessor :server

    def initialize msg, res = nil, exc = nil
      @message = msg; @result = res
      @exception = exc && EncapsulatedException.new(exc)
      @identifier = @message.identifier
    end
  end

  class Coder
    include Log, Initialization

    def encode obj
      _encode obj
    end

    def decode obj
      obj and _decode obj
    end

    # Coder subclasses:
    def _subclass_responsibility *args
      raise "subclass responsibility"
    end
    alias :_encode :_subclass_responsibility
    alias :_decode :_subclass_responsibility
  end

    class Identity < self
      def _encode obj
        obj
      end

      def _decode obj
        obj
      end

      # Completely stateless.
      def dup; self; end
    end

  class Transport
    include Log, Initialization, AdditionalData, Message::Delay, ThreadVariable, Conduit

    attr_accessor :encoder, :decoder, :one_way

    # Transport#send_message
    def send_message message; ...; end 
    # Transport#receive_message
    def receive_message stream; ...; end 

    # Transport#send_result
    def send_result result, stream, message_state; ...; end 

    # Transport#receive_result
    def receive_result message, opaque_result; ...; end 

    def initialize *args
      @verbose = 0
      super
    end

    # Incremented for each message sent or received.
    attr_accessor :message_count

    # A Proc to call within #receive_message, after #_receive_message.
    # trans.after_receive_message(trans, message)
    attr_accessor :after_receive_message

    # A Proc to call within #send_message, before #_send_message.
    # trans.before_send_message(trans, message)
    attr_accessor :before_send_message

    # Proc to call with #invoke_message! if result.exception.
    # trans.on_result_exception.call(trans, result)
    attr_accessor :on_result_exception

    # Proc to call with exception, if exception occurs within #serve_message!, but outside
    # Message#invoke!.
    #
    # trans.on_exception.call(trans, exception, :message, Message_instance, nil)
    # trans.on_exception.call(trans, exception, :result, Message_instance, Result_instance)
    attr_accessor :on_exception

    attr_accessor :needs_message_identifier, :needs_message_timestamp
    def needs_message_identifier? m; @needs_message_identifier; end
    def needs_message_timestamp?  m; @needs_message_timestamp; end

    attr_accessor :verbose

    def _subclass_responsibility *args
      raise Error::SubclassResponsibility "subclass responsibility"
    end
    alias :_send_message :_subclass_responsibility
    alias :_receive_message :_subclass_responsibility
    alias :_send_result :_subclass_responsibility
    alias :_receive_result :_subclass_responsibility

    # Serve a Message.
    def serve_message! in_stream, out_stream; ...; end 
    # Transport Server Support
    def stop! force = false; ...; end 
    # Transport Support
    def encoder; ...; end 

  end

  module Client
    class Proxy
      attr_accessor :transport

      def transport
        @transport ||= Transport::Local.new
      end
    end
  end

require 'example_helper' # !COMMENT
pr Email.send_email(:pdf_invoice,
                    :to => "user@email.com",
                    :customer => @customer)

require 'example_helper'
pr Email.asir.send_email(:pdf_invoice,
                           :to => "user@email.com",
                           :customer => @customer)

require 'asir'
# Added .asir support.
module Email
  include ASIR::Client # Email.asir
  def send_email template_name, options
    $stderr.puts "*** #{$$}: Email.send_mail #{template_name.inspect} #{options.inspect}"
    :ok
  end
  def do_raise msg
    raise msg
  end
  extend self
end

require 'example_helper'
begin
  Email.asir.transport = t =
    ASIR::Transport::Subprocess.new

  pr Email.asir.send_email(:pdf_invoice,
                             :to => "user@email.com",
                             :customer => @customer)
end

    class Subprocess < Local
      def initialize *args
        @one_way = true; super
      end

      def _send_message message, message_payload
        Process.fork do
          send_result(super, nil, nil)
        end
      end

      # one-way; no Result
      def _receive_result message, opaque_result
      end

      # one-way; no Result
      def _send_result message, result, result_payload, stream, message_state
      end
    end

require 'example_helper'
begin
  Email.asir.transport = t =
    ASIR::Transport::Subprocess.new(:one_way => true)
  pr(Email.asir.send_email(:pdf_invoice,
                             :to => "user@email.com",
                             :customer => @customer) { | resp |
     pr [ :in_block, resp.result ] })
end

require 'example_helper'
begin
  File.unlink(service_log = "#{__FILE__}.service.log") rescue nil
  Email.asir.transport = t =
    ASIR::Transport::File.new(:file => service_log)
  t.encoder =
    ASIR::Coder::Yaml.new
  pr Email.asir.send_email(:pdf_invoice,
                             :to => "user@email.com",
                             :customer => @customer)
ensure
  t.close
  puts "\x1a\n#{service_log.inspect} contents:"
  puts File.read(service_log)
end

require 'example_helper'
begin
  service_log = "#{__FILE__.sub('ex05', 'ex04')}.service.log"
  Email.asir.transport = t =
    ASIR::Transport::File.new(:file => service_log)
  t.encoder =
    ASIR::Coder::Yaml.new
  t.serve_file!
end

$stderr.puts "  #{$$} at #{__FILE__}:#{__LINE__}"

require 'example_helper'
begin
  File.unlink(service_pipe = "service.pipe") rescue nil
  Email.asir.transport = t =
    ASIR::Transport::File.new(:file => service_pipe)
  t.encoder =
    ASIR::Coder::Yaml.new
  server_process do
    t.prepare_server!
    t.run_server!
  end
  pr Email.asir.send_email(:pdf_invoice, :to => "user@email.com", :customer => @customer)
ensure
  t.close; sleep 1; server_kill
end

    class File < Stream
      def prepare_server!
        # _log [ :prepare_pipe_server!, file ]
        unless ::File.exist? file
          system(cmd = "mkfifo #{file.inspect}") or raise "cannot run #{cmd.inspect}"
          ::File.chmod(perms, file) rescue nil if perms
        end
      end
      alias :prepare_pipe_server! :prepare_server!

      def run_server!
        # _log [ :run_pipe_server!, file ]
        with_server_signals! do
          @running = true
          while @running
            serve_file!
          end
        end
      end
      alias :run_pipe_server! :run_server!
    end

    class Chain < self
      attr_accessor :encoders

      def _encode obj
        encoders.each do | e |
          obj = e.dup.encode(obj)
        end
        obj
      end

      def _decode obj
        encoders.reverse_each do | e |
          obj = e.dup.decode(obj)
        end
        obj
      end
    end
  end
end

require 'asir/coder'

module ASIR
  class Coder

require 'example_helper'
begin
  File.unlink(service_pipe = "service.pipe") rescue nil
  Email.asir.transport = t =
    ASIR::Transport::File.new(:file => service_pipe)
  t.encoder =
    ASIR::Coder::Chain.new(:encoders =>
      [ ASIR::Coder::Marshal.new,
        s = ASIR::Coder::Sign.new(:secret => 'abc123'),
        ASIR::Coder::Yaml.new,
      ])
  t.prepare_pipe_server!
  server_process do
    t.run_pipe_server!
  end
  pr Email.asir.send_email(:pdf_invoice, :to => "user@email.com", :customer => @customer)
ensure
  t.close; sleep 1; server_kill
end

require 'example_helper'
begin
  File.unlink(service_pipe = "service.pipe") rescue nil
  Email.asir.transport = t =
    ASIR::Transport::File.new(:file => service_pipe)
  t.encoder =
    ASIR::Coder::Chain.new(:encoders =>
      [ ASIR::Coder::Marshal.new,
        s = ASIR::Coder::Sign.new(:secret => 'abc123'),
        ASIR::Coder::Yaml.new,
      ])
  t.prepare_pipe_server!
  server_process do
    t.run_pipe_server!
  end
  s.secret = 'I do not know the secret! :('
  pr Email.asir.send_email(:pdf_invoice, :to => "user@email.com", :customer => @customer)
ensure
  t.close; sleep 1; server_kill
end

require 'example_helper'
begin
  Email.asir.transport = t =
    ASIR::Transport::TcpSocket.new(:port => 30909)
  t.encoder =
    ASIR::Coder::Marshal.new
  t.prepare_server!
  server_process do
    t.run_server!
  end
  pr Email.asir.send_email(:pdf_invoice,
                             :to => "user@email.com", :customer => @customer)
ensure
  t.close; sleep 1; server_kill
end

require 'example_helper'
begin
  Email.asir.transport = t =
    ASIR::Transport::TcpSocket.new(:port => 30910)
  t.encoder =
    ASIR::Coder::Marshal.new
  t.prepare_server!
  server_process do
    t.run_server!
  end
  pr Email.asir.do_raise("Raise Me!")
rescue Exception => err
  pr [ :exception, err ]
ensure
  t.close; sleep 1; server_kill
end

Email.do_raise("DOH!")
#
# ->
result.exception = ee = EncapsulatedException.new(...)
ee.exception_class = "::RuntimeError"
ee.exception_message = "DOH!"
ee.exception_backtrace = [ ... ]

  class EncapsulatedException
    include ObjectResolving, AdditionalData
    attr_accessor :exception_class, :exception_message, :exception_backtrace

    def initialize exc
      @exception_class     = exc.class.name
      @exception_message   = exc.message
      @exception_backtrace = exc.backtrace
    end

    def invoke!
      raise resolve_object(@exception_class), @exception_message, @exception_backtrace
    end
  end

require 'example_helper'
begin
  File.unlink(service_log = "#{__FILE__}.service.log") rescue nil
  Email.asir.transport = t =
    ASIR::Transport::Fallback.new(:transports => [
      tcp = ASIR::Transport::TcpSocket.new(:port => 31911,
                                           :encoder => ASIR::Coder::Marshal.new),
      ASIR::Transport::Broadcast.new(:transports => [
        file = ASIR::Transport::File.new(:file => service_log,
                                         :encoder => ASIR::Coder::Yaml.new),
        ASIR::Transport::Subprocess.new,
      ]),
    ])
  pr Email.asir.send_email(:pdf_invoice,
                             :to => "user@email.com", :customer => @customer)
  server_process do
    tcp.prepare_server!
    tcp.run_server!
  end
  pr Email.asir.send_email(:pdf_invoice,
                             :to => "user2@email.com", :customer => @customer)
ensure
  file.close rescue nil;
  tcp.close rescue nil; sleep 1; server_kill
  puts "\x1a\n#{service_log.inspect} contents:"
  puts File.read(service_log)
end

  class Transport
    def send_message message
      @message_count ||= 0; @message_count += 1
      message.create_timestamp! if needs_message_timestamp? message
      message.create_identifier! if needs_message_identifier? message
      @before_send_message.call(self, message) if @before_send_message
      relative_message_delay! message
      message_payload = encoder.dup.encode(message)
      opaque_result = _send_message(message, message_payload)
      receive_result(message, opaque_result)
    end
  end

    class Local < self
      # Returns Result object after invoking Message.
      def _send_message message, message_payload
        invoke_message!(message)
      end

      # Returns Result object from #send_message.
      def _receive_result message, opaque_result
        opaque_result
      end
    end

    class Null < self
      def _send_message message, message_payload
        nil
      end
    end

    class Yaml < self
      def _encode obj
        case obj
        when Message, Result
          obj = obj.encode_more!
        end
        ::YAML::dump(obj)
      rescue ::Exception
        require 'pp'
        msg = "#{self}: failed to encode: #{$!.inspect}:\n  #{PP.pp(obj, '')}"
        $stderr.puts msg
        raise Error, msg
      end

      def _decode obj
        case obj = ::YAML::load(obj)
        when Message, Result
          obj.decode_more!
        else
          obj
        end
      end
    end # class
  end # class
end # module

require 'asir'

require 'zlib'

module ASIR
  class Coder
    class Zlib < self
      attr_accessor :compression_level

      def _encode obj
        raise TypeError unless String === obj
        ::Zlib::Deflate.deflate(obj, @compression_level || ::Zlib::DEFAULT_COMPRESSION)
      end
      def _decode obj
        raise TypeError unless String === obj
        ::Zlib::Inflate.inflate(obj)
      end
    end
  end
end

module ASIR

    class File < Stream
      include PayloadIO # _write, _read
      attr_accessor :file, :mode, :perms, :stream

      def initialize opts = nil; @one_way = true; super; end

      # Writes a Message payload String.
      def _send_message message, message_payload
        _write message_payload, stream
      ensure
        close if ::File.pipe?(file)
      end

      # Returns a Message payload String.
      def _receive_message stream, additional_data
        [ _read(stream), nil ]
      end

      # one-way; no Result.
      def _send_result message, result, result_payload, stream, message_state
        nil
      end

      # one-way; no Result.
      def _receive_result message, opaque_result
        nil
      end

      # File Transport Support
      def stream; ...; end 
      # Process (receive) messages from a file.
      def serve_file!; ...; end 
      # Named Pipe Server
      def prepare_server!; ...; end 
    end

    class File < Stream
      def stream
        @stream ||=
          begin
            stream = ::File.open(file, mode || "w+")
            ::File.chmod(perms, file) rescue nil if @perms
            after_connect!(stream) if respond_to?(:after_connect!)
            stream
          end
      end
    end

    module PayloadIO
      class UnexpectedResponse < Error; end

      HEADER = "# asir_payload_size: "
      FOOTER = "\n# asir_payload_end"

      def _write payload, stream
        stream.write HEADER
        stream.puts payload.size
        stream.write payload
        stream.puts FOOTER
        stream.flush
        stream
      end

      def _read stream
        size = /\d+$/.match(stream.readline.chomp)[0].to_i # HEADER (size)
        payload = stream.read(size)
        stream.readline # FOOTER
        stream.readline
        payload
      end

      def _read_line_and_expect! stream, regexp
        line = stream.readline
        unless match = regexp.match(line)
          _log { "_read_line_and_expect! #{stream} #{regexp.inspect} !~ #{line.inspect}" }
          raise UnexpectedResponse, "expected #{regexp.inspect}, received #{line.inspect}"
        end
        match
      end

    end
  end
end
require 'asir/transport/http'
require 'rack'

module ASIR
  class Transport

  class Transport
    def receive_message stream
      @message_count ||= 0; @message_count += 1
      additional_data = { }
      if req_and_state = _receive_message(stream, additional_data)
        message = req_and_state[0] = encoder.dup.decode(req_and_state.first)
        message.additional_data!.update(additional_data) if message
        if @after_receive_message
          begin
            @after_receive_message.call(self, message)
          rescue ::Exception => exc
            _log { [ :receive_message, :after_receive_message, :exception, exc ] }
          end
        end
      end
      req_and_state
    end
  end

    class File < Stream
      def serve_file!
        ::File.open(file, "r") do | stream |
          @running = true
          _serve_stream! stream, nil # One-way: no result stream.
        end
      end
    end

    class Stream < self

      # Serve all Messages from a stream.
      def serve_stream! in_stream, out_stream; ...; end 
      # Serve a Message from a stream.
      def serve_stream_message! in_stream, out_stream; ...; end 

    class Stream < self
      def serve_stream! in_stream, out_stream
        with_server_signals! do
          @running = true
          _serve_stream! in_stream, out_stream
        end
      end

      def _serve_stream! in_stream, out_stream
        while @running && ! stream_eof?(in_stream)
          begin
            serve_stream_message! in_stream, out_stream
          rescue Error::Terminate => err
            @running = false
            _log [ :serve_stream_terminate, err ]
          rescue ::Exception => err
            _log [ :serve_stream_error, err ]
          end
        end
      end

      # Subclasses can override this method.
      def stream_eof? stream
        stream.eof?
      end
    end

    class Stream < self
      def serve_stream_message! in_stream, out_stream
        serve_message! in_stream, out_stream
      end
    end
    end

    class Marshal < self
      def _encode obj
        ::Marshal.dump(obj)
      end

      def _decode obj
        ::Marshal.load(obj)
      end
    end # class

    class Sign < self
      attr_accessor :secret, :function

      def _encode obj
        payload = obj.to_s
        { :function  => function,
          :signature => ::Digest.const_get(function).
                          new.hexdigest(secret + payload),
          :payload   => payload }
      end

      def _decode obj
        raise SignatureError, "expected Hash, given #{obj.class}" unless Hash === obj
        payload = obj[:payload]
        raise SignatureError, "signature invalid" unless obj == _encode(payload)
        payload
      end

      # Sign Coder Support
      class SignatureError < Error; end; ...; end 
    end 

    class TcpSocket < ConnectionOriented
      # TCP Socket Client
      def _client_connect!; ...; end 
      # TCP Socket Server
      def _server!; ...; end 

    class TcpSocket < ConnectionOriented
      def _server!
        @server = TCPServer.open(port)
        @server.setsockopt(Socket::SOL_SOCKET, Socket::SO_KEEPALIVE, false)
      end

      def _server_accept_connection! server
        socket = server.accept
        [ socket, socket ] # Use same socket for in_stream and out_stream
      end

      def _server_close_connection! stream, out_stream
        stream.close rescue nil
      end
    end
    end

    class TcpSocket < ConnectionOriented
      def _client_connect!
        sock = TCPSocket.open(address, port)
      end
    end

  class Transport
    def send_result result, stream, message_state
      message = result.message
      if @one_way && message.block
        message.block.call(result)
      else
        result.message = nil # avoid sending back entire Message.
        result_payload = decoder.dup.encode(result)
        _send_result(message, result, result_payload, stream, message_state)
      end
    end
  end

  class Transport
    def receive_result message, opaque_result
      result_payload = _receive_result(message, opaque_result)
      result = decoder.dup.decode(result_payload)
      if result && ! message.one_way
        if exc = result.exception
          invoker.invoke!(exc, self)
        else
          if ! @one_way && message.block
            message.block.call(result)
          end
          result.result
        end
      end
    end
  end

    class Beanstalk < TcpSocket
      def _send_message message, message_payload
        stream.with_stream! do | s |
          begin
            match = 
              _beanstalk(s, 
                         "put #{message[:beanstalk_priority] || @priority} #{message[:beanstalk_delay] || @delay} #{message[:beanstalk_ttr] || @ttr} #{message_payload.size}\r\n",
                         /\AINSERTED (\d+)\r\n\Z/,
                         message_payload)
            job_id = message[:beanstalk_job_id] = match[1].to_i
            _log { "beanstalk_job_id = #{job_id.inspect}" } if @verbose >= 2
          rescue ::Exception => exc
            message[:beanstalk_error] = exc
            close
            raise exc
          end
        end
      end

      RESERVE = "reserve\r\n".freeze
    end

    class Beanstalk < TcpSocket
      def _receive_message channel, additional_data
        channel.with_stream! do | stream |
          begin
            match = 
              _beanstalk(stream,
                         RESERVE,
                         /\ARESERVED (\d+) (\d+)\r\n\Z/)
            additional_data[:beanstalk_job_id] = match[1].to_i
            additional_data[:beanstalk_message_size] = 
              size = match[2].to_i
            message_payload = stream.read(size)
            _read_line_and_expect! stream, /\A\r\n\Z/
            # Pass the original stream used to #_send_result below.
            [ message_payload, stream ]
          rescue ::Exception => exc
            _log { [ :_receive_message, :exception, exc ] }
            additional_data[:beanstalk_error] = exc
            channel.close
          end
        end
      end
    end

    class Beanstalk < TcpSocket
      def _send_result message, result, result_payload, channel, stream
        #
        # There is a possibility here the following could happen:
        #
        #   _receive_message
        #     channel == #<Channel:1>   
        #     channel.stream == #<TCPSocket:1234>
        #   end
        #   ...
        #   ERROR OCCURES:
        #      channel.stream.close
        #      channel.stream = nil
        #   ...
        #   _send_result 
        #     channel == #<Channel:1>
        #     channel.stream == #<TCPSocket:5678> # NEW CONNECTION
        #     stream.write "delete #{job_id}"
        #   ...
        #
        # Therefore: _receiver_message passes the original message stream to us.
        # We insure that the same stream is still the active one and use it.
        channel.with_stream! do | maybe_other_stream |
          _log [ :_send_result, "stream lost" ] if maybe_other_stream != stream
          job_id = message[:beanstalk_job_id] or raise "no beanstalk_job_id"
          _beanstalk(stream,
                     "delete #{job_id}\r\n",
                     /\ADELETED\r\n\Z/)
        end
      end
    end

    class Beanstalk < TcpSocket
      def _receive_result message, opaque_result
        nil
      end
    end

    class Fallback < self
      include Composite

      def _send_message message, message_payload
        result = sent = first_exception = nil
        transports.each do | transport |
          begin
            result = transport.send_message(message)
            sent = true
            break
          rescue ::Exception => exc
            first_exception ||= exc
            _handle_send_message_exception! transport, message, exc
          end
        end
        unless sent
          if first_exception && @reraise_first_exception
            $! = first_exception
            raise
          end
          raise FallbackError, "fallback failed"
        end
        result
      end
      class FallbackError < Error; end
    end

    class Buffer < self
      include Delegation

      # Transport to send_message.
      attr_accessor :transport

      def initialize *args
        super
        @messages = Queue.new
        @messages_mutex = Mutex.new
        @paused = 0
        @paused_mutex = Mutex.new
      end

      # If paused, queue messages,
      # Otherwise delegate immediately to #transport.
      def _send_message message, message_payload
        return nil if @ignore
        if paused?
          @messages_mutex.synchronize do
            @messages << message
          end
          nil
        else
          @transport.send_message(message)
        end
      end

      # Returns true if currently paused.
      # Messages are queued until #resume!.
      def paused?
        @paused > 0
      end

      # Pauses all messages until resume!.
      # May be called multiple times.
      def pause!
        @paused_mutex.synchronize do
          @paused += 1
        end
        self
      end

      # Will automatically call #flush! when not #paused?.
      def resume!
        should_flush = @paused_mutex.synchronize do
          @paused -= 1 if @paused > 0
          @paused == 0
        end
        flush! if should_flush
        self
      end

      def size
        @messages_mutex.synchronize do
          @messages.size
        end
      end

      # Will flush pending Messages even if ! #paused?.
      def flush!
        clear!.each do | message |
          @transport.send_message(message)
        end
        self
      end

      # Clear all pending Messages without sending them.
      # Returns Array of Messages that would have been sent.
      def clear!
        messages = [ ]
        @messages_mutex.synchronize do
          @messages.size.times do
            messages << @messages.shift(true)
          end
        end
        messages
      end

      # Take Message from head of Queue.
      def shift non_block=false
        @messages.shift(non_block)
      end

      # Processes queue.
      # Usually used in worker Thread.
      def process! non_block=false
        @running = true
        while @running && message = shift(non_block)
          @transport.send_message(message)
        end
        message
      end

      # Stop processing queue.
      def stop!
        @messages_mutex.synchronize do
          @ignore = true; @running = false
        end
        self
      end
    end

module ASIR
  # Reusable constants to avoid unnecessary garbage.
  EMPTY_ARRAY = [ ].freeze; EMPTY_HASH =  { }.freeze; EMPTY_STRING = ''.freeze
  MODULE_SEP = '::'.freeze; IDENTITY_LAMBDA = lambda { | x | x }
end

  Foo.new(:bar => 1, :baz => 2)

  obj = Foo.new; obj.bar = 1; obj.baz = 2; obj

  module Initialization
    def initialize opts = nil
      opts ||= EMPTY_HASH
      initialize_before_opts if respond_to? :initialize_before_opts
      opts.each do | k, v |
        send(:"#{k}=", v)
      end
      initialize_after_opts if respond_to? :initialize_after_opts
    end
  end
end

module ASIR

  module Log
    attr_accessor :_logger

    def self.included target
      super
      target.send(:extend, ClassMethods)
    end

    @@enabled = false
    def self.enabled
      @@enabled
    end
    def self.enabled= x
      @@enabled = x
    end

    module ClassMethods
      def _log_enabled= x
        (Thread.current[:'ASIR::Log.enabled'] ||= { })[self] = x
      end
      def _log_enabled?
        (Thread.current[:'ASIR::Log.enabled'] ||= { })[self]
      end
    end

    def _log_enabled= x
      @_log_enabled = x
    end

    def _log_enabled?
      ASIR::Log.enabled || 
        @_log_enabled || 
        self.class._log_enabled?
    end

    def _log msg = nil
      return unless _log_enabled?
      msg ||= yield if block_given?
      msg = String === msg ? msg : _log_format(msg)
      msg = "  #{$$} #{Module === self ? self : self.class} #{msg}"
      case @_logger
      when Proc
        @_logger.call msg
      when IO
        @_logger.puts msg
      else
        $stderr.puts msg
      end
      nil
    end

    def _log_result msg
      _log { 
        msg = String === msg ? msg : _log_format(msg);
        "#{msg} => ..." }
      result = yield
      _log { "#{msg} => \n    #{result.inspect}" }
      result
    end

    def _log_format obj
      case obj
      when Exception
        msg = "#{obj.inspect}"
        msg << "\n    #{obj.backtrace * "\n    "}" if false
        msg
      when Array
        obj.map { | x | _log_format x } * ", "
      else
        obj.inspect
      end
    end
  end # module
end # module

require 'asir'
require 'asir/environment'
require 'time'

module ASIR
class Main
  attr_accessor :env, :args, :exit_code
  # Delegate getter/setters to @env.
  [ :verb, :adjective, :object, :identifier,
    :config_rb,
    :verbose,
    :log_dir, :log_file,
    :pid_dir, :pid_file,
  ].
    map{|g| [ g, :"#{g}=" ]}.
    flatten.each do | m |
      define_method(m) { | *args | @env.send(m, *args) }
    end
  attr_accessor :progname
  # Options:
  attr_accessor :force

  # Transport selected from asir.phase = :transport.
  attr_accessor :transport

  def initialize
    self.env = ASIR::Environment.new
    self.progname = File.basename($0)
    self.exit_code = 0
  end

  def parse_args! args = ARGV.dup
    self.args = args
    until args.empty?
      case args.first
      when /^([a-z0-9_]+=)(.*)/i
        k, v = $1.to_sym, $2
        args.shift
        v = v.to_i if v == v.to_i.to_s
        send(k, v)
      else
        break
      end
    end
    self.verb, self.adjective, self.object, self.identifier = args.map{|x| x.to_sym}
    self.identifier ||= :'0'
    self
  end

  def config! *args
    @env.config! *args
  end

  def log_str
    "#{Time.now.gmtime.iso8601(4)} #{$$} #{log_str_no_time}"
  end

  def log_str_no_time
    "#{progname} #{verb} #{adjective} #{object} #{identifier}"
  end

  def run!
    unless verb && adjective && object
      self.exit_code = 1
      return usage!
    end
    config!(:configure)
    # $stderr.puts "log_file = #{log_file.inspect}"
    case self.verb
    when :restart
      self.verb = :stop
      _run_verb! && sleep(1)
      self.verb = :start
      _run_verb!
    else
      _run_verb!
    end
    self
  rescue ::Exception => exc
    $stderr.puts "#{log_str} ERROR\n#{exc.inspect}\n  #{exc.backtrace * "\n  "}"
    self.exit_code += 1
    self
  end

  def _run_verb!
    sel = :"#{verb}_#{adjective}_#{object}!"
    if verbose >= 3
      $stderr.puts "  verb      = #{verb.inspect}"
      $stderr.puts "  adjective = #{adjective.inspect}"
      $stderr.puts "  object    = #{object.inspect}"
      $stderr.puts "  sel       = #{sel.inspect}"
    end
    send(sel)
  rescue ::Exception => exc
    $stderr.puts "#{log_str} ERROR\n#{exc.inspect}\n  #{exc.backtrace * "\n  "}"
    self.exit_code += 1
    raise
    nil
  end

  def method_missing sel, *args
    log "method_missing #{sel}" if verbose >= 3
    case sel.to_s
    when /^start_([^_]+)_worker!$/
      _start_worker!
    when /^status_([^_]+)_([^_]+)!$/
      pid = server_pid
      puts "#{log_str} pid #{pid}"
      system("ps -fw -p #{pid}")
    when /^log_([^_]+)_([^_]+)!$/
      puts log_file
    when /^taillog_([^_]+)_([^_]+)!$/
      exec "tail -f #{log_file.inspect}"
    when /^pid_([^_]+)_([^_]+)!$/
      pid = server_pid rescue nil
      alive = process_running? pid
      puts "#{pid_file} #{pid || :NA} #{alive}"
    when /^alive_([^_]+)_([^_]+)!$/
      pid = server_pid rescue nil
      alive = process_running? pid
      puts "#{pid_file} #{pid || :NA} #{alive}" if @verbose
      self.exit_code += 1 unless alive
    when /^stop_([^_]+)_([^_]+)!$/
      kill_server!
    else
      super
    end
  end

  def usage!
    $stderr.puts <<"END"
SYNOPSIS:
  asir [ <<options>> ... ] <<verb>> <<adjective>> <<object>> [ <<identifier>> ]

OPTIONS:
  config_rb=file.rb ($ASIR_LOG_DIR)
  pid_dir=dir/      ($ASIR_PID_DIR)
  log_dir=dir/      ($ASIR_LOG_DIR)
  verbose=[0-9]

VERBS:
  start
  stop
  restart
  status
  log
  pid
  alive

ADJECTIVE-OBJECTs:
  beanstalk conduit
  beanstalk worker
  zmq worker
  webrick worker
  resque conduit
  resque worker

EXAMPLES:

  export ASIR_CONFIG_RB="some_system/asir_config.rb"
  asir start beanstalk conduit
  asir status beanstalk conduit

  asir start webrick worker
  asir pid   webrick worker

  asir start beanstalk worker 1
  asir start beanstalk worker 2

  asir start zmq worker
  asir start zmq worker 1
  asir start zmq worker 2
END
  end

  def start_beanstalk_conduit!
    _start_conduit!
  end

  def start_resque_conduit!
    _start_conduit!
  end

  def _start_conduit!
    config!(:environment)
    self.transport = config!(:transport)
    fork_server! do
      transport.start_conduit! :fork => false
    end
  end

  def _start_worker! type = adjective
    log "start_worker! #{type}"
    type = type.to_s
    fork_server! do
      transport_file = "asir/transport/#{type}"
      log "loading #{transport_file}"
      require transport_file
      _create_transport ASIR::Transport.const_get(type[0..0].upcase + type[1..-1])
      _run_workers!
    end
  end

  def fork_server! cmd = nil, &blk
    pid = Process.fork do
      run_server! cmd, &blk
    end
    log "forked pid #{pid}"
    Process.detach(pid) # Forks a Thread?  We are gonna exit anyway.
    File.open(pid_file, "w+") { | o | o.puts pid }
    File.chmod(0666, pid_file) rescue nil

    # Wait and check if process still exists.
    sleep 3
    unless process_running? pid
      raise "Server process #{pid} died to soon?"
    end

    self
  end

  def run_server! cmd = nil
    lf = File.open(log_file, "a+")
    File.chmod(0666, log_file) rescue nil
    $stdin.close rescue nil
    STDIN.close rescue nil
    STDOUT.reopen(lf)
    $stdout.reopen(lf) if $stdout.object_id != STDOUT.object_id
    STDERR.reopen(lf)
    $stderr.reopen(lf) if $stderr.object_id != STDERR.object_id
    # Process.daemon rescue nil # Ruby 1.9.x only.
    lf.puts "#{log_str} starting pid #{$$}"
    begin
      if cmd
        exec(cmd)
      else
        yield
      end
    ensure
      lf.puts "#{log_str} finished pid #{$$}"
      File.unlink(pid_file) rescue nil
    end
    self
  rescue ::Exception => exc
    msg = "ERROR pid #{$$}\n#{exc.inspect}\n  #{exc.backtrace * "\n  "}"
    log msg, :stderr
    raise
    self
  end

  def kill_server!
    log "#{log_str} kill"
    pid = server_pid
    stop_pid! pid
  rescue ::Exception => exc
    log "#{log_str} ERROR\n#{exc.inspect}\n  #{exc.backtrace * "\n  "}", :stderr
    raise
  end

  def log msg, to_stderr = false
    if to_stderr
      $stderr.puts "#{log_str_no_time} #{msg}"
    end
    File.open(log_file, "a+") do | log |
      log.puts "#{log_str} #{msg}"
    end
  end

  def server_pid
    pid = File.read(pid_file).chomp!
    pid.to_i
  end

  def _create_transport default_class
    config!(:environment)
    case transport = config!(:transport)
    when default_class
      self.transport = transport
    else
      raise "Expected config to return a #{default_class}, not a #{transport.class}"
    end
  end

  def worker_pids
    (@worker_pids ||= { })[adjective] ||= { }
  end

  def _run_workers!
    $0 = "#{progname} #{adjective} #{object} #{identifier}"

    worker_id = 0
    transport.prepare_server!
    worker_processes = transport[:worker_processes] || 1
    (worker_processes - 1).times do
      wid = worker_id += 1
      pid = Process.fork do
        _run_transport_server! wid
      end
      Process.setgprp(pid, 0) rescue nil
      worker_pids[wid] = pid
      log "forked #{wid} pid #{pid}"
    end

    _run_transport_server!
  ensure
    log "worker 0 stopped"
    _stop_workers!
  end

  def _run_transport_server! wid = 0
    log "running transport worker #{transport.class} #{wid}"
    config!(:start)
    $0 += " #{wid} #{transport.uri rescue nil}"
    old_arg0 = $0.dup
    after_receive_message = transport.after_receive_message || lambda { | transport, message | nil }
    transport.after_receive_message = lambda do | transport, message |
      $0 = "#{old_arg0} #{transport.message_count} #{message.identifier}"
      after_receive_message.call(transport, message)
    end
    transport.run_server!
    self
  end

  def _stop_workers!
    workers = worker_pids.dup
    worker_pids.clear
    workers.each do | wid, pid |
      config!(:stop)
      stop_pid! pid, "wid #{wid} "
    end
    workers.each do | wid, pid |
      wr = Process.waitpid(pid) rescue nil
      log "stopped #{wid} pid #{pid} => #{wr.inspect}", :stderr
    end
  ensure
    worker_pids.clear
  end

  def stop_pid! pid, msg = nil
    log "stopping #{msg}pid #{pid}", :stderr
    if process_running? pid
      log "TERM pid #{pid}"
      Process.kill('TERM', pid) rescue nil
      sleep 3
      if force or process_running? pid
        log "KILL pid #{pid}", :stderr
        Process.kill('KILL', pid) rescue nil
      end
      if process_running? pid
        log "cant-stop pid #{pid}", :stderr
      end
    else
      log "not-running? pid #{pid}", :stderr
    end
  end

  def process_running? pid
    case pid
    when false, nil
      pid
    when Integer
      Process.kill(0, pid)
    else
      raise TypeError, "expected false, nil, Integer; given #{pid.inspect}"
    end
    true
  rescue ::Errno::ESRCH
    false
  rescue ::Exception => exc
    $stderr.puts "  DEBUG: process_running? #{pid} => #{exc.inspect}"
    false
  end

end # class
end # module

module ASIR
  class Message
    module Delay
      # Returns the number of seconds from now, that the message should be delayed.
      # If message.delay is Numeric, sets message.delay to the Time to delay til.
      # If message.delay is Time, returns (now - message.delay).to_f
      # Returns Float if message.delay was set, or nil.
      # Returns 0 if delay has already expired.
      def relative_message_delay! message, now = nil
        case delay = message.delay
        when nil
        when Numeric
          now ||= Time.now
          delay = delay.to_f
          message.delay = (now + delay).utc
        when Time
          now ||= Time.now
          delay = (delay - now).to_f
          delay = 0 if delay < 0
        else
          raise TypeError, "Expected message.delay to be Numeric or Time, given #{delay.class}"
        end
        delay
      end

      def wait_for_delay! message
        while (delay = relative_message_delay!(message)) && delay > 0 
          sleep delay
        end
        self
      end

    end
  end
end

module ASIR

    class Null < self
      def _encode obj
        nil
      end

      def _decode obj
        nil
      end

      # Completely stateless.
      def dup; self; end
    end

    class Sign < self
      class SignatureError < Error; end

      def initialize_before_opts
        @function = :SHA1
      end
    end

    class Broadcast < self
      include Composite

      def _send_message message, message_payload
        result = first_exception = nil
        transports.each do | transport |
          begin
            result = transport.send_message(message)
          rescue ::Exception => exc
            first_exception ||= exc
            _handle_send_message_exception! transport, message, exc
            raise unless @continue_on_exception
          end
        end
        if first_exception && @reraise_first_exception
          $! = first_exception
          raise
        end
        result
      end

    end

    class HTTP < self
      attr_accessor :uri, :server, :debug

      # Client-side: HTTPClient

      def client
        @client ||=
          Channel.new(:on_connect =>
            lambda { | channel | ::HTTPClient.new })
      end

      def close
        @client.close if @client
      ensure
        @client = nil unless Channel === @client
      end

      # Send the Message payload String using HTTP POST.
      # Returns the HTTPClient::Request response object.
      def _send_message message, message_payload
        client.with_stream! do | client |
          client.post(uri, message_payload)
        end
      end

      # Recieve the Result payload String from the opaque
      # HTTPClient::Request response object returned from #_send_message.
      def _receive_result message, http_result_message
        # $stderr.puts " ### http_result_message.content.encoding = #{http_result_message.content.encoding.inspect}" rescue nil
        # $stderr.puts " ### http_result_message.content = #{http_result_message.content.inspect}" rescue nil
        http_result_message.content.to_s
      end

      CONTENT_TYPE = 'Content-Type'.freeze
      APPLICATION_BINARY = 'application/binary'.freeze

    end

  class Transport
    def encoder
      @encoder ||=
        Coder::Identity.new
    end

    def decoder
      @decoder ||=
        encoder
    end

    # Invokes the Message object, returns a Result object.
    def invoke_message! message
      result = nil
      Transport.with_attr! :current, self do
        with_attr! :message, message do
          wait_for_delay! message
          result = invoker.invoke!(message, self)
          # Hook for Exceptions.
          if @on_result_exception && result.exception
            @on_result_exception.call(self, result)
          end
        end
      end
      result
    end
    # The current Message being handled.
    attr_accessor_thread :message

    # The current active Transport.
    cattr_accessor_thread :current

    # The Invoker responsible for invoking the Message.
    attr_accessor :invoker
    def invoker
      @invoker ||= Invoker.new
    end
  end

  module Identity
    attr_accessor :identifier, :timestamp

    # Optional: Opaque data about the Client that created the Message.
    attr_accessor :client

    # Optional: Opaque data about the Service that handled the Result.
    attr_accessor :server

    # Creates a thread-safe unique identifier.
    def create_identifier!
      @identifier ||= 
        @@identifier_mutex.synchronize do
          if @@uuid_pid != $$
            @@uuid_pid = $$
            @@uuid = nil
          end
          "#{@@counter += 1}-#{@@uuid ||= ::ASIR::UUID.generate}".freeze
        end
    end
    @@counter ||= 0; @@uuid ||= nil; @@uuid_pid = nil; @@identifier_mutex ||= Mutex.new

    # Creates a timestamp.
    def create_timestamp!
      @timestamp ||= 
        ::Time.now.gmtime
    end
  end
end

module ASIR

    module Delay
      # Returns the number of seconds from now, that the message should be delayed.
      # If message.delay is Numeric, sets message.delay to the Time to delay til.
      # If message.delay is Time, returns (now - message.delay).to_f
      # Returns Float if message.delay was set, or nil.
      # Returns 0 if delay has already expired.
      def relative_message_delay! message, now = nil
        case delay = message.delay
        when nil
        when Numeric
          now ||= Time.now
          delay = delay.to_f
          message.delay = (now + delay).utc
        when Time
          now ||= Time.now
          delay = (delay - now).to_f
          delay = 0 if delay < 0
        else
          raise TypeError, "Expected message.delay to be Numeric or Time, given #{delay.class}"
        end
        delay
      end

      def wait_for_delay! message
        while (delay = relative_message_delay!(message)) && delay > 0 
          sleep delay
        end
        self
      end

    end
  end
end
module ASIR
  class Transport

  module CodeMore
    include ObjectResolving # resolve_object()
    include CodeBlock # encode_block!, decode_block!

    def encode_more!
      obj = encode_block! # may self.dup
      unless ::String === @receiver_class
        obj ||= self.dup # dont dup twice.
        obj.receiver = @receiver.name if ::Module === @receiver
        obj.receiver_class = @receiver_class.name
        if resp = obj.result and resp.message == self
          resp.message = obj
        end
      end
      obj || self
    end

    def decode_more!
      decode_block!
      if ::String === @receiver_class
        @receiver_class = resolve_object(@receiver_class)
        @receiver = resolve_object(@receiver) if ::Module === @receiver_class
        unless @receiver_class === @receiver
          raise Error, "receiver #{@receiver.class.name} is not a #{@receiver_class}" 
        end
      end
      self
    end

    # Mixin for Result.
    module Result
      def encode_more!
        @message = @message.encode_more! if @message
        self
      end

      def decode_more!
        @message = @message.decode_more! if @message
        self
      end
    end
  end

  module Client
    def self.included target
      super
      target.extend ModuleMethods if Module === target
      target.send(:include, InstanceMethods) if Class === target
    end

    module CommonMethods
      def asir_options &blk
        asir._configure(&blk)
      end
      alias_method :"#{ASIR::Config.client_method}_options", :asir_options if ASIR::Config.client_method
    end

    module ModuleMethods
      include CommonMethods
      # Proxies are cached for Module/Class methods because serialization will not include
      # Transport.
      def asir
        @_asir ||= ASIR::Client::Proxy.new(self, self)
      end
      alias_method ASIR::Config.client_method, :asir if ASIR::Config.client_method
    end

    module InstanceMethods
      include CommonMethods
      # Proxies are not cached in instances because receiver is to be serialized by
      # its Transport's Coder.
      def asir
        proxy = self.class.asir.dup
        proxy.receiver = self
        proxy
      end
      alias_method ASIR::Config.client_method, :asir if ASIR::Config.client_method
    end
  end

  module Client
    class Proxy
      attr_accessor :before_send_message

      # If true, this Message is one-way, even if the Transport is bi-directional.
      attr_accessor :_one_way

      # A Proc to call with the Message object before sending to transport#send_message(message).
      # See #_configure.
      attr_accessor :__configure

      # Returns a new Client Proxy with a block to be called with the Message.
      # This block can configure additional options of the Message before
      # it is sent to the Transport.
      #
      # Call sequence:
      #
      #   proxy.__configure.call(message, proxy).
      def _configure &blk
        proxy = @receiver == @receiver_class ? self.dup : self
        proxy.__configure = blk
        proxy
      end
      alias :_options :_configure
    end
  end

  module AdditionalData
    def additional_data
      @additional_data || EMPTY_HASH
    end
    def additional_data!
      @additional_data ||= { }
    end
    def additional_data= x
      @additional_data = x
    end
    def [] key
      @additional_data && @additional_data[key]
    end
    def []= key, value
      (@additional_data ||= { })[key] = value
    end

    def self.included target
      super
      target.extend(ModuleMethods)
    end

    module ModuleMethods
      # Provide a getter method that delegates to addtional_data[...].
      def addr_getter *names
        names.each do | name |
          name = name.to_sym
          define_method(name) { | | addtional_data[name] }
        end
      end

      # Provide getter and setter methods that delegate to addtional_data[...].
      def addr_accessor *names
        addr_getter *names
        names.each do | name |
          name = name.to_sym
          define_method(:"#{name}=") { | v | addtional_data[name] = v }
        end
      end
    end

  end

  module Client
    class Proxy
      def initialize rcvr, rcvr_class
        @receiver = rcvr
        @receiver_class = rcvr_class
        _configure!
      end

      def _configure!
        key = @receiver_class
        (@@config_callbacks[key] ||
         @@config_callbacks[key.name] ||
         @@config_callbacks[nil] ||
         IDENTITY_LAMBDA).call(self)
        self
      end

      @@config_callbacks ||= { }
      def self.config_callbacks
        @@config_callbacks
      end
    end
  end

  module CodeBlock
    # Most coders cannot serialize Procs.
    # But we can attempt to serialize a String representing a Proc.
    def encode_block!
      obj = nil
      if @block && ! ::String === @block_code
        obj ||= self.dup
        obj.block_code = CodeBlock.block_to_code(obj.block)
        obj.block = nil 
      end
      obj
    end

    def decode_block!
      if ::String === @block_code
        @block ||= CodeBlock.code_to_block(@block_code)
        @block_code = nil
      end
      self
    end

    # Returns a block_cache Hash.
    # Flushed every 1000 accesses.
    def self.block_cache
      cache = Thread.current[:'ASIR::CodeBlock.block_cache'] ||= { }
      count = Thread.current[:'ASIR::CodeBlock.block_cache_count'] ||= 0
      count += 1
      if count >= 1000
        cache.clear
        count = 0
      end
      Thread.current[:'ASIR::CodeBlock.block_cache_count'] = count
      cache
    end

    # Uses ruby2ruby, if loaded.
    def self.block_to_code block
      (block_cache[block.object_id] ||=
        [ block.respond_to?(:to_ruby) && block.to_ruby, block ]).
        first
    end

    # Calls eval.
    # May be unsafe.
    def self.code_to_block code
      (block_cache[code.dup.freeze] ||=
        [ eval(@block_code), code ]).
        first
    end

  end
end
module ASIR