#
# 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.
#

"""
A high level messaging API for python.

Areas that still need work:

  - definition of the arguments for L{Session.sender} and L{Session.receiver}
  - standard L{Message} properties
  - L{Message} content encoding
  - protocol negotiation/multiprotocol impl
"""

from logging import getLogger
from math import ceil
from qpid.codec010 import StringCodec
from qpid.concurrency import synchronized, Waiter, Condition
from qpid.datatypes import Serial, uuid4
from qpid.messaging.constants import *
from qpid.messaging.exceptions import *
from qpid.messaging.message import *
from qpid.ops import PRIMITIVE
from qpid.util import default, URL
from threading import Thread, RLock

log = getLogger("qpid.messaging")

static = staticmethod

class Endpoint(object):
  """
  Base class for all endpoint objects types.
  @undocumented: __init__, __setattr__
  """
  def __init__(self):
    self._async_exception_notify_handler = None
    self.error = None

  def _ecwait(self, predicate, timeout=None):
    result = self._ewait(lambda: self.closed or predicate(), timeout)
    self.check_closed()
    return result

  @synchronized
  def set_async_exception_notify_handler(self, handler):
    """
    Register a callable that will be invoked when the driver thread detects an
    error on the Endpoint. The callable is invoked with the instance of the
    Endpoint object passed as the first argument. The second argument is an
    Exception instance describing the failure.

    @param handler: invoked by the driver thread when an error occurs.
    @type handler: callable object taking an Endpoint and an Exception as
    arguments.
    @return: None
    @note: The exception will also be raised the next time the application
    invokes one of the blocking messaging APIs.
    @warning: B{Use with caution} This callback is invoked in the context of
    the driver thread. It is B{NOT} safe to call B{ANY} of the messaging APIs
    from within this callback. This includes any of the Endpoint's methods. The
    intent of the handler is to provide an efficient way to notify the
    application that an exception has occurred in the driver thread. This can
    be useful for those applications that periodically poll the messaging layer
    for events. In this case the callback can be used to schedule a task that
    retrieves the error using the Endpoint's get_error() or check_error()
    methods.
    """
    self._async_exception_notify_handler = handler

  def __setattr__(self, name, value):
    """
    Intercept any attempt to set the endpoint error flag and invoke the
    callback if registered.
    """
    super(Endpoint, self).__setattr__(name, value)
    if name == 'error' and value is not None:
        if self._async_exception_notify_handler:
            self._async_exception_notify_handler(self, value)


