# Source code for aioxmpp.avatar.service

    ########################################################################
    # File name: service.py
    # This file is part of: aioxmpp
    #
    # LICENSE
    #
    # This program is free software: you can redistribute it and/or modify
    # it under the terms of the GNU Lesser General Public License as
    # published by the Free Software Foundation, either version 3 of the
    # License, or (at your option) any later version.
    #
    # This program is distributed in the hope that it will be useful, but
    # WITHOUT ANY WARRANTY; without even the implied warranty of
    # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
    # Lesser General Public License for more details.
    #
    # You should have received a copy of the GNU Lesser General Public
    # License along with this program.  If not, see
    # <http://www.gnu.org/licenses/>.
    #
    ########################################################################
    import asyncio
    import collections
    import hashlib
    import logging
    import warnings

    import aioxmpp
    import aioxmpp.callbacks as callbacks
    import aioxmpp.service as service
    import aioxmpp.disco as disco
    import aioxmpp.pep as pep
    import aioxmpp.presence as presence
    import aioxmpp.pubsub as pubsub
    import aioxmpp.vcard as vcard

    from aioxmpp.cache import LRUDict
    from aioxmpp.utils import namespaces, gather_reraise_multi

    from . import xso as avatar_xso

    logger = logging.getLogger(__name__)

    def normalize_id(id_):
        """
        Normalize a SHA1 sum encoded as hexadecimal number in ASCII.

        This does nothing but lowercase the string as to enable robust
        comparison.
        """
        return id_.lower()

    class AvatarSet:
        """
        A list of sources of an avatar.

        Exactly one of the sources must include image data in the
        ``image/png`` format. The others provide the location of the
        image data as an URL.

        Adding pointer avatar data is not yet supported.

        .. automethod:: add_avatar_image
        """

        def __init__(self):
            self._image_bytes = None
            self._png_id = None
            self._metadata = avatar_xso.Metadata()

        @property
        def image_bytes(self):
            """
            The image data bytes for MIME type ``text/png``.
            """
            return self._image_bytes

        @property
        def metadata(self):
            """
            The :class:`Metadata` XSO corresponding to this avatar set.
            """
            return self._metadata

        @property
        def png_id(self):
            """
            The SHA1 of the ``image/png`` image data.

            This id is always normalized in the sense of :function:`normalize_id`.
            """
            return self._png_id

        def add_avatar_image(self, mime_type, *, id_=None,
                             image_bytes=None, width=None, height=None,
                             url=None, nbytes=None):
            """
            Add a source of the avatar image.

            All sources of an avatar image added to an avatar set must be
            *the same image*, in different formats and sizes.

            :param mime_type: The MIME type of the avatar image.
            :param id_: The SHA1 of the image data.
            :param nbytes: The size of the image data in bytes.
            :param image_bytes: The image data, this must be supplied only
                                in one call.
            :param url: The URL of the avatar image.
            :param height: The height of the image in pixels (optional).
            :param width: The width of the image in pixels (optional).

            `id_` and `nbytes` may be omitted if and only if `image_data`
            is given and `mime_type` is ``image/png``. If they are
            supplied *and* image data is given, they are checked to match
            the image data.

            It is the caller's responsibility to assure that the provided
            links exist and the files have the correct SHA1 sums.
            """

            if mime_type == "image/png":
                if image_bytes is not None:
                    if self._image_bytes is not None:
                        raise RuntimeError(
                            "Only one avatar image may be published directly."
                        )

                    sha1 = hashlib.sha1()
                    sha1.update(image_bytes)
                    id_computed = normalize_id(sha1.hexdigest())
                    if id_ is not None:
                        id_ = normalize_id(id_)
                        if id_ != id_computed:
                            raise RuntimeError(
                                "The given id does not match the SHA1 of "
                                "the image data."
                            )
                    else:
                        id_ = id_computed

                    nbytes_computed = len(image_bytes)
                    if nbytes is not None:
                        if nbytes != nbytes_computed:
                            raise RuntimeError(
                                "The given length does not match the length "
                                "of the image data."
                            )
                    else:
                        nbytes = nbytes_computed

                    self._image_bytes = image_bytes
                    self._png_id = id_

            if image_bytes is None and url is None:
                raise RuntimeError(
                    "Either the image bytes or an url to retrieve the avatar "
                    "image must be given."
                )

            if nbytes is None:
                raise RuntimeError(
                    "Image data length is not given an not inferable "
                    "from the other arguments."
                )

            if id_ is None:
                raise RuntimeError(
                    "The SHA1 of the image data is not given an not inferable "
                    "from the other arguments."
                )

            if image_bytes is not None and mime_type != "image/png":
                raise RuntimeError(
                    "The image bytes can only be given for image/png data."
                )

            self._metadata.info[mime_type].append(
                avatar_xso.Info(
                    id_=id_, mime_type=mime_type, nbytes=nbytes,
                    width=width, height=height, url=url
                )
            )

    class AbstractAvatarDescriptor:
        """
        Description of the properties of and how to retrieve a specific
        avatar.

        The following attribues are available for all instances:

        .. autoattribute:: remote_jid

        .. autoattribute:: id_

        .. autoattribute:: normalized_id

        .. autoattribute:: can_get_image_bytes_via_xmpp

        .. autoattribute:: has_image_data_in_pubsub

        The following attributes may be :data:`None` and are supposed to
        be used as hints for selection of the avatar to download:

        .. autoattribute:: nbytes

        .. autoattribute:: width

        .. autoattribute:: height

        .. autoattribute:: mime_type

        If this attribute is not :data:`None` it is an URL that points to
        the location of the avatar image:

        .. autoattribute:: url

        The image data belonging to the descriptor can be retrieved by the
        following coroutine:

        .. automethod:: get_image_bytes
        """

        def __init__(self, remote_jid, id_, *, mime_type=None,
                     nbytes=None, width=None, height=None, url=None):
            self._remote_jid = remote_jid
            self._mime_type = mime_type
            self._id = id_
            self._nbytes = nbytes
            self._width = width
            self._height = height
            self._url = url

        def __eq__(self, other):
            return (self._remote_jid == other._remote_jid and
                    self._mime_type == other._mime_type and
                    self._id == other._id and
                    self._nbytes == other._nbytes and
                    self._width == other._width and
                    self._height == other._height and
                    self._url == other._url)

        @asyncio.coroutine
        def get_image_bytes(self):
            """
            Try to retrieve the image data corresponding to this avatar
            descriptor.

            :returns: the image contents
            :rtype: :class:`bytes`

            :raises NotImplementedError: if we do not implement the
                capability to retrieve the image data of this type. It is
                guaranteed to not raise :class:`NotImplementedError` if
                :attr:`can_get_image_bytes_via_xmpp` is true.

            :raises RuntimeError: if the image data described by this
                descriptor is not at the specified location.

            :raises aiomxpp.XMPPCancelError: if trying to retrieve the
                image data causes an XMPP error.
            """
            raise NotImplementedError

        @property
        def can_get_image_bytes_via_xmpp(self):
            """
            Return whether :meth:`get_image_bytes` raises
            :class:`NotImplementedError`.
            """
            return False

        @property
        def has_image_data_in_pubsub(self):
            """
            Whether the image can be retrieved from PubSub.

            .. deprecated:: 0.10

               Use :attr:`can_get_image_bytes_via_xmpp` instead.

               As we support vCard based avatars now the name of this is
               misleading.

               This attribute will be removed in aioxmpp 1.0
            """
            warnings.warn(
                "the has_image_data_in_pubsub attribute is deprecated and will be"
                " removed in 1.0",
                DeprecationWarning,
                stacklevel=1
            )
            return self.can_get_image_bytes_via_xmpp

        @property
        def remote_jid(self):
            """
            The remote JID this avatar belongs to.
            """
            return self._remote_jid

        @property
        def url(self):
            """
            The URL where the avatar image data can be found.

            This may be :data:`None` if the avatar is not given as an URL
            of the image data.
            """
            return self._url

        @property
        def width(self):
            """
            The width of the avatar image in pixels.

            This is :data:`None` if this information is not supplied.
            """
            return self._width

        @property
        def height(self):
            """
            The height of the avatar image in pixels.

            This is :data:`None` if this information is not supplied.
            """
            return self._height

        @property
        def nbytes(self):
            """
            The size of the avatar image data in bytes.
            """
            return self._nbytes

        @property
        def id_(self):
            """
            The SHA1 of the image encoded as hexadecimal number in ASCII.

            This is the original value returned from the underlying
            protocol and should be used for any further interaction with
            the underlying protocol.
            """
            return self._id

        @property
        def normalized_id(self):
            """
            The normalized SHA1 of the image data.

            This is supposed to be used for caching and comparison.
            """
            return normalize_id(self._id)

        @property
        def mime_type(self):
            """
            The MIME type of the image data.
            """
            return self._mime_type

    class PubsubAvatarDescriptor(AbstractAvatarDescriptor):

        def __init__(self, remote_jid, id_, *, pubsub=None, **kwargs):
            super().__init__(remote_jid, id_, **kwargs)
            self._pubsub = pubsub

        def __eq__(self, other):
            return (isinstance(other, PubsubAvatarDescriptor) and
                    super().__eq__(other))

        @property
        def can_get_image_bytes_via_xmpp(self):
            return True

        @asyncio.coroutine
        def get_image_bytes(self):
            image_data = yield from self._pubsub.get_items_by_id(
                self._remote_jid,
                namespaces.xep0084_data,
                [self.id_],
            )
            if not image_data.payload.items:
                raise RuntimeError("Avatar image data is not set.")

            item, = image_data.payload.items
            return item.registered_payload.data

    class HttpAvatarDescriptor(AbstractAvatarDescriptor):

        @asyncio.coroutine
        def get_image_bytes(self):
            raise NotImplementedError

        def __eq__(self, other):
            return (isinstance(other, HttpAvatarDescriptor) and
                    super().__eq__(other))

    class VCardAvatarDescriptor(AbstractAvatarDescriptor):

        def __init__(self, remote_jid, id_, *, vcard=None, image_bytes=None,
                     **kwargs):
            super().__init__(remote_jid, id_, **kwargs)
            self._vcard = vcard
            self._image_bytes = image_bytes

        def __eq__(self, other):
            # NOTE: we explicitely do *not* check for the equality of
            # image bytes: image bytes is a hidden optimization
            return (isinstance(other, VCardAvatarDescriptor) and
                    super().__eq__(other))

        @property
        def can_get_image_bytes_via_xmpp(self):
            return True

        @asyncio.coroutine
        def get_image_bytes(self):
            if self._image_bytes is not None:
                return self._image_bytes

            logger.debug("retrieving vCard %s", self._remote_jid)
            vcard = yield from self._vcard.get_vcard(self._remote_jid)
            photo = vcard.get_photo_data()
            if photo is None:
                raise RuntimeError("Avatar image is not set")

            logger.debug("returning vCard avatar %s", self._remote_jid)
            return photo

    class AvatarService(service.Service):
        """
        Access and publish User Avatars (:xep:`84`). Fallback to vCard
        based avatars (:xep:`153`) if no PEP avatar is available.

        This service provides an interface for accessing the avatar of other
        entities in the network, getting notifications on avatar changes and
        publishing an avatar for this entity.

        .. versionchanged:: 0.10

           Support for :xep:`vCard-Based Avatars <153>` was added.

        Observing avatars:

        .. note:: :class:`AvatarService` only caches the metadata, not the
                  actual image data. This is the job of the caller.

        .. signal:: on_metadata_changed(jid, metadata)

            Fires when avatar metadata changes.

            :param jid: The JID which the avatar belongs to.
            :param metadata: The new metadata descriptors.
            :type metadata: a sequence of
                :class:`~aioxmpp.avatar.service.AbstractAvatarDescriptor`
                instances

        .. automethod:: get_avatar_metadata

        .. automethod:: subscribe

        Publishing avatars:

        .. automethod:: publish_avatar_set

        .. automethod:: disable_avatar

        .. automethod:: wipe_avatar

        Configuration:

        .. autoattribute:: synchronize_vcard

        .. autoattribute:: advertise_vcard

        .. attribute:: avatar_pep

           The PEP descriptor for claiming the avatar metadata namespace.
           The value is a :class:`~aioxmpp.pep.service.RegisteredPEPNode`,
           whose :attr:`~aioxmpp.pep.service.RegisteredPEPNode.notify`
           property can be used to disable or enable the notification
           feature.

        .. autoattribute:: metadata_cache_size
           :annotation: = 200
        """

        ORDER_AFTER = [
            disco.DiscoClient,
            disco.DiscoServer,
            pubsub.PubSubClient,
            pep.PEPClient,
            vcard.VCardService,
            presence.PresenceClient,
            presence.PresenceServer,
        ]

        avatar_pep = pep.register_pep_node(
            namespaces.xep0084_metadata,
            notify=True,
        )

        on_metadata_changed = callbacks.Signal()

        def __init__(self, client, **kwargs):
            super().__init__(client, **kwargs)
            self._has_pep_avatar = set()
            self._metadata_cache = LRUDict()
            self._metadata_cache.maxsize = 200
            self._pubsub = self.dependencies[pubsub.PubSubClient]
            self._pep = self.dependencies[pep.PEPClient]
            self._presence_server = self.dependencies[presence.PresenceServer]
            self._disco = self.dependencies[disco.DiscoClient]
            self._vcard = self.dependencies[vcard.VCardService]
            # we use this lock to prevent race conditions between different
            # calls of the methods by one client.
            # XXX: Other, independent clients may still cause inconsistent
            # data by race conditions, this should be fixed by at least
            # checking for consistent data after an update.
            self._publish_lock = asyncio.Lock()
            self._synchronize_vcard = False
            self._advertise_vcard = True
            self._vcard_resource_interference = set()
            self._vcard_id = None
            self._vcard_rehashing_for = None
            self._vcard_rehash_task = None

        @property
        def metadata_cache_size(self):
            """
            Maximum number of cache entries in the avatar metadata cache.

            This is mostly a measure to prevent malicious peers from
            exhausting memory by spamming vCard based avatar metadata for
            different resources.

            .. versionadded:: 0.10

            """
            return self._metadata_cache.maxsize

        @metadata_cache_size.setter
        def metadata_cache_size(self, value):
            self._metadata_cache.maxsize = value

        @property
        def synchronize_vcard(self):
            """
            Set this property to true to enable publishing the a vCard avatar.

            This property defaults to false. For the setting true to have
            effect, you have to publish your avatar with :meth:`publish_avatar_set`
            or :meth:`disable_avatar` *after* this switch has been set to true.
            """
            return self._synchronize_vcard

        @synchronize_vcard.setter
        def synchronize_vcard(self, value):
            self._synchronize_vcard = bool(value)

        @property
        def advertise_vcard(self):
            """
            Set this property to false to disable advertisement of the vCard
            avatar via presence broadcast.

            Note, that this reduces traffic, since it makes the presence
            stanzas smaller and we no longer have to recalculate the hash,
            this also disables vCard advertisement for all other
            ressources of the bare local jid, by the business rules of
            :xep:`0153`.

            Note that, when enabling this feature again the vCard has to
            be fetched from the server to recalculate the hash.
            """
            return self._advertise_vcard

        @advertise_vcard.setter
        def advertise_vcard(self, value):
            self._advertise_vcard = bool(value)
            if self._advertise_vcard:
                self._vcard_id = None
                self._start_rehash_task()

        @service.depfilter(aioxmpp.stream.StanzaStream,
                           "service_outbound_presence_filter")
        def _attach_vcard_notify_to_presence(self, stanza):
            if self._advertise_vcard:
                if self._vcard_resource_interference:
                    # do not advertise the hash if there is resource interference
                    stanza.xep0153_x = avatar_xso.VCardTempUpdate()
                else:
                    stanza.xep0153_x = avatar_xso.VCardTempUpdate(self._vcard_id)

            return stanza

        def _update_metadata(self, cache_jid, metadata):
            try:
                cached_metadata = self._metadata_cache[cache_jid]
            except KeyError:
                pass
            else:
                if cached_metadata == metadata:
                    return

            self._metadata_cache[cache_jid] = metadata
            self.on_metadata_changed(
                cache_jid,
                metadata
            )

        def _handle_notify(self, full_jid, stanza):
            # handle resource interference as per XEP-153 business rules,
            # we go along with this tracking even if vcard advertisement
            # is off
            if (full_jid.bare() == self.client.local_jid.bare() and
                    full_jid != self.client.local_jid):
                if stanza.xep0153_x is None:
                    self._vcard_resource_interference.add(full_jid)
                else:
                    if self._vcard_resource_interference:
                        self._vcard_resource_interference.discard(full_jid)
                        if not self._vcard_resource_interference:
                            self._vcard_id = None

            # otherwise ignore stanzas without xep0153_x payload, or
            # no photo tag.
            if stanza.xep0153_x is None:
                return

            if stanza.xep0153_x.photo is None:
                return

            # special case MUC presence – otherwise the vcard is retrieved
            # for the bare jid
            if stanza.xep0045_muc_user is not None:
                cache_jid = full_jid
            else:
                cache_jid = full_jid.bare()

            if cache_jid not in self._has_pep_avatar:
                metadata = self._cook_vcard_notify(cache_jid, stanza)
                self._update_metadata(cache_jid, metadata)

            # trigger the download of the vCard and calculation of the
            # vCard avatar hash, if some other resource of our bare jid
            # reported a hash distinct from ours!
            # don't do this if there is a non-compliant resource, we don't
            # send the hash in that case anyway
            if (full_jid.bare() == self.client.local_jid.bare() and
                    full_jid != self.client.local_jid and
                    self._advertise_vcard and
                    not self._vcard_resource_interference):
                if (self._vcard_id is None or
                        stanza.xep0153_x.photo.lower() !=
                        self._vcard_id.lower()):

                    # do not rehash if we alread have a rehash task that
                    # was triggered by an update with the same hash
                    if (self._vcard_rehashing_for is None or
                            self._vcard_rehashing_for !=
                                stanza.xep0153_x.photo.lower()):
                        self._vcard_rehashing_for = stanza.xep0153_x.photo.lower()
                        self._start_rehash_task()

        def _start_rehash_task(self):
            if self._vcard_rehash_task is not None:
                self._vcard_rehash_task.cancel()

            self._vcard_id = None
            # as per XEP immediately resend the presence with empty update
            # element, as this is not synchronous it might already contaiin
            # the new hash, but this is okay as well (as it makes the cached
            # presence stanzas coherent as well).
            self._presence_server.resend_presence()

            self._vcard_rehash_task = asyncio.ensure_future(
                self._calculate_vcard_id()
            )

            def set_new_vcard_id(fut):
                self._vcard_rehashing_for = None
                if not fut.cancelled():
                    self._vcard_id = fut.result()

            self._vcard_rehash_task.add_done_callback(
                set_new_vcard_id
            )

        @asyncio.coroutine
        def _calculate_vcard_id(self):
            self.logger.debug("updating vcard hash")
            vcard = yield from self._vcard.get_vcard()
            self.logger.debug("got vcard for hash update: %s", vcard)
            photo = vcard.get_photo_data()

            # if no photo is set in the vcard, set an empty <photo> element
            # in the update; according to the spec this means the avatar
            # is disabled
            if photo is None:
                self.logger.debug("no photo in vcard, advertising as such")
                return ""

            sha1 = hashlib.sha1()
            sha1.update(photo)
            new_hash = sha1.hexdigest().lower()
            self.logger.debug("updated hash to %s", new_hash)
            return new_hash

        @service.depsignal(presence.PresenceClient, "on_available")
        def _handle_on_available(self, full_jid, stanza):
            self._handle_notify(full_jid, stanza)

        @service.depsignal(presence.PresenceClient, "on_changed")
        def _handle_on_changed(self, full_jid, stanza):
            self._handle_notify(full_jid, stanza)

        @service.depsignal(presence.PresenceClient, "on_unavailable")
        def _handle_on_unavailable(self, full_jid, stanza):
            if full_jid.bare() == self.client.local_jid.bare():
                if self._vcard_resource_interference:
                    self._vcard_resource_interference.discard(full_jid)
                    if not self._vcard_resource_interference:
                        self._start_rehash_task()

            # correctly handle MUC avatars
            if stanza.xep0045_muc_user is not None:
                self._metadata_cache.pop(full_jid, None)

        def _cook_vcard_notify(self, jid, stanza):
            result = []
            # note: an empty photo element correctly
            # results in an empty avatar metadata list
            if stanza.xep0153_x.photo:
                result.append(
                    VCardAvatarDescriptor(
                        remote_jid=jid,
                        id_=stanza.xep0153_x.photo,
                        mime_type=None,
                        vcard=self._vcard,
                        nbytes=None,
                    )
                )
            return result

        def _cook_metadata(self, jid, items):
            def iter_metadata_info_nodes(items):
                for item in items:
                    yield from item.registered_payload.iter_info_nodes()

            result = []
            for info_node in iter_metadata_info_nodes(items):
                if info_node.url is not None:
                    descriptor = HttpAvatarDescriptor(
                        remote_jid=jid,
                        id_=info_node.id_,
                        mime_type=info_node.mime_type,
                        nbytes=info_node.nbytes,
                        width=info_node.width,
                        height=info_node.height,
                        url=info_node.url,
                    )
                else:
                    descriptor = PubsubAvatarDescriptor(
                        remote_jid=jid,
                        id_=info_node.id_,
                        mime_type=info_node.mime_type,
                        nbytes=info_node.nbytes,
                        width=info_node.width,
                        height=info_node.height,
                        pubsub=self._pubsub,
                    )
                result.append(descriptor)

            return result

        @service.attrsignal(avatar_pep, "on_item_publish")
        def _handle_pubsub_publish(self, jid, node, item, *, message=None):
            # update the metadata cache
            metadata = self._cook_metadata(jid, [item])
            self._has_pep_avatar.add(jid)
            self._update_metadata(jid, metadata)

        @asyncio.coroutine
        def _get_avatar_metadata_vcard(self, jid):
            logger.debug("trying vCard avatar as fallback for %s", jid)
            vcard = yield from self._vcard.get_vcard(jid)
            photo = vcard.get_photo_data()
            mime_type = vcard.get_photo_mime_type()
            if photo is None:
                return []

            logger.debug("success vCard avatar as fallback for %s",
                         jid)
            sha1 = hashlib.sha1()
            sha1.update(photo)
            return [VCardAvatarDescriptor(
                remote_jid=jid,
                id_=sha1.hexdigest(),
                mime_type=mime_type,
                nbytes=len(photo),
                vcard=self._vcard,
                image_bytes=photo,
            )]

        @asyncio.coroutine
        def _get_avatar_metadata_pep(self, jid):
            try:
                metadata_raw = yield from self._pubsub.get_items(
                    jid,
                    namespaces.xep0084_metadata,
                    max_items=1
                )
            except aioxmpp.XMPPCancelError as e:
                # transparently map feature-not-implemented and
                # item-not-found to be equivalent unset avatar
                if e.condition in (
                        aioxmpp.ErrorCondition.FEATURE_NOT_IMPLEMENTED,
                        aioxmpp.ErrorCondition.ITEM_NOT_FOUND):
                    return []
                raise

            self._has_pep_avatar.add(jid)
            return self._cook_metadata(jid, metadata_raw.payload.items)

        @asyncio.coroutine
        def get_avatar_metadata(self, jid, *, require_fresh=False,
                                disable_pep=False):
            """
            Retrieve a list of avatar descriptors.

            :param jid: the JID for which to retrieve the avatar metadata.
            :type jid: :class:`aioxmpp.JID`
            :param require_fresh: if true, do not return results from the
                avatar metadata chache, but retrieve them again from the server.
            :type require_fresh: :class:`bool`
            :param disable_pep: if true, do not try to retrieve the avatar
                via pep, only try the vCard fallback. This usually only
                useful when querying avatars via MUC, where the PEP request
                would be invalid (since it would be for a full jid).
            :type disable_pep: :class:`bool`

            :returns: an iterable of avatar descriptors.
            :rtype: a :class:`list` of
                :class:`~aioxmpp.avatar.service.AbstractAvatarDescriptor`
                instances

            Returning an empty list means that the avatar not set.

            We mask a :class:`XMPPCancelError` in the case that it is
            ``feature-not-implemented`` or ``item-not-found`` and return
            an empty list of avatar descriptors, since this is
            semantically equivalent to not having an avatar.

            .. note::

               It is usually an error to get the avatar for a full jid,
               normally, the avatar is set for the bare jid of a user. The
               exception are vCard avatars over MUC, where the IQ requests
               for the vCard may be translated by the MUC server. It is
               recommended to use the `disable_pep` option in that case.
            """

            if require_fresh:
                self._metadata_cache.pop(jid, None)
            else:
                try:
                    return self._metadata_cache[jid]
                except KeyError:
                    pass

            if disable_pep:
                metadata = []
            else:
                metadata = yield from self._get_avatar_metadata_pep(jid)

            # try the vcard fallback, note: we don't try this
            # if the PEP avatar is disabled!
            if not metadata and jid not in self._has_pep_avatar:
                metadata = yield from self._get_avatar_metadata_vcard(jid)

            # if a notify was fired while we waited for the results, then
            # use the version in the cache, this will mitigate the race
            # condition because if our version is actually newer we will
            # soon get another notify for this version change!
            if jid not in self._metadata_cache:
                self._update_metadata(jid, metadata)
            return self._metadata_cache[jid]

        @asyncio.coroutine
        def subscribe(self, jid):
            """
            Explicitly subscribe to metadata change notifications for `jid`.
            """
            yield from self._pubsub.subscribe(jid, namespaces.xep0084_metadata)

        @aioxmpp.service.depsignal(aioxmpp.stream.StanzaStream,
                                   "on_stream_destroyed")
        def handle_stream_destroyed(self, reason):
            self._metadata_cache.clear()
            self._vcard_resource_interference.clear()
            self._has_pep_avatar.clear()

        @asyncio.coroutine
        def publish_avatar_set(self, avatar_set):
            """
            Make `avatar_set` the current avatar of the jid associated with this
            connection.

            If :attr:`synchronize_vcard` is true and PEP is available the
            vCard is only synchronized if the PEP update is successful.

            This means publishing the ``image/png`` avatar data and the
            avatar metadata set in pubsub. The `avatar_set` must be an
            instance of :class:`AvatarSet`. If :attr:`synchronize_vcard` is
            true the avatar is additionally published in the user vCard.
            """
            id_ = avatar_set.png_id

            done = False
            with (yield from self._publish_lock):
                if (yield from self._pep.available()):
                    yield from self._pep.publish(
                        namespaces.xep0084_data,
                        avatar_xso.Data(avatar_set.image_bytes),
                        id_=id_
                    )

                    yield from self._pep.publish(
                        namespaces.xep0084_metadata,
                        avatar_set.metadata,
                        id_=id_
                    )
                    done = True

                if self._synchronize_vcard:
                    my_vcard = yield from self._vcard.get_vcard()
                    my_vcard.set_photo_data("image/png",
                                            avatar_set.image_bytes)
                    self._vcard_id = avatar_set.png_id
                    yield from self._vcard.set_vcard(my_vcard)
                    self._presence_server.resend_presence()
                    done = True

            if not done:
                raise RuntimeError(
                    "failed to publish avatar: no protocol available"
                )

        @asyncio.coroutine
        def _disable_vcard_avatar(self):
            my_vcard = yield from self._vcard.get_vcard()
            my_vcard.clear_photo_data()
            self._vcard_id = ""
            yield from self._vcard.set_vcard(my_vcard)
            self._presence_server.resend_presence()

        @asyncio.coroutine
        def disable_avatar(self):
            """
            Temporarily disable the avatar.

            If :attr:`synchronize_vcard` is true, the vCard avatar is
            disabled (even if disabling the PEP avatar fails).

            This is done by setting the avatar metadata node empty and if
            :attr:`synchronize_vcard` is true, downloading the vCard,
            removing the avatar data and re-uploading the vCard.

            This method does not error if neither protocol is active.

            :raises aioxmpp.errors.GatherError: if an exception is raised
                by the spawned tasks.
            """

            with (yield from self._publish_lock):
                todo = []
                if self._synchronize_vcard:
                    todo.append(self._disable_vcard_avatar())

                if (yield from self._pep.available()):
                    todo.append(self._pep.publish(
                        namespaces.xep0084_metadata,
                        avatar_xso.Metadata()
                    ))

                yield from gather_reraise_multi(*todo, message="disable_avatar")

        @asyncio.coroutine
        def wipe_avatar(self):
            """
            Remove all avatar data stored on the server.

            If :attr:`synchronize_vcard` is true, the vCard avatar is
            disabled even if disabling the PEP avatar fails.

            This is equivalent to :meth:`disable_avatar` for vCard-based
            avatars, but will also remove the data PubSub node for
            PEP avatars.

            This method does not error if neither protocol is active.

            :raises aioxmpp.errors.GatherError: if an exception is raised
                by the spawned tasks.
            """

            @asyncio.coroutine
            def _wipe_pep_avatar():
                yield from self._pep.publish(
                    namespaces.xep0084_metadata,
                    avatar_xso.Metadata()
                )
                yield from self._pep.publish(
                    namespaces.xep0084_data,
                    avatar_xso.Data(b'')
                )

            with (yield from self._publish_lock):
                todo = []
                if self._synchronize_vcard:
                    todo.append(self._disable_vcard_avatar())

                if (yield from self._pep.available()):
                    todo.append(_wipe_pep_avatar())

                yield from gather_reraise_multi(*todo, message="wipe_avatar")