proton-c/bindings/ruby/lib/messenger/messenger.rb - qpid-proton - Git at Google

 #
 # Licensed to the Apache Software Foundation (ASF) under one
 # or more contributor license agreements.  See the NOTICE file
 # distributed with this work for additional information
 # regarding copyright ownership.  The ASF licenses this file
 # to you under the Apache License, Version 2.0 (the
 # "License"); you may not use this file except in compliance
 # with the License.  You may obtain a copy of the License at
 #
 #   http://www.apache.org/licenses/LICENSE-2.0
 #
 # Unless required by applicable law or agreed to in writing,
 # software distributed under the License is distributed on an
 # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
 #

 module Qpid::Proton::Messenger

   # The +Messenger+ class defines a high level interface for
   # sending and receiving Messages. Every Messenger contains
   # a single logical queue of incoming messages and a single
   # logical queue of outgoing messages. These messages in these
   # queues may be destined for, or originate from, a variety of
   # addresses.
   #
   # The messenger interface is single-threaded.  All methods
   # except one ( #interrupt ) are intended to be used from within
   # the messenger thread.
   #
   # === Sending & Receiving Messages
   #
   # The Messenger class works in conjuction with the Message class. The
   # Message class is a mutable holder of message content.
   #
   # The put method copies its Message to the outgoing queue, and may
   # send queued messages if it can do so without blocking.  The send
   # method blocks until it has sent the requested number of messages,
   # or until a timeout interrupts the attempt.
   #
   # Similarly, the recv method receives messages into the incoming
   # queue, and may block as it attempts to receive the requested number
   # of messages,  or until timeout is reached. It may receive fewer
   # than the requested number.  The get method pops the
   # eldest Message off the incoming queue and copies it into the Message
   # object that you supply.  It will not block.
   #
   # The blocking attribute allows you to turn off blocking behavior entirely,
   # in which case send and recv will do whatever they can without
   # blocking, and then return.  You can then look at the number
   # of incoming and outgoing messages to see how much outstanding work
   # still remains.
   #
   class Messenger

     include Qpid::Proton::Util::ErrorHandler

     can_raise_error [:send, :receive, :password=, :start, :stop,
                      :perform_put, :perform_get, :interrupt,
                      :route, :rewrite, :accept, :reject,
                      :incoming_window=, :outgoing_window=]

     # Creates a new +Messenger+.
     #
     # The +name+ parameter is optional. If one is not provided then
     # a unique name is generated.
     #
     # ==== Options
     #
     # * name - the name (def. nil)
     #
     def initialize(name = nil)
       @impl = Cproton.pn_messenger(name)
       @selectables = {}
       ObjectSpace.define_finalizer(self, self.class.finalize!(@impl))
     end

     def self.finalize!(impl) # :nodoc:
       proc {
         Cproton.pn_messenger_free(impl)
       }
     end

     # Returns the name.
     #
     def name
       Cproton.pn_messenger_name(@impl)
     end

     # This property contains the password for the Messenger.private_key
     # file, or +nil+ if the file is not encrypted.
     #
     # ==== Arguments
     #
     # * password - the password
     #
     def password=(password)
       Cproton.pn_messenger_set_password(@impl, password)
     end

     # Returns the password property for the Messenger.private_key file.
     #
     def password
       Cproton.pn_messenger_get_password(@impl)
     end

     # Sets the timeout period, in milliseconds.
     #
     # A negative timeout period implies an infinite timeout.
     #
     # ==== Options
     #
     # * timeout - the timeout period
     #
     def timeout=(timeout)
       raise TypeError.new("invalid timeout: #{timeout}") if timeout.nil?
       Cproton.pn_messenger_set_timeout(@impl, timeout)
     end

     # Returns the timeout period
     #
     def timeout
       Cproton.pn_messenger_get_timeout(@impl)
     end

     # Returns true if blocking mode is enabled.
     #
     # Enable or disable blocking behavior during message sending
     # and receiving.  This affects every blocking call, with the
     # exception of work().  Currently, the affected calls are
     # send, recv, and stop.
     def blocking?
       Cproton.pn_messenger_is_blocking(@impl)
     end

     # Sets the blocking mode.
     def blocking=(blocking)
       Cproton.pn_messenger_set_blocking(@impl, blocking)
     end

     # Returns true if passive mode is enabled.
     #
     def passive?
       Cproton.pn_messenger_is_passive(@impl)
     end

     # Turns passive mode on or off.
     #
     # When set to passive mode, Messenger will not attempt to perform I/O
     # operations internally. In this mode it is necesssary to use the
     # Selectable type to drive any I/O needed to perform requestioned
     # actions.
     #
     # In this mode Messenger will never block.
     #
     def passive=(mode)
       Cproton.pn_messenger_set_passive(@impl, mode)
     end

     def deadline
       tstamp = Cproton.pn_messenger_deadline(@impl)
       return tstamp / 1000.0 unless tstamp.nil?
     end

     # Reports whether an error occurred.
     #
     def error?
       !Cproton.pn_messenger_errno(@impl).zero?
     end

     # Returns the most recent error number.
     #
     def errno
       Cproton.pn_messenger_errno(@impl)
     end

     # Returns the most recent error message.
     #
     def error
       Cproton.pn_error_text(Cproton.pn_messenger_error(@impl))
     end

     # Clears the current error state.
     #
     def clear_error
       error = Cproton.pn_messenger_error(@impl)
       unless error.nil?
         Cproton.pn_error_clear(error)
       end
     end

     # Currently a no-op placeholder.
     # For future compatibility, do not send or recv messages
     # before starting the +Messenger+.
     #
     def start
       Cproton.pn_messenger_start(@impl)
     end

     # Stops the +Messenger+, preventing it from sending or receiving
     # any more messages.
     #
     def stop
       Cproton.pn_messenger_stop(@impl)
     end

     # Returns true if a Messenger is in the stopped state.
     # This function does not block.
     #
     def stopped?
       Cproton.pn_messenger_stopped(@impl)
     end

     # Subscribes the Messenger to messages originating from the
     # specified source. The source is an address as specified in the
     # Messenger introduction with the following addition. If the
     # domain portion of the address begins with the '~' character, the
     # Messenger will interpret the domain as host/port, bind to it,
     # and listen for incoming messages. For example "~0.0.0.0",
     # "amqp://~0.0.0.0" will all bind to any local interface and
     # listen for incoming messages.  An address of "amqps://~0.0.0.0"
     # will only permit incoming SSL connections.
     #
     # ==== Options
     #
     # * address - the source address to be subscribe
     # * timeout - an optional time-to-live value, in seconds, for the
     #             subscription
     #
     def subscribe(address, timeout=0)
       raise TypeError.new("invalid address: #{address}") if address.nil?
       subscription = Cproton.pn_messenger_subscribe_ttl(@impl, address, timeout)
       raise Qpid::Proton::ProtonError.new("Subscribe failed") if subscription.nil?
       Subscription.new(subscription)
     end

     # Path to a certificate file for the +Messenger+.
     #
     # This certificate is used when the +Messenger+ accepts or establishes
     # SSL/TLS connections.  This property must be specified for the
     # Messenger to accept incoming SSL/TLS connections and to establish
     # client authenticated outgoing SSL/TLS connection.  Non client authenticated
     # outgoing SSL/TLS connections do not require this property.
     #
     # ==== Options
     #
     # * certificate - the certificate
     #
     def certificate=(certificate)
       Cproton.pn_messenger_set_certificate(@impl, certificate)
     end

     # Returns the path to a certificate file.
     #
     def certificate
       Cproton.pn_messenger_get_certificate(@impl)
     end

     # Path to a private key file for the +Messenger+.
     #
     # The property must be specified for the +Messenger+ to accept incoming
     # SSL/TLS connections and to establish client authenticated outgoing
     # SSL/TLS connections.  Non client authenticated SSL/TLS connections
     # do not require this property.
     #
     # ==== Options
     #
     # * key - the key file
     #
     def private_key=(key)
       Cproton.pn_messenger_set_private_key(@impl, key)
     end

     # Returns the path to a private key file.
     #
     def private_key
       Cproton.pn_messenger_get_private_key(@impl)
     end

     # A path to a database of trusted certificates for use in verifying the
     # peer on an SSL/TLS connection. If this property is +nil+, then the
     # peer will not be verified.
     #
     # ==== Options
     #
     # * certificates - the certificates path
     #
     def trusted_certificates=(certificates)
       Cproton.pn_messenger_set_trusted_certificates(@impl,certificates)
     end

     # The path to the databse of trusted certificates.
     #
     def trusted_certificates
       Cproton.pn_messenger_get_trusted_certificates(@impl)
     end

     # Places the content contained in the message onto the outgoing
     # queue of the Messenger.
     #
     # This method will never block, however it will send any unblocked
     # Messages in the outgoing queue immediately and leave any blocked
     # Messages remaining in the outgoing queue.
     # The send call may then be used to block until the outgoing queue
     # is empty.  The outgoing attribute may be used to check the depth
     # of the outgoing queue.
     #
     # ==== Options
     #
     # * message - the message
     #
     def put(message)
       if message.nil?
         raise TypeError.new("invalid message: #{message}")
       end
       unless message.kind_of?(Qpid::Proton::Message)
         raise ::ArgumentError.new("invalid message type: #{message.class}")
       end
       # encode the message first
       message.pre_encode
       perform_put(message)
       return outgoing_tracker
     end

     private

     def perform_put(message) # :nodoc:
       Cproton.pn_messenger_put(@impl, message.impl)
     end

     public


     # This call will block until the indicated number of messages
     # have been sent, or until the operation times out.
     # If n is -1 this call will block until all outgoing messages
     # have been sent. If n is 0 then this call will send whatever
     # it can without blocking.
     #
     def send(n = -1)
       Cproton.pn_messenger_send(@impl, n)
     end

     # Moves the message from the head of the incoming message queue into
     # the supplied message object. Any content in the supplied message
     # will be overwritten.
     # A tracker for the incoming Message is returned.  The tracker can
     # later be used to communicate your acceptance or rejection of the
     # Message.
     #
     # If no message is provided in the argument, then one is created. In
     # either case, the one returned will be the fetched message.
     #
     # ==== Options
     #
     # * msg - the (optional) +Message+ instance to be used
     #
     def get(msg = nil)
       msg_impl = nil
       if msg.nil? then
         msg_impl = nil
       else
         msg_impl = msg.impl
       end
       perform_get(msg_impl)
       msg.post_decode unless msg.nil?
       return incoming_tracker
     end

     private

     def perform_get(msg) # :nodoc:
       Cproton.pn_messenger_get(@impl, msg)
     end

     public

     # Receives up to limit messages into the incoming queue.  If no value
     # for limit is supplied, this call will receive as many messages as it
     # can buffer internally.  If the Messenger is in blocking mode, this
     # call will block until at least one Message is available in the
     # incoming queue.
     #
     # Options ====
     #
     # * limit - the maximum number of messages to receive
     #
     def receive(limit = -1)
       Cproton.pn_messenger_recv(@impl, limit)
     end

     # Returns true if the messenger is currently receiving data.
     def receiving?
       Cproton.pn_messenger_receiving(@impl)
     end

     # Attempts interrupting of the messenger thread.
     #
     # The Messenger interface is single-threaded, and this is the only
     # function intended to be called from outside of is thread.
     #
     # Call this from a non-Messenger thread to interrupt it while it
     # is blocking. This will cause a ::InterruptError to be raised.
     #
     # If there is no currently blocking call, then the next blocking
     # call will be affected, even if it is within the same thread that
     # originated the interrupt.
     #
     def interrupt
       Cproton.pn_messenger_interrupt(@impl)
     end

     # Sends or receives any outstanding messages queued for a Messenger.
     #
     # This will block for the indicated timeout.  This method may also do I/O
     # other than sending and receiving messages.  For example, closing
     # connections after stop() has been called.
     #
     def work(timeout=-1)
       err = Cproton.pn_messenger_work(@impl, timeout)
       if (err == Cproton::PN_TIMEOUT) then
         return false
       else
         check_for_error(err)
         return true
       end
     end

     # Returns the number messages in the outgoing queue that have not been
     # transmitted.
     #
     def outgoing
       Cproton.pn_messenger_outgoing(@impl)
     end

     # Returns the number of messages in the incoming queue that have not
     # been retrieved.
     #
     def incoming
       Cproton.pn_messenger_incoming(@impl)
     end

     # Adds a routing rule to the Messenger's internal routing table.
     #
     # The route procedure may be used to influence how a Messenger will
     # internally treat a given address or class of addresses. Every call
     # to the route procedure will result in Messenger appending a routing
     # rule to its internal routing table.
     #
     # Whenever a Message is presented to a Messenger for delivery, it
     # will match the address of this message against the set of routing
     # rules in order. The first rule to match will be triggered, and
     # instead of routing based on the address presented in the message,
     # the Messenger will route based on the address supplied in the rule.
     #
     # The pattern matching syntax supports two types of matches, a '%'
     # will match any character except a '/', and a '*' will match any
     # character including a '/'.
     #
     # A routing address is specified as a normal AMQP address, however it
     # may additionally use substitution variables from the pattern match
     # that triggered the rule.
     #
     # ==== Arguments
     #
     # * pattern - the address pattern
     # * address - the target address
     #
     # ==== Examples
     #
     #   # route messages sent to foo to the destionaty amqp://foo.com
     #   messenger.route("foo", "amqp://foo.com")
     #
     #   # any message to foobar will be routed to amqp://foo.com/bar
     #   messenger.route("foobar", "amqp://foo.com/bar")
     #
     #   # any message to bar/<path> will be routed to the same path within
     #   # the amqp://bar.com domain
     #   messenger.route("bar/*", "amqp://bar.com/$1")
     #
     #   # route all Message objects over TLS
     #   messenger.route("amqp:*", "amqps:$1")
     #
     #   # supply credentials for foo
     #   messenger.route("amqp://foo.com/*", "amqp://user:password@foo.com/$1")
     #
     #   # supply credentials for all domains
     #   messenger.route("amqp://*", "amqp://user:password@$1")
     #
     #   # route all addresses through a single proxy while preserving the
     #   # original destination
     #   messenger.route("amqp://%$/*", "amqp://user:password@proxy/$1/$2")
     #
     #   # route any address through a single broker
     #   messenger.route("*", "amqp://user:password@broker/$1")
     #
     def route(pattern, address)
       Cproton.pn_messenger_route(@impl, pattern, address)
     end

     # Similar to #route, except that the destination of
     # the Message is determined before the message address is rewritten.
     #
     # The outgoing address is only rewritten after routing has been
     # finalized.  If a message has an outgoing address of
     # "amqp://0.0.0.0:5678", and a rewriting rule that changes its
     # outgoing address to "foo", it will still arrive at the peer that
     # is listening on "amqp://0.0.0.0:5678", but when it arrives there,
     # the receiver will see its outgoing address as "foo".
     #
     # The default rewrite rule removes username and password from addresses
     # before they are transmitted.
     #
     # ==== Arguments
     #
     # * pattern - the outgoing address
     # * address - the target address
     #
     def rewrite(pattern, address)
       Cproton.pn_messenger_rewrite(@impl, pattern, address)
     end

     def selectable
       impl = Cproton.pn_messenger_selectable(@impl)

       # if we don't have any selectables, then return
       return nil if impl.nil?

       fd = Cproton.pn_selectable_get_fd(impl)

       selectable = @selectables[fd]
       if selectable.nil?
         selectable = Selectable.new(self, impl)
         @selectables[fd] = selectable
       end
       return selectable
     end

     # Returns a +Tracker+ for the message most recently sent via the put
     # method.
     #
     def outgoing_tracker
       impl = Cproton.pn_messenger_outgoing_tracker(@impl)
       return nil if impl == -1
       Tracker.new(impl)
     end

     # Returns a +Tracker+ for the most recently received message.
     #
     def incoming_tracker
       impl = Cproton.pn_messenger_incoming_tracker(@impl)
       return nil if impl == -1
       Tracker.new(impl)
     end

     # Signal the sender that you have acted on the Message
     # pointed to by the tracker.  If no tracker is supplied,
     # then all messages that have been returned by the get
     # method are accepted, except those that have already been
     # auto-settled by passing beyond your incoming window size.
     #
     # ==== Options
     #
     # * tracker - the tracker
     #
     def accept(tracker = nil)
       raise TypeError.new("invalid tracker: #{tracker}") unless tracker.nil? or valid_tracker?(tracker)
       if tracker.nil? then
         tracker = self.incoming_tracker
         flag = Cproton::PN_CUMULATIVE
       else
         flag = 0
       end
       Cproton.pn_messenger_accept(@impl, tracker.impl, flag)
     end

     # Rejects the incoming message identified by the tracker.
     # If no tracker is supplied, all messages that have been returned
     # by the get method are rejected, except those that have already
     # been auto-settled by passing beyond your outgoing window size.
     #
     # ==== Options
     #
     # * tracker - the tracker
     #
     def reject(tracker)
       raise TypeError.new("invalid tracker: #{tracker}") unless tracker.nil? or valid_tracker?(tracker)
       if tracker.nil? then
         tracker = self.incoming_tracker
         flag = Cproton::PN_CUMULATIVE
       else
         flag = 0
       end
       Cproton.pn_messenger_reject(@impl, tracker.impl, flag)
     end

     # Gets the last known remote state of the delivery associated with
     # the given tracker, as long as the Message is still within your
     # outgoing window. (Also works on incoming messages that are still
     # within your incoming queue. See TrackerStatus for details on the
     # values returned.
     #
     # ==== Options
     #
     # * tracker - the tracker
     #
     def status(tracker)
       raise TypeError.new("invalid tracker: #{tracker}") unless valid_tracker?(tracker)
       TrackerStatus.by_value(Cproton.pn_messenger_status(@impl, tracker.impl))
     end

     # Frees a Messenger from tracking the status associated
     # with a given tracker. If you don't supply a tracker, all
     # outgoing messages up to the most recent will be settled.
     #
     # ==== Options
     #
     # * tracker - the tracker
     #
     # ==== Examples
     #
     def settle(tracker)
       raise TypeError.new("invalid tracker: #{tracker}") unless valid_tracker?(tracker)
       if tracker.nil? then
         tracker = self.incoming_tracker
         flag = Cproton::PN_CUMULATIVE
       else
         flag = 0
       end
       Cproton.pn_messenger_settle(@impl, tracker.impl, flag)
     end

     # Sets the incoming window.
     #
     # The Messenger will track the remote status of this many incoming
     # deliveries after they have been accepted or rejected.
     #
     # Messages enter this window only when you take them into your application
     # using get().  If your incoming window size is n, and you get n+1 messages
     # without explicitly accepting or rejecting the oldest message, then the
     # message that passes beyond the edge of the incoming window will be
     # assigned the default disposition of its link.
     #
     # ==== Options
     #
     # * window - the window size
     #
     def incoming_window=(window)
       raise TypeError.new("invalid window: #{window}") unless valid_window?(window)
       Cproton.pn_messenger_set_incoming_window(@impl, window)
     end

     # Returns the incoming window.
     #
     def incoming_window
       Cproton.pn_messenger_get_incoming_window(@impl)
     end

     # Sets the outgoing window.
     #
     # The Messenger will track the remote status of this many outgoing
     # deliveries after calling send.
     # A Message enters this window when you call the put() method with the
     # message.  If your outgoing window size is n, and you call put n+1
     # times, status information will no longer be available for the
     # first message.
     #
     # ==== Options
     #
     # * window - the window size
     #
     def outgoing_window=(window)
       raise TypeError.new("invalid window: #{window}") unless valid_window?(window)
       Cproton.pn_messenger_set_outgoing_window(@impl, window)
     end

     # Returns the outgoing window.
     #
     def outgoing_window
       Cproton.pn_messenger_get_outgoing_window(@impl)
     end

     # Unregisters a selectable object.
     def unregister_selectable(fileno) # :nodoc:
       @selectables.delete(fileno)
     end

     private

     def valid_tracker?(tracker)
       !tracker.nil? && tracker.is_a?(Tracker)
     end

     def valid_window?(window)
       !window.nil? && [Float, Fixnum].include?(window.class)
     end

   end

 end
	#
	# Licensed to the Apache Software Foundation (ASF) under one
	# or more contributor license agreements. See the NOTICE file
	# distributed with this work for additional information
	# regarding copyright ownership. The ASF licenses this file
	# to you under the Apache License, Version 2.0 (the
	# "License"); you may not use this file except in compliance
	# with the License. You may obtain a copy of the License at
	#
	# http://www.apache.org/licenses/LICENSE-2.0
	#
	# Unless required by applicable law or agreed to in writing,
	# software distributed under the License is distributed on an
	# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	# KIND, either express or implied. See the License for the
	# specific language governing permissions and limitations
	# under the License.
	#

	module Qpid::Proton::Messenger

	# The +Messenger+ class defines a high level interface for
	# sending and receiving Messages. Every Messenger contains
	# a single logical queue of incoming messages and a single
	# logical queue of outgoing messages. These messages in these
	# queues may be destined for, or originate from, a variety of
	# addresses.
	#
	# The messenger interface is single-threaded. All methods
	# except one ( #interrupt ) are intended to be used from within
	# the messenger thread.
	#
	# === Sending & Receiving Messages
	#
	# The Messenger class works in conjuction with the Message class. The
	# Message class is a mutable holder of message content.
	#
	# The put method copies its Message to the outgoing queue, and may
	# send queued messages if it can do so without blocking. The send
	# method blocks until it has sent the requested number of messages,
	# or until a timeout interrupts the attempt.
	#
	# Similarly, the recv method receives messages into the incoming
	# queue, and may block as it attempts to receive the requested number
	# of messages, or until timeout is reached. It may receive fewer
	# than the requested number. The get method pops the
	# eldest Message off the incoming queue and copies it into the Message
	# object that you supply. It will not block.
	#
	# The blocking attribute allows you to turn off blocking behavior entirely,
	# in which case send and recv will do whatever they can without
	# blocking, and then return. You can then look at the number
	# of incoming and outgoing messages to see how much outstanding work
	# still remains.
	#
	class Messenger

	include Qpid::Proton::Util::ErrorHandler

	can_raise_error [:send, :receive, :password=, :start, :stop,
	:perform_put, :perform_get, :interrupt,
	:route, :rewrite, :accept, :reject,
	:incoming_window=, :outgoing_window=]

	# Creates a new +Messenger+.
	#
	# The +name+ parameter is optional. If one is not provided then
	# a unique name is generated.
	#
	# ==== Options
	#
	# * name - the name (def. nil)
	#
	def initialize(name = nil)
	@impl = Cproton.pn_messenger(name)
	@selectables = {}
	ObjectSpace.define_finalizer(self, self.class.finalize!(@impl))
	end

	def self.finalize!(impl) # :nodoc:
	proc {
	Cproton.pn_messenger_free(impl)
	}
	end

	# Returns the name.
	#
	def name
	Cproton.pn_messenger_name(@impl)
	end

	# This property contains the password for the Messenger.private_key
	# file, or +nil+ if the file is not encrypted.
	#
	# ==== Arguments
	#
	# * password - the password
	#
	def password=(password)
	Cproton.pn_messenger_set_password(@impl, password)
	end

	# Returns the password property for the Messenger.private_key file.
	#
	def password
	Cproton.pn_messenger_get_password(@impl)
	end

	# Sets the timeout period, in milliseconds.
	#
	# A negative timeout period implies an infinite timeout.
	#
	# ==== Options
	#
	# * timeout - the timeout period
	#
	def timeout=(timeout)
	raise TypeError.new("invalid timeout: #{timeout}") if timeout.nil?
	Cproton.pn_messenger_set_timeout(@impl, timeout)
	end

	# Returns the timeout period
	#
	def timeout
	Cproton.pn_messenger_get_timeout(@impl)
	end

	# Returns true if blocking mode is enabled.
	#
	# Enable or disable blocking behavior during message sending
	# and receiving. This affects every blocking call, with the
	# exception of work(). Currently, the affected calls are
	# send, recv, and stop.
	def blocking?
	Cproton.pn_messenger_is_blocking(@impl)
	end

	# Sets the blocking mode.
	def blocking=(blocking)
	Cproton.pn_messenger_set_blocking(@impl, blocking)
	end

	# Returns true if passive mode is enabled.
	#
	def passive?
	Cproton.pn_messenger_is_passive(@impl)
	end

	# Turns passive mode on or off.
	#
	# When set to passive mode, Messenger will not attempt to perform I/O
	# operations internally. In this mode it is necesssary to use the
	# Selectable type to drive any I/O needed to perform requestioned
	# actions.
	#
	# In this mode Messenger will never block.
	#
	def passive=(mode)
	Cproton.pn_messenger_set_passive(@impl, mode)
	end

	def deadline
	tstamp = Cproton.pn_messenger_deadline(@impl)
	return tstamp / 1000.0 unless tstamp.nil?
	end

	# Reports whether an error occurred.
	#
	def error?
	!Cproton.pn_messenger_errno(@impl).zero?
	end

	# Returns the most recent error number.
	#
	def errno
	Cproton.pn_messenger_errno(@impl)
	end

	# Returns the most recent error message.
	#
	def error
	Cproton.pn_error_text(Cproton.pn_messenger_error(@impl))
	end

	# Clears the current error state.
	#
	def clear_error
	error = Cproton.pn_messenger_error(@impl)
	unless error.nil?
	Cproton.pn_error_clear(error)
	end
	end

	# Currently a no-op placeholder.
	# For future compatibility, do not send or recv messages
	# before starting the +Messenger+.
	#
	def start
	Cproton.pn_messenger_start(@impl)
	end

	# Stops the +Messenger+, preventing it from sending or receiving
	# any more messages.
	#
	def stop
	Cproton.pn_messenger_stop(@impl)
	end

	# Returns true if a Messenger is in the stopped state.
	# This function does not block.
	#
	def stopped?
	Cproton.pn_messenger_stopped(@impl)
	end

	# Subscribes the Messenger to messages originating from the
	# specified source. The source is an address as specified in the
	# Messenger introduction with the following addition. If the
	# domain portion of the address begins with the '~' character, the
	# Messenger will interpret the domain as host/port, bind to it,
	# and listen for incoming messages. For example "~0.0.0.0",
	# "amqp://~0.0.0.0" will all bind to any local interface and
	# listen for incoming messages. An address of "amqps://~0.0.0.0"
	# will only permit incoming SSL connections.
	#
	# ==== Options
	#
	# * address - the source address to be subscribe
	# * timeout - an optional time-to-live value, in seconds, for the
	# subscription
	#
	def subscribe(address, timeout=0)
	raise TypeError.new("invalid address: #{address}") if address.nil?
	subscription = Cproton.pn_messenger_subscribe_ttl(@impl, address, timeout)
	raise Qpid::Proton::ProtonError.new("Subscribe failed") if subscription.nil?
	Subscription.new(subscription)
	end

	# Path to a certificate file for the +Messenger+.
	#
	# This certificate is used when the +Messenger+ accepts or establishes
	# SSL/TLS connections. This property must be specified for the
	# Messenger to accept incoming SSL/TLS connections and to establish
	# client authenticated outgoing SSL/TLS connection. Non client authenticated
	# outgoing SSL/TLS connections do not require this property.
	#
	# ==== Options
	#
	# * certificate - the certificate
	#
	def certificate=(certificate)
	Cproton.pn_messenger_set_certificate(@impl, certificate)
	end

	# Returns the path to a certificate file.
	#
	def certificate
	Cproton.pn_messenger_get_certificate(@impl)
	end

	# Path to a private key file for the +Messenger+.
	#
	# The property must be specified for the +Messenger+ to accept incoming
	# SSL/TLS connections and to establish client authenticated outgoing
	# SSL/TLS connections. Non client authenticated SSL/TLS connections
	# do not require this property.
	#
	# ==== Options
	#
	# * key - the key file
	#
	def private_key=(key)
	Cproton.pn_messenger_set_private_key(@impl, key)
	end

	# Returns the path to a private key file.
	#
	def private_key
	Cproton.pn_messenger_get_private_key(@impl)
	end

	# A path to a database of trusted certificates for use in verifying the
	# peer on an SSL/TLS connection. If this property is +nil+, then the
	# peer will not be verified.
	#
	# ==== Options
	#
	# * certificates - the certificates path
	#
	def trusted_certificates=(certificates)
	Cproton.pn_messenger_set_trusted_certificates(@impl,certificates)
	end

	# The path to the databse of trusted certificates.
	#
	def trusted_certificates
	Cproton.pn_messenger_get_trusted_certificates(@impl)
	end

	# Places the content contained in the message onto the outgoing
	# queue of the Messenger.
	#
	# This method will never block, however it will send any unblocked
	# Messages in the outgoing queue immediately and leave any blocked
	# Messages remaining in the outgoing queue.
	# The send call may then be used to block until the outgoing queue
	# is empty. The outgoing attribute may be used to check the depth
	# of the outgoing queue.
	#
	# ==== Options
	#
	# * message - the message
	#
	def put(message)
	if message.nil?
	raise TypeError.new("invalid message: #{message}")
	end
	unless message.kind_of?(Qpid::Proton::Message)
	raise ::ArgumentError.new("invalid message type: #{message.class}")
	end
	# encode the message first
	message.pre_encode
	perform_put(message)
	return outgoing_tracker
	end

	private

	def perform_put(message) # :nodoc:
	Cproton.pn_messenger_put(@impl, message.impl)
	end

	public


	# This call will block until the indicated number of messages
	# have been sent, or until the operation times out.
	# If n is -1 this call will block until all outgoing messages
	# have been sent. If n is 0 then this call will send whatever
	# it can without blocking.
	#
	def send(n = -1)
	Cproton.pn_messenger_send(@impl, n)
	end

	# Moves the message from the head of the incoming message queue into
	# the supplied message object. Any content in the supplied message
	# will be overwritten.
	# A tracker for the incoming Message is returned. The tracker can
	# later be used to communicate your acceptance or rejection of the
	# Message.
	#
	# If no message is provided in the argument, then one is created. In
	# either case, the one returned will be the fetched message.
	#
	# ==== Options
	#
	# * msg - the (optional) +Message+ instance to be used
	#
	def get(msg = nil)
	msg_impl = nil
	if msg.nil? then
	msg_impl = nil
	else
	msg_impl = msg.impl
	end
	perform_get(msg_impl)
	msg.post_decode unless msg.nil?
	return incoming_tracker
	end

	private

	def perform_get(msg) # :nodoc:
	Cproton.pn_messenger_get(@impl, msg)
	end

	public

	# Receives up to limit messages into the incoming queue. If no value
	# for limit is supplied, this call will receive as many messages as it
	# can buffer internally. If the Messenger is in blocking mode, this
	# call will block until at least one Message is available in the
	# incoming queue.
	#
	# Options ====
	#
	# * limit - the maximum number of messages to receive
	#
	def receive(limit = -1)
	Cproton.pn_messenger_recv(@impl, limit)
	end

	# Returns true if the messenger is currently receiving data.
	def receiving?
	Cproton.pn_messenger_receiving(@impl)
	end

	# Attempts interrupting of the messenger thread.
	#
	# The Messenger interface is single-threaded, and this is the only
	# function intended to be called from outside of is thread.
	#
	# Call this from a non-Messenger thread to interrupt it while it
	# is blocking. This will cause a ::InterruptError to be raised.
	#
	# If there is no currently blocking call, then the next blocking
	# call will be affected, even if it is within the same thread that
	# originated the interrupt.
	#
	def interrupt
	Cproton.pn_messenger_interrupt(@impl)
	end

	# Sends or receives any outstanding messages queued for a Messenger.
	#
	# This will block for the indicated timeout. This method may also do I/O
	# other than sending and receiving messages. For example, closing
	# connections after stop() has been called.
	#
	def work(timeout=-1)
	err = Cproton.pn_messenger_work(@impl, timeout)
	if (err == Cproton::PN_TIMEOUT) then
	return false
	else
	check_for_error(err)
	return true
	end
	end

	# Returns the number messages in the outgoing queue that have not been
	# transmitted.
	#
	def outgoing
	Cproton.pn_messenger_outgoing(@impl)
	end

	# Returns the number of messages in the incoming queue that have not
	# been retrieved.
	#
	def incoming
	Cproton.pn_messenger_incoming(@impl)
	end

	# Adds a routing rule to the Messenger's internal routing table.
	#
	# The route procedure may be used to influence how a Messenger will
	# internally treat a given address or class of addresses. Every call
	# to the route procedure will result in Messenger appending a routing
	# rule to its internal routing table.
	#
	# Whenever a Message is presented to a Messenger for delivery, it
	# will match the address of this message against the set of routing
	# rules in order. The first rule to match will be triggered, and
	# instead of routing based on the address presented in the message,
	# the Messenger will route based on the address supplied in the rule.
	#
	# The pattern matching syntax supports two types of matches, a '%'
	# will match any character except a '/', and a '*' will match any
	# character including a '/'.
	#
	# A routing address is specified as a normal AMQP address, however it
	# may additionally use substitution variables from the pattern match
	# that triggered the rule.
	#
	# ==== Arguments
	#
	# * pattern - the address pattern
	# * address - the target address
	#
	# ==== Examples
	#
	# # route messages sent to foo to the destionaty amqp://foo.com
	# messenger.route("foo", "amqp://foo.com")
	#
	# # any message to foobar will be routed to amqp://foo.com/bar
	# messenger.route("foobar", "amqp://foo.com/bar")
	#
	# # any message to bar/<path> will be routed to the same path within
	# # the amqp://bar.com domain
	# messenger.route("bar/*", "amqp://bar.com/$1")
	#
	# # route all Message objects over TLS
	# messenger.route("amqp:*", "amqps:$1")
	#
	# # supply credentials for foo
	# messenger.route("amqp://foo.com/*", "amqp://user:password@foo.com/$1")
	#
	# # supply credentials for all domains
	# messenger.route("amqp://*", "amqp://user:password@$1")
	#
	# # route all addresses through a single proxy while preserving the
	# # original destination
	# messenger.route("amqp://%$/*", "amqp://user:password@proxy/$1/$2")
	#
	# # route any address through a single broker
	# messenger.route("*", "amqp://user:password@broker/$1")
	#
	def route(pattern, address)
	Cproton.pn_messenger_route(@impl, pattern, address)
	end

	# Similar to #route, except that the destination of
	# the Message is determined before the message address is rewritten.
	#
	# The outgoing address is only rewritten after routing has been
	# finalized. If a message has an outgoing address of
	# "amqp://0.0.0.0:5678", and a rewriting rule that changes its
	# outgoing address to "foo", it will still arrive at the peer that
	# is listening on "amqp://0.0.0.0:5678", but when it arrives there,
	# the receiver will see its outgoing address as "foo".
	#
	# The default rewrite rule removes username and password from addresses
	# before they are transmitted.
	#
	# ==== Arguments
	#
	# * pattern - the outgoing address
	# * address - the target address
	#
	def rewrite(pattern, address)
	Cproton.pn_messenger_rewrite(@impl, pattern, address)
	end

	def selectable
	impl = Cproton.pn_messenger_selectable(@impl)

	# if we don't have any selectables, then return
	return nil if impl.nil?

	fd = Cproton.pn_selectable_get_fd(impl)

	selectable = @selectables[fd]
	if selectable.nil?
	selectable = Selectable.new(self, impl)
	@selectables[fd] = selectable
	end
	return selectable
	end

	# Returns a +Tracker+ for the message most recently sent via the put
	# method.
	#
	def outgoing_tracker
	impl = Cproton.pn_messenger_outgoing_tracker(@impl)
	return nil if impl == -1
	Tracker.new(impl)
	end

	# Returns a +Tracker+ for the most recently received message.
	#
	def incoming_tracker
	impl = Cproton.pn_messenger_incoming_tracker(@impl)
	return nil if impl == -1
	Tracker.new(impl)
	end

	# Signal the sender that you have acted on the Message
	# pointed to by the tracker. If no tracker is supplied,
	# then all messages that have been returned by the get
	# method are accepted, except those that have already been
	# auto-settled by passing beyond your incoming window size.
	#
	# ==== Options
	#
	# * tracker - the tracker
	#
	def accept(tracker = nil)
	raise TypeError.new("invalid tracker: #{tracker}") unless tracker.nil? or valid_tracker?(tracker)
	if tracker.nil? then
	tracker = self.incoming_tracker
	flag = Cproton::PN_CUMULATIVE
	else
	flag = 0
	end
	Cproton.pn_messenger_accept(@impl, tracker.impl, flag)
	end

	# Rejects the incoming message identified by the tracker.
	# If no tracker is supplied, all messages that have been returned
	# by the get method are rejected, except those that have already
	# been auto-settled by passing beyond your outgoing window size.
	#
	# ==== Options
	#
	# * tracker - the tracker
	#
	def reject(tracker)
	raise TypeError.new("invalid tracker: #{tracker}") unless tracker.nil? or valid_tracker?(tracker)
	if tracker.nil? then
	tracker = self.incoming_tracker
	flag = Cproton::PN_CUMULATIVE
	else
	flag = 0
	end
	Cproton.pn_messenger_reject(@impl, tracker.impl, flag)
	end

	# Gets the last known remote state of the delivery associated with
	# the given tracker, as long as the Message is still within your
	# outgoing window. (Also works on incoming messages that are still
	# within your incoming queue. See TrackerStatus for details on the
	# values returned.
	#
	# ==== Options
	#
	# * tracker - the tracker
	#
	def status(tracker)
	raise TypeError.new("invalid tracker: #{tracker}") unless valid_tracker?(tracker)
	TrackerStatus.by_value(Cproton.pn_messenger_status(@impl, tracker.impl))
	end

	# Frees a Messenger from tracking the status associated
	# with a given tracker. If you don't supply a tracker, all
	# outgoing messages up to the most recent will be settled.
	#
	# ==== Options
	#
	# * tracker - the tracker
	#
	# ==== Examples
	#
	def settle(tracker)
	raise TypeError.new("invalid tracker: #{tracker}") unless valid_tracker?(tracker)
	if tracker.nil? then
	tracker = self.incoming_tracker
	flag = Cproton::PN_CUMULATIVE
	else
	flag = 0
	end
	Cproton.pn_messenger_settle(@impl, tracker.impl, flag)
	end

	# Sets the incoming window.
	#
	# The Messenger will track the remote status of this many incoming
	# deliveries after they have been accepted or rejected.
	#
	# Messages enter this window only when you take them into your application
	# using get(). If your incoming window size is n, and you get n+1 messages
	# without explicitly accepting or rejecting the oldest message, then the
	# message that passes beyond the edge of the incoming window will be
	# assigned the default disposition of its link.
	#
	# ==== Options
	#
	# * window - the window size
	#
	def incoming_window=(window)
	raise TypeError.new("invalid window: #{window}") unless valid_window?(window)
	Cproton.pn_messenger_set_incoming_window(@impl, window)
	end

	# Returns the incoming window.
	#
	def incoming_window
	Cproton.pn_messenger_get_incoming_window(@impl)
	end

	# Sets the outgoing window.
	#
	# The Messenger will track the remote status of this many outgoing
	# deliveries after calling send.
	# A Message enters this window when you call the put() method with the
	# message. If your outgoing window size is n, and you call put n+1
	# times, status information will no longer be available for the
	# first message.
	#
	# ==== Options
	#
	# * window - the window size
	#
	def outgoing_window=(window)
	raise TypeError.new("invalid window: #{window}") unless valid_window?(window)
	Cproton.pn_messenger_set_outgoing_window(@impl, window)
	end

	# Returns the outgoing window.
	#
	def outgoing_window
	Cproton.pn_messenger_get_outgoing_window(@impl)
	end

	# Unregisters a selectable object.
	def unregister_selectable(fileno) # :nodoc:
	@selectables.delete(fileno)
	end

	private

	def valid_tracker?(tracker)
	!tracker.nil? && tracker.is_a?(Tracker)
	end

	def valid_window?(window)
	!window.nil? && [Float, Fixnum].include?(window.class)
	end

	end

	end