class Connection(Endpoint):

  """
  A Connection manages a group of L{Sessions<Session>} and connects
  them with a remote endpoint.
  """

  @static
  def establish(url=None, timeout=None, **options):
    """
    Constructs a L{Connection} with the supplied parameters and opens
    it.
    """
    conn = Connection(url, **options)
    conn.open(timeout=timeout)
    return conn

  def __init__(self, url=None, **options):
    """
    Creates a connection. A newly created connection must be opened
    with the Connection.open() method before it can be used.

    @type url: str
    @param url: [ <username> [ / <password> ] @ ] <host> [ : <port> ]
    @type host: str
    @param host: the name or ip address of the remote host (overriden by url)
    @type port: int
    @param port: the port number of the remote host (overriden by url)
    @type transport: str
    @param transport: one of tcp, tcp+tls, or ssl (alias for tcp+tls)
    @type heartbeat: int
    @param heartbeat: heartbeat interval in seconds

    @type username: str
    @param username: the username for authentication (overriden by url)
    @type password: str
    @param password: the password for authentication (overriden by url)
    @type sasl_mechanisms: str
    @param sasl_mechanisms: space separated list of permitted sasl mechanisms
    @type sasl_service: str
    @param sasl_service: the service name if needed by the SASL mechanism in use
    @type sasl_min_ssf: int
    @param sasl_min_ssf: the minimum acceptable security strength factor
    @type sasl_max_ssf: int
    @param sasl_max_ssf: the maximum acceptable security strength factor

    @type reconnect: bool
    @param reconnect: enable/disable automatic reconnect
    @type reconnect_timeout: float
    @param reconnect_timeout: total time to attempt reconnect
    @type reconnect_interval_min: float
    @param reconnect_interval_min: minimum interval between reconnect attempts
    @type reconnect_interval_max: float
    @param reconnect_interval_max: maximum interval between reconnect attempts
    @type reconnect_interval: float
    @param reconnect_interval: set both min and max reconnect intervals
    @type reconnect_limit: int
    @param reconnect_limit: limit the total number of reconnect attempts
    @type reconnect_urls: list[str]
    @param reconnect_urls: list of backup hosts specified as urls

    @type address_ttl: float
    @param address_ttl: time until cached address resolution expires

    @type ssl_keyfile: str
    @param ssl_keyfile: file with client's private key (PEM format)
    @type ssl_certfile: str
    @param ssl_certfile: file with client's public (eventually priv+pub) key (PEM format)
    @type ssl_trustfile: str
    @param ssl_trustfile: file trusted certificates to validate the server
    @type ssl_skip_hostname_check: bool
    @param ssl_skip_hostname_check: disable verification of hostname in
    certificate. Use with caution - disabling hostname checking leaves you
    vulnerable to Man-in-the-Middle attacks.

    @rtype: Connection
    @return: a disconnected Connection
    """
    super(Connection, self).__init__()
    # List of all attributes
    opt_keys = ['host', 'transport', 'port', 'heartbeat', 'username', 'password', 'sasl_mechanisms', 'sasl_service', 'sasl_min_ssf', 'sasl_max_ssf', 'reconnect', 'reconnect_timeout', 'reconnect_interval', 'reconnect_interval_min', 'reconnect_interval_max', 'reconnect_limit', 'reconnect_urls', 'reconnect_log', 'address_ttl', 'tcp_nodelay', 'ssl_keyfile', 'ssl_certfile', 'ssl_trustfile', 'ssl_skip_hostname_check', 'client_properties', 'protocol' ]
    # Create all attributes on self and set to None.
    for key in opt_keys:
        setattr(self, key, None)
    # Get values from options, check for invalid options
    for (key, value) in options.iteritems():
        if key in opt_keys:
            setattr(self, key, value)
        else:
            raise ConnectionError("Unknown connection option %s with value %s" %(key, value))

    # Now handle items that need special treatment or have speical defaults:
    if self.host:
        url = default(url, self.host)
    if isinstance(url, basestring):
        url = URL(url)
    self.host = url.host

    if self.transport is None:
      if url.scheme == url.AMQP:
        self.transport = "tcp"
      elif url.scheme == url.AMQPS:
        self.transport = "ssl"
      else:
        self.transport = "tcp"
    if self.transport in ("ssl", "tcp+tls"):
      self.port = default(url.port, default(self.port, AMQPS_PORT))
    else:
      self.port = default(url.port, default(self.port, AMQP_PORT))

    if self.protocol and self.protocol != "amqp0-10":
        raise ConnectionError("Connection option 'protocol' value '" + value + "' unsupported (must be amqp0-10)")
      
    self.username = default(url.user, self.username)
    self.password = default(url.password, self.password)
    self.auth_username = None
    self.sasl_service = default(self.sasl_service, "qpidd")

    self.reconnect = default(self.reconnect, False)
    self.reconnect_interval_min = default(self.reconnect_interval_min,
                                          default(self.reconnect_interval, 1))
    self.reconnect_interval_max = default(self.reconnect_interval_max,
                                          default(self.reconnect_interval, 2*60))
    self.reconnect_urls = default(self.reconnect_urls, [])
    self.reconnect_log = default(self.reconnect_log, True)

    self.address_ttl = default(self.address_ttl, 60)
    self.tcp_nodelay = default(self.tcp_nodelay, False)

    self.ssl_keyfile = default(self.ssl_keyfile, None)
    self.ssl_certfile = default(self.ssl_certfile, None)
    self.ssl_trustfile = default(self.ssl_trustfile, None)
    # if ssl_skip_hostname_check was not explicitly set, this will be None
    self._ssl_skip_hostname_check_actual = options.get("ssl_skip_hostname_check")
    self.ssl_skip_hostname_check = default(self.ssl_skip_hostname_check, False)
    self.client_properties = default(self.client_properties, {})

    self.options = options


    self.id = str(uuid4())
    self.session_counter = 0
    self.sessions = {}
    self._open = False
    self._connected = False
    self._transport_connected = False
    self._lock = RLock()
    self._condition = Condition(self._lock)
    self._waiter = Waiter(self._condition)
    self._modcount = Serial(0)
    from driver import Driver
    self._driver = Driver(self)

  def _wait(self, predicate, timeout=None):
    return self._waiter.wait(predicate, timeout=timeout)

  def _wakeup(self):
    self._modcount += 1
    self._driver.wakeup()

  def check_error(self):
    if self.error:
      self._condition.gc()
      e = self.error
      if isinstance(e, ContentError):
          """ forget the content error. It will be
          raised this time but won't block future calls
          """
          self.error = None
      raise e

  def get_error(self):
    return self.error

  def _ewait(self, predicate, timeout=None):
    result = self._wait(lambda: self.error or predicate(), timeout)
    self.check_error()
    return result

  def check_closed(self):
    if not self._connected:
      self._condition.gc()
      raise ConnectionClosed()

  @synchronized
  def session(self, name=None, transactional=False):
    """
    Creates or retrieves the named session. If the name is omitted or
    None, then a unique name is chosen based on a randomly generated
    uuid.

    @type name: str
    @param name: the session name
    @rtype: Session
    @return: the named Session
    """

    if name is None:
      name = "%s:%s" % (self.id, self.session_counter)
      self.session_counter += 1
    else:
      name = "%s:%s" % (self.id, name)

    if self.sessions.has_key(name):
      return self.sessions[name]
    else:
      ssn = Session(self, name, transactional)
      self.sessions[name] = ssn
      self._wakeup()
      return ssn

  @synchronized
  def _remove_session(self, ssn):
    self.sessions.pop(ssn.name, 0)

  @synchronized
  def open(self, timeout=None):
    """
    Opens a connection.
    """
    if self._open:
      raise ConnectionError("already open")
    self._open = True
    if self.reconnect and self.reconnect_timeout > 0:
        timeout = self.reconnect_timeout
    self.attach(timeout=timeout)

  @synchronized
  def opened(self):
    """
    Return true if the connection is open, false otherwise.
    """
    return self._open

  @synchronized
  def attach(self, timeout=None):
    """
    Attach to the remote endpoint.
    """
    if not self._connected:
      self._connected = True
      self._driver.start()
      self._wakeup()
    if not self._ewait(lambda: self._transport_connected and not self._unlinked(), timeout=timeout):
      self.reconnect = False
      raise Timeout("Connection attach timed out")

  def _unlinked(self):
    return [l
            for ssn in self.sessions.values()
            if not (ssn.error or ssn.closed)
            for l in ssn.senders + ssn.receivers
            if not (l.linked or l.error or l.closed)]

  @synchronized
  def detach(self, timeout=None):
    """
    Detach from the remote endpoint.
    """
    if self._connected:
      self._connected = False
      self._wakeup()
      cleanup = True
    else:
      cleanup = False
    try:
      if not self._wait(lambda: not self._transport_connected, timeout=timeout):
        raise Timeout("detach timed out")
    finally:
      if cleanup:
        self._driver.stop()
      self._condition.gc()

  @synchronized
  def attached(self):
    """
    Return true if the connection is attached, false otherwise.
    """
    return self._connected

  @synchronized
  def close(self, timeout=None):
    """
    Close the connection and all sessions.
    """
    try:
      for ssn in self.sessions.values():
        ssn.close(timeout=timeout)
    finally:
      self.detach(timeout=timeout)
      self._open = False


