# Source code for rpyc.utils.helpers

    """
    Helpers and wrappers for common RPyC tasks
    """
    import time
    from rpyc.lib import spawn
    from rpyc.lib.colls import WeakValueDict
    from rpyc.lib.compat import callable
    from rpyc.core.consts import HANDLE_BUFFITER, HANDLE_CALL
    from rpyc.core.netref import syncreq, asyncreq

    def buffiter(obj, chunk=10, max_chunk=1000, factor=2):
        """Buffered iterator - reads the remote iterator in chunks starting with
        *chunk*, multiplying the chunk size by *factor* every time, as an
        exponential-backoff, up to a chunk of *max_chunk* size.

        ``buffiter`` is very useful for tight loops, where you fetch an element
        from the other side with every iterator. Instead of being limited by the
        network's latency after every iteration, ``buffiter`` fetches a "chunk"
        of elements every time, reducing the amount of network I/Os.

        :param obj: An iterable object (supports ``iter()``)
        :param chunk: the initial chunk size
        :param max_chunk: the maximal chunk size
        :param factor: the factor by which to multiply the chunk size after every
                       iterator (up to *max_chunk*). Must be >= 1.

        :returns: an iterator

        Example::

            cursor = db.get_cursor()
            for id, name, dob in buffiter(cursor.select("Id", "Name", "DoB")):
                print id, name, dob
        """
        if factor < 1:
            raise ValueError("factor must be >= 1, got %r" % (factor,))
        it = iter(obj)
        count = chunk
        while True:
            items = syncreq(it, HANDLE_BUFFITER, count)
            count = min(count * factor, max_chunk)
            if not items:
                break
            for elem in items:
                yield elem

    def restricted(obj, attrs, wattrs=None):
        """Returns a 'restricted' version of an object, i.e., allowing access only to a subset of its
        attributes. This is useful when returning a "broad" or "dangerous" object, where you don't
        want the other party to have access to all of its attributes.

        .. versionadded:: 3.2

        :param obj: any object
        :param attrs: the set of attributes exposed for reading (``getattr``) or writing (``setattr``).
                      The same set will serve both for reading and writing, unless wattrs is explicitly
                      given.
        :param wattrs: the set of attributes exposed for writing (``setattr``). If ``None``,
                       ``wattrs`` will default to ``attrs``. To disable setting attributes completely,
                       set to an empty tuple ``()``.
        :returns: a restricted view of the object

        Example::

            class MyService(rpyc.Service):
                def exposed_open(self, filename):
                    f = open(filename, "r")
                    return rpyc.restricted(f, {"read", "close"})   # disallow access to `seek` or `write`

        """
        if wattrs is None:
            wattrs = attrs

        class Restricted(object):
            def _rpyc_getattr(self, name):
                if name not in attrs:
                    raise AttributeError(name)
                return getattr(obj, name)
            __getattr__ = _rpyc_getattr

            def _rpyc_setattr(self, name, value):
                if name not in wattrs:
                    raise AttributeError(name)
                setattr(obj, name, value)
            __setattr__ = _rpyc_setattr
        return Restricted()

    class _Async(object):
        """Creates an async proxy wrapper over an existing proxy. Async proxies
        are cached. Invoking an async proxy will return an AsyncResult instead of
        blocking"""

        __slots__ = ("proxy", "__weakref__")

        def __init__(self, proxy):
            self.proxy = proxy

        def __call__(self, *args, **kwargs):
            return asyncreq(self.proxy, HANDLE_CALL, args, tuple(kwargs.items()))

        def __repr__(self):
            return "async_(%r)" % (self.proxy,)

    _async_proxies_cache = WeakValueDict()

    def async_(proxy):
        """
        Returns an asynchronous "version" of the given proxy. Invoking the returned
        proxy will not block; instead it will return an
        :class:`rpyc.core.async_.AsyncResult` object that you can test for completion

        :param proxy: any **callable** RPyC proxy

        :returns: the proxy, wrapped by an asynchronous wrapper

        Example::

            async_sleep = rpyc.async_(conn.modules.time.sleep)
            res = async_sleep(5)

        .. _async_note:

        .. note::
           In order to avoid overloading the GC, the returned asynchronous wrapper is
           cached as a weak reference. Therefore, do not use::

               rpyc.async_(foo)(5)

           Always store the returned asynchronous wrapper in a variable, e.g. ::

               a_foo = rpyc.async_(foo)
               a_foo(5)

        .. note::
            Furthermore, async requests provide **no guarantee on execution
            order**. In particular, multiple subsequent async requests may be
            executed in reverse order.
        """
        pid = id(proxy)
        if pid in _async_proxies_cache:
            return _async_proxies_cache[pid]
        if not hasattr(proxy, "____conn__") or not hasattr(proxy, "____id_pack__"):
            raise TypeError("'proxy' must be a Netref: %r", (proxy,))
        if not callable(proxy):
            raise TypeError("'proxy' must be callable: %r" % (proxy,))
        caller = _Async(proxy)
        _async_proxies_cache[id(caller)] = _async_proxies_cache[pid] = caller
        return caller

    async_.__doc__ = _Async.__doc__
    globals()['async'] = async_         # backward compatibility alias

    class timed(object):
        """Creates a timed asynchronous proxy. Invoking the timed proxy will
        run in the background and will raise an :class:`rpyc.core.async_.AsyncResultTimeout`
        exception if the computation does not terminate within the given time frame

        :param proxy: any **callable** RPyC proxy
        :param timeout: the maximal number of seconds to allow the operation to run

        :returns: a ``timed`` wrapped proxy

        Example::

            t_sleep = rpyc.timed(conn.modules.time.sleep, 6) # allow up to 6 seconds
            t_sleep(4) # okay
            t_sleep(8) # will time out and raise AsyncResultTimeout
        """

        __slots__ = ("__weakref__", "proxy", "timeout")

        def __init__(self, proxy, timeout):
            self.proxy = async_(proxy)
            self.timeout = timeout

        def __call__(self, *args, **kwargs):
            res = self.proxy(*args, **kwargs)
            res.set_expiry(self.timeout)
            return res

        def __repr__(self):
            return "timed(%r, %r)" % (self.proxy.proxy, self.timeout)

    class BgServingThread(object):
        """Runs an RPyC server in the background to serve all requests and replies
        that arrive on the given RPyC connection. The thread is started upon the
        the instantiation of the ``BgServingThread`` object; you can use the
        :meth:`stop` method to stop the server thread

        Example::

            conn = rpyc.connect(...)
            bg_server = BgServingThread(conn)
            ...
            bg_server.stop()

        .. note::
           For a more detailed explanation of asynchronous operation and the role of the
           ``BgServingThread``, see :ref:`tut5`

        """
        # these numbers are magical...
        SERVE_INTERVAL = 0.0
        SLEEP_INTERVAL = 0.1

        def __init__(self, conn, callback=None):
            self._conn = conn
            self._active = True
            self._callback = callback
            self._thread = spawn(self._bg_server)

        def __del__(self):
            if self._active:
                self.stop()

        def _bg_server(self):
            try:
                while self._active:
                    self._conn.serve(self.SERVE_INTERVAL)
                    time.sleep(self.SLEEP_INTERVAL)  # to reduce contention
            except Exception:
                if self._active:
                    self._active = False
                    if self._callback is None:
                        raise
                    self._callback()

        def stop(self):
            """stop the server thread. once stopped, it cannot be resumed. you will
            have to create a new BgServingThread object later."""
            assert self._active
            self._active = False
            self._thread.join()
            self._conn = None

    def classpartial(*args, **kwargs):
        """Bind arguments to a class's __init__."""
        cls, args = args[0], args[1:]

        class Partial(cls):
            __doc__ = cls.__doc__

            def __new__(self):
                return cls(*args, **kwargs)
        Partial.__name__ = cls.__name__
        return Partial