class Session(Endpoint):

  """
  Sessions provide a linear context for sending and receiving
  L{Messages<Message>}. L{Messages<Message>} are sent and received
  using the L{Sender.send} and L{Receiver.fetch} methods of the
  L{Sender} and L{Receiver} objects associated with a Session.

  Each L{Sender} and L{Receiver} is created by supplying either a
  target or source address to the L{sender} and L{receiver} methods of
  the Session. The address is supplied via a string syntax documented
  below.

  Addresses
  =========

  An address identifies a source or target for messages. In its
  simplest form this is just a name. In general a target address may
  also be used as a source address, however not all source addresses
  may be used as a target, e.g. a source might additionally have some
  filtering criteria that would not be present in a target.

  A subject may optionally be specified along with the name. When an
  address is used as a target, any subject specified in the address is
  used as the default subject of outgoing messages for that target.
  When an address is used as a source, any subject specified in the
  address is pattern matched against the subject of available messages
  as a filter for incoming messages from that source.

  The options map contains additional information about the address
  including:

    - policies for automatically creating, and deleting the node to
      which an address refers

    - policies for asserting facts about the node to which an address
      refers

    - extension points that can be used for sender/receiver
      configuration

  Mapping to AMQP 0-10
  --------------------
  The name is resolved to either an exchange or a queue by querying
  the broker.

  The subject is set as a property on the message. Additionally, if
  the name refers to an exchange, the routing key is set to the
  subject.

  Syntax
  ------
  The following regular expressions define the tokens used to parse
  addresses::
    LBRACE: \\{
    RBRACE: \\}
    LBRACK: \\[
    RBRACK: \\]
    COLON:  :
    SEMI:   ;
    SLASH:  /
    COMMA:  ,
    NUMBER: [+-]?[0-9]*\\.?[0-9]+
    ID:     [a-zA-Z_](?:[a-zA-Z0-9_-]*[a-zA-Z0-9_])?
    STRING: "(?:[^\\\\"]|\\\\.)*"|\'(?:[^\\\\\']|\\\\.)*\'
    ESC:    \\\\[^ux]|\\\\x[0-9a-fA-F][0-9a-fA-F]|\\\\u[0-9a-fA-F][0-9a-fA-F][0-9a-fA-F][0-9a-fA-F]
    SYM:    [.#*%@$^!+-]
    WSPACE: [ \\n\\r\\t]+

  The formal grammar for addresses is given below::
    address = name [ "/" subject ] [ ";" options ]
       name = ( part | quoted )+
    subject = ( part | quoted | "/" )*
     quoted = STRING / ESC
       part = LBRACE / RBRACE / COLON / COMMA / NUMBER / ID / SYM
    options = map
        map = "{" ( keyval ( "," keyval )* )? "}"
     keyval = ID ":" value
      value = NUMBER / STRING / ID / map / list
       list = "[" ( value ( "," value )* )? "]"

  This grammar resuls in the following informal syntax::

    <name> [ / <subject> ] [ ; <options> ]

  Where options is::

    { <key> : <value>, ... }

  And values may be:
    - numbers
    - single, double, or non quoted strings
    - maps (dictionaries)
    - lists

  Options
  -------
  The options map permits the following parameters::

    <name> [ / <subject> ] ; {
      create: always | sender | receiver | never,
      delete: always | sender | receiver | never,
      assert: always | sender | receiver | never,
      mode: browse | consume,
      node: {
        type: queue | topic,
        durable: True | False,
        x-declare: { ... <declare-overrides> ... },
        x-bindings: [<binding_1>, ... <binding_n>]
      },
      link: {
        name: <link-name>,
        durable: True | False,
        reliability: unreliable | at-most-once | at-least-once | exactly-once,
        x-declare: { ... <declare-overrides> ... },
        x-bindings: [<binding_1>, ... <binding_n>],
        x-subscribe: { ... <subscribe-overrides> ... }
      }
    }

  Bindings are specified as a map with the following options::

    {
      exchange: <exchange>,
      queue: <queue>,
      key: <key>,
      arguments: <arguments>
    }

  The create, delete, and assert policies specify who should perfom
  the associated action:

   - I{always}: the action will always be performed
   - I{sender}: the action will only be performed by the sender
   - I{receiver}: the action will only be performed by the receiver
   - I{never}: the action will never be performed (this is the default)

  The node-type is one of:

    - I{topic}: a topic node will default to the topic exchange,
      x-declare may be used to specify other exchange types
    - I{queue}: this is the default node-type

  The x-declare map permits protocol specific keys and values to be
  specified when exchanges or queues are declared. These keys and
  values are passed through when creating a node or asserting facts
  about an existing node.

  Examples
  --------
  A simple name resolves to any named node, usually a queue or a
  topic::

    my-queue-or-topic

  A simple name with a subject will also resolve to a node, but the
  presence of the subject will cause a sender using this address to
  set the subject on outgoing messages, and receivers to filter based
  on the subject::

    my-queue-or-topic/my-subject

  A subject pattern can be used and will cause filtering if used by
  the receiver. If used for a sender, the literal value gets set as
  the subject::

    my-queue-or-topic/my-*

  In all the above cases, the address is resolved to an existing node.
  If you want the node to be auto-created, then you can do the
  following. By default nonexistent nodes are assumed to be queues::

    my-queue; {create: always}

  You can customize the properties of the queue::

    my-queue; {create: always, node: {durable: True}}

  You can create a topic instead if you want::

    my-queue; {create: always, node: {type: topic}}

  You can assert that the address resolves to a node with particular
  properties::

    my-transient-topic; {
      assert: always,
      node: {
        type: topic,
        durable: False
      }
    }
 """

  def __init__(self, connection, name, transactional):
    super(Session, self).__init__()
    self.connection = connection
    self.name = name
    self.log_id = "%x" % id(self)

    self.transactional = transactional

    self.committing = False
    self.committed = True
    self.aborting = False
    self.aborted = False

    self.next_sender_id = 0
    self.senders = []
    self.next_receiver_id = 0
    self.receivers = []
    self.outgoing = []
    self.incoming = []
    self.unacked = []
    self.acked = []
    # XXX: I hate this name.
    self.ack_capacity = UNLIMITED

    self.closing = False
    self.closed = False

    self._lock = connection._lock
    self._msg_received_notify_handler = None

  def __repr__(self):
    return "<Session %s>" % self.name

  def _wait(self, predicate, timeout=None):
    return self.connection._wait(predicate, timeout=timeout)

  def _wakeup(self):
    self.connection._wakeup()

  def check_error(self):
    self.connection.check_error()
    if self.error:
      raise self.error

  def get_error(self):
    err = self.connection.get_error()
    if err:
      return err
    else:
      return self.error

  def _ewait(self, predicate, timeout=None):
    result = self.connection._ewait(lambda: self.error or predicate(), timeout)
    self.check_error()
    return result

  def check_closed(self):
    if self.closed:
      raise SessionClosed()

  def _notify_message_received(self, msg):
      self.incoming.append(msg)
      if self._msg_received_notify_handler:
          try:
              # new callback parameter: the Session
              self._msg_received_notify_handler(self)
          except TypeError:
              # backward compatibility with old API, no Session
              self._msg_received_notify_handler()

  @synchronized
  def sender(self, target, **options):
    """
    Creates a L{Sender} that may be used to send L{Messages<Message>}
    to the specified target.

    @type target: str
    @param target: the target to which messages will be sent
    @rtype: Sender
    @return: a new Sender for the specified target
    """
    target = _mangle(target)
    sender = Sender(self, self.next_sender_id, target, options)
    self.next_sender_id += 1
    self.senders.append(sender)
    if not self.closed and self.connection._connected:
      self._wakeup()
      try:
        sender._ewait(lambda: sender.linked)
      except LinkError, e:
        sender.close()
        raise e
    return sender

  @synchronized
  def receiver(self, source, **options):
    """
    Creates a receiver that may be used to fetch L{Messages<Message>}
    from the specified source.

    @type source: str
    @param source: the source of L{Messages<Message>}
    @rtype: Receiver
    @return: a new Receiver for the specified source
    """
    source = _mangle(source)
    receiver = Receiver(self, self.next_receiver_id, source, options)
    self.next_receiver_id += 1
    self.receivers.append(receiver)
    if not self.closed and self.connection._connected:
      self._wakeup()
      try:
        receiver._ewait(lambda: receiver.linked)
      except LinkError, e:
        receiver.close()
        raise e
    return receiver

  @synchronized
  def _count(self, predicate):
    result = 0
    for msg in self.incoming:
      if predicate(msg):
        result += 1
    return result

  def _peek(self, receiver):
    for msg in self.incoming:
      if msg._receiver == receiver:
        return msg

  def _pop(self, receiver):
    i = 0
    while i < len(self.incoming):
      msg = self.incoming[i]
      if msg._receiver == receiver:
        del self.incoming[i]
        return msg
      else:
        i += 1

  @synchronized
  def _get(self, receiver, timeout=None):
    if self._ewait(lambda: ((self._peek(receiver) is not None) or
                            self.closing or receiver.closed),
                   timeout):
      msg = self._pop(receiver)
      if msg is not None:
        msg._receiver.returned += 1
        self.unacked.append(msg)
        log.debug("RETR[%s]: %s", self.log_id, msg)
        return msg
    return None

  @synchronized
  def set_message_received_notify_handler(self, handler):
      """
      Register a callable that will be invoked when a Message arrives on the
      Session.

      @param handler: invoked by the driver thread when an error occurs.
      @type handler: a callable object taking a Session instance as its only
      argument
      @return: None

      @note: When using this method it is recommended to also register
      asynchronous error callbacks on all endpoint objects. Doing so will cause
      the application to be notified if an error is raised by the driver
      thread. This is necessary as after a driver error occurs the message received
      callback may never be invoked again. See
      L{Endpoint.set_async_exception_notify_handler}

      @warning: B{Use with caution} This callback is invoked in the context of
      the driver thread. It is B{NOT} safe to call B{ANY} of the public
      messaging APIs from within this callback, including any of the passed
      Session's methods. The intent of the handler is to provide an efficient
      way to notify the application that a message has arrived.  This can be
      useful for those applications that need to schedule a task to poll for
      received messages without blocking in the messaging API.  The scheduled
      task may then retrieve the message using L{next_receiver} and
      L{Receiver.fetch}
      """
      self._msg_received_notify_handler = handler

  @synchronized
  def set_message_received_handler(self, handler):
      """@deprecated: Use L{set_message_received_notify_handler} instead.
      """
      self._msg_received_notify_handler = handler

  @synchronized
  def next_receiver(self, timeout=None):
    if self._ecwait(lambda: self.incoming, timeout):
      return self.incoming[0]._receiver
    else:
      raise Empty

  @synchronized
  def acknowledge(self, message=None, disposition=None, sync=True):
    """
    Acknowledge the given L{Message}. If message is None, then all
    unacknowledged messages on the session are acknowledged.

    @type message: Message
    @param message: the message to acknowledge or None
    @type sync: boolean
    @param sync: if true then block until the message(s) are acknowledged
    """
    if message is None:
      messages = self.unacked[:]
    else:
      messages = [message]

    for m in messages:
      if self.ack_capacity is not UNLIMITED:
        if self.ack_capacity <= 0:
          # XXX: this is currently a SendError, maybe it should be a SessionError?
          raise InsufficientCapacity("ack_capacity = %s" % self.ack_capacity)
        self._wakeup()
        self._ecwait(lambda: len(self.acked) < self.ack_capacity)
      m._disposition = disposition
      self.unacked.remove(m)
      self.acked.append(m)

    self._wakeup()
    if sync:
      self._ecwait(lambda: not [m for m in messages if m in self.acked])

  @synchronized
  def commit(self, timeout=None):
    """
    Commit outstanding transactional work. This consists of all
    message sends and receives since the prior commit or rollback.
    """
    if not self.transactional:
      raise NontransactionalSession()
    self.committing = True
    self._wakeup()
    try:
      if not self._ecwait(lambda: not self.committing, timeout=timeout):
        raise Timeout("commit timed out")
    except TransactionError:
      raise
    except Exception, e:
      self.error = TransactionAborted(text="Transaction aborted: %s"%e)
      raise self.error
    if self.aborted:
      raise TransactionAborted()
    assert self.committed

  @synchronized
  def rollback(self, timeout=None):
    """
    Rollback outstanding transactional work. This consists of all
    message sends and receives since the prior commit or rollback.
    """
    if not self.transactional:
      raise NontransactionalSession()
    self.aborting = True
    self._wakeup()
    if not self._ecwait(lambda: not self.aborting, timeout=timeout):
      raise Timeout("rollback timed out")
    assert self.aborted

  @synchronized
  def sync(self, timeout=None):
    """
    Sync the session.
    """
    for snd in self.senders:
      snd.sync(timeout=timeout)
    if not self._ewait(lambda: not self.outgoing and not self.acked, timeout=timeout):
      raise Timeout("session sync timed out")

  @synchronized
  def close(self, timeout=None):
    """
    Close the session.
    """
    if self.error: return
    self.sync(timeout=timeout)

    for link in self.receivers + self.senders:
      link.close(timeout=timeout)

    if not self.closing:
      self.closing = True
      self._wakeup()

    try:
      if not self._ewait(lambda: self.closed, timeout=timeout):
        raise Timeout("session close timed out")
    finally:
      self.connection._remove_session(self)


class MangledString(str): pass

def _mangle(addr):
  if addr and addr.startswith("#"):
    return MangledString(str(uuid4()) + addr)
  else:
    return addr

class Sender(Endpoint):

  """
  Sends outgoing messages.
  """

  def __init__(self, session, id, target, options):
    super(Sender, self).__init__()
    self.session = session
    self.id = id
    self.target = target
    self.options = options
    self.capacity = options.get("capacity", UNLIMITED)
    self.threshold = 0.5
    self.durable = options.get("durable")
    self.queued = Serial(0)
    self.synced = Serial(0)
    self.acked = Serial(0)
    self.linked = False
    self.closing = False
    self.closed = False
    self._lock = self.session._lock

  def _wakeup(self):
    self.session._wakeup()

  def check_error(self):
    self.session.check_error()
    if self.error:
      raise self.error

  def get_error(self):
    err = self.session.get_error()
    if err:
      return err
    else:
      return self.error

  def _ewait(self, predicate, timeout=None):
    result = self.session._ewait(lambda: self.error or predicate(), timeout)
    self.check_error()
    return result

  def check_closed(self):
    if self.closed:
      raise LinkClosed()

  @synchronized
  def unsettled(self):
    """
    Returns the number of messages awaiting acknowledgment.
    @rtype: int
    @return: the number of unacknowledged messages
    """
    return self.queued - self.acked

  @synchronized
  def available(self):
    if self.capacity is UNLIMITED:
      return UNLIMITED
    else:
      return self.capacity - self.unsettled()

  @synchronized
  def send(self, object, sync=True, timeout=None):
    """
    Send a message. If the object passed in is of type L{unicode},
    L{str}, L{list}, or L{dict}, it will automatically be wrapped in a
    L{Message} and sent. If it is of type L{Message}, it will be sent
    directly. If the sender capacity is not L{UNLIMITED} then send
    will block until there is available capacity to send the message.
    If the timeout parameter is specified, then send will throw an
    L{InsufficientCapacity} exception if capacity does not become
    available within the specified time.

    @type object: unicode, str, list, dict, Message
    @param object: the message or content to send

    @type sync: boolean
    @param sync: if true then block until the message is sent

    @type timeout: float
    @param timeout: the time to wait for available capacity
    """

    if not self.session.connection._connected or self.session.closing:
      raise Detached()

    self._ecwait(lambda: self.linked, timeout=timeout)

    if isinstance(object, Message):
      message = object
    else:
      message = Message(object)

    if message.durable is None:
      message.durable = self.durable

    if self.capacity is not UNLIMITED:
      if self.capacity <= 0:
        raise InsufficientCapacity("capacity = %s" % self.capacity)
      if not self._ecwait(self.available, timeout=timeout):
        raise InsufficientCapacity("capacity = %s" % self.capacity)

    # XXX: what if we send the same message to multiple senders?
    message._sender = self
    if self.capacity is not UNLIMITED:
      message._sync = sync or self.available() <= int(ceil(self.threshold*self.capacity))
    else:
      message._sync = sync
    self.session.outgoing.append(message)
    self.queued += 1

    if sync:
      self.sync(timeout=timeout)
      assert message not in self.session.outgoing
    else:
      self._wakeup()

  @synchronized
  def sync(self, timeout=None):
    mno = self.queued
    if self.synced < mno:
      self.synced = mno
      self._wakeup()
    try:
      if not self._ewait(lambda: self.acked >= mno, timeout=timeout):
        raise Timeout("sender sync timed out")
    except ContentError:
      # clean bad message so we can continue
      self.acked = mno
      self.session.outgoing.pop(0)
      raise

  @synchronized
  def close(self, timeout=None):
    """
    Close the Sender.
    """
    # avoid erroring out when closing a sender that was never
    # established
    if self.acked < self.queued:
      self.sync(timeout=timeout)

    if not self.closing:
      self.closing = True
      self._wakeup()

    try:
      if not self.session._ewait(lambda: self.closed, timeout=timeout):
        raise Timeout("sender close timed out")
    finally:
      try:
        self.session.senders.remove(self)
      except ValueError:
        pass


class Receiver(Endpoint):

  """
  Receives incoming messages from a remote source. Messages may be
  fetched with L{fetch}.
  """

  def __init__(self, session, id, source, options):
    super(Receiver, self).__init__()
    self.session = session
    self.id = id
    self.source = source
    self.options = options

    self.granted = Serial(0)
    self.draining = False
    self.impending = Serial(0)
    self.received = Serial(0)
    self.returned = Serial(0)

    self.linked = False
    self.closing = False
    self.closed = False
    self._lock = self.session._lock
    self._capacity = 0
    self._set_capacity(options.get("capacity", 0), False)
    self.threshold = 0.5

  @synchronized
  def _set_capacity(self, c, wakeup=True):
    if c is UNLIMITED:
      self._capacity = c.value
    else:
      self._capacity = c
    self._grant()
    if wakeup:
      self._wakeup()

  def _get_capacity(self):
    if self._capacity == UNLIMITED.value:
      return UNLIMITED
    else:
      return self._capacity

  capacity = property(_get_capacity, _set_capacity)

  def _wakeup(self):
    self.session._wakeup()

  def check_error(self):
    self.session.check_error()
    if self.error:
      raise self.error

  def get_error(self):
    err = self.session.get_error()
    if err:
      return err
    else:
      return self.error

  def _ewait(self, predicate, timeout=None):
    result = self.session._ewait(lambda: self.error or predicate(), timeout)
    self.check_error()
    return result

  def check_closed(self):
    if self.closed:
      raise LinkClosed()

  @synchronized
  def unsettled(self):
    """
    Returns the number of acknowledged messages awaiting confirmation.
    """
    return len([m for m in self.session.acked if m._receiver is self])

  @synchronized
  def available(self):
    """
    Returns the number of messages available to be fetched by the
    application.

    @rtype: int
    @return: the number of available messages
    """
    return self.received - self.returned

  @synchronized
  def fetch(self, timeout=None):
    """
    Fetch and return a single message. A timeout of None will block
    forever waiting for a message to arrive, a timeout of zero will
    return immediately if no messages are available.

    @type timeout: float
    @param timeout: the time to wait for a message to be available
    """

    self._ecwait(lambda: self.linked)

    if self._capacity == 0:
      self.granted = self.returned + 1
      self._wakeup()
    self._ecwait(lambda: self.impending >= self.granted)
    msg = self.session._get(self, timeout=timeout)
    if msg is None:
      self.check_closed()
      self.draining = True
      self._wakeup()
      self._ecwait(lambda: not self.draining)
      msg = self.session._get(self, timeout=0)
      self._grant()
      self._wakeup()
      if msg is None:
        raise Empty()
    elif self._capacity not in (0, UNLIMITED.value):
      t = int(ceil(self.threshold * self._capacity))
      if self.received - self.returned <= t:
        self.granted = self.returned + self._capacity
        self._wakeup()
    return msg

  def _grant(self):
    if self._capacity == UNLIMITED.value:
      self.granted = UNLIMITED
    else:
      self.granted = self.returned + self._capacity

  @synchronized
  def close(self, timeout=None):
    """
    Close the receiver.
    """
    if not self.closing:
      self.closing = True
      self._wakeup()

    try:
      if not self.session._ewait(lambda: self.closed, timeout=timeout):
        raise Timeout("receiver close timed out")
    finally:
      try:
        self.session.receivers.remove(self)
      except ValueError:
        pass


__all__ = ["Connection", "Endpoint", "Session", "Sender", "Receiver"]