from pypy.interpreter.baseobjspace import W_Root
from pypy.interpreter.error import OperationError, oefmt
from pypy.interpreter.pyopcode import SApplicationException, Yield
from pypy.interpreter.pycode import CO_YIELD_INSIDE_TRY
from pypy.interpreter.typedef import TypeDef, make_weakref_descr, interp_attrproperty, GetSetProperty
from pypy.interpreter.typedef import interp_attrproperty_w
from pypy.interpreter.gateway import interp2app
from pypy.interpreter.astcompiler import consts
from pypy.objspace.std.util import generic_alias_class_getitem
from rpython.rlib import jit, rgc, rweakref
from rpython.rlib.objectmodel import specialize
from rpython.rlib.rarithmetic import r_uint


class GeneratorOrCoroutine(W_Root):
    _immutable_fields_ = ['pycode']

    def __init__(self, frame, name=None, qualname=None):
        self.space = frame.space
        self.frame = frame     # turned into None when frame_finished_execution
        self.pycode = frame.pycode  # usually never None but see __setstate__
        self.running = False
        self._name = name           # may be null, use get_name()
        self._qualname = qualname   # may be null, use get_qualname()
        if (isinstance(self, Coroutine)
                or self.pycode.co_flags & CO_YIELD_INSIDE_TRY):
            self.register_finalizer(self.space)
        self.saved_operr = None
        self.previous_gen_or_coroutine = None

    def get_name(self):
        # 'name' is a byte string that is valid utf-8
        if self._name is not None:
            return self._name
        elif self.pycode is None:
            return "<finished>"
        else:
            return self.pycode.co_name

    def get_qualname(self):
        # 'qualname' is a unicode string
        if self._qualname is not None:
            return self._qualname
        return self.get_name()

    def descr__repr__(self, space):
        addrstring = self.getaddrstring(space)
        return space.newtext("<%s object %s at 0x%s>" %
                          (self.KIND, self.get_qualname(), addrstring))

    def descr_send(self, w_arg):
        """send(arg) -> send 'arg' into generator/coroutine,
return next yielded value or raise StopIteration."""
        return self.send_ex(w_arg)

    def send_error(self, operr):
        return self.send_ex(SApplicationException(operr))

    def send_ex(self, w_arg_or_err):
        assert w_arg_or_err is not None
        pycode = self.pycode
        if pycode is not None:
            if jit.we_are_jitted() and should_not_inline(pycode):
                generatorentry_driver.jit_merge_point(gen=self,
                                                      w_arg=w_arg_or_err,
                                                      pycode=pycode)
        return self._send_ex(w_arg_or_err)

    def _send_ex(self, w_arg_or_err):
        space = self.space

        frame = self.frame
        if frame is None:
            if isinstance(self, Coroutine):
                # NB. CPython checks a flag 'closing' here, but instead
                # we can simply not be here at all if frame is None in
                # this case
                raise oefmt(space.w_RuntimeError,
                            "cannot reuse already awaited coroutine")
            # xxx a bit ad-hoc, but we don't want to go inside
            # execute_frame() if the frame is actually finished
            if isinstance(w_arg_or_err, SApplicationException):
                operr = w_arg_or_err.operr
            elif isinstance(self, AsyncGenerator):
                operr = OperationError(space.w_StopAsyncIteration, space.w_None)
            else:
                operr = OperationError(space.w_StopIteration, space.w_None)
            raise operr

        w_result = self._invoke_execute_frame(w_arg_or_err)
        assert w_result is not None

        # if the frame is now marked as finished, it was RETURNed from
        if frame.frame_finished_execution:
            self.frame_is_finished()
            if isinstance(self, AsyncGenerator):
                assert space.is_w(w_result, space.w_None), (
                    "getting non-None here should be forbidden by the bytecode")
                raise OperationError(space.w_StopAsyncIteration, space.w_None)
            else:
                if space.is_w(w_result, space.w_None):
                    raise OperationError(space.w_StopIteration, space.w_None)
                else:
                    raise stopiteration_value(space, w_result)
        else:
            return w_result     # YIELDed

    def _invoke_execute_frame(self, w_arg_or_err):
        space = self.space
        frame = self.frame
        if self.running:
            raise oefmt(space.w_ValueError, "%s already executing", self.KIND)
        #
        # Optimization only: after we've started a Coroutine without
        # CO_YIELD_INSIDE_TRY, then Coroutine._finalize_() will be a no-op
        if frame.last_instr == -1:
            if (isinstance(self, Coroutine) and
                not (self.pycode.co_flags & CO_YIELD_INSIDE_TRY)):
                rgc.may_ignore_finalizer(self)

            if (not space.is_w(w_arg_or_err, space.w_None) and
                not isinstance(w_arg_or_err, SApplicationException)):
                raise oefmt(space.w_TypeError,
                            "can't send non-None value to a just-started %s",
                            self.KIND)
        ec = space.getexecutioncontext()
        ec.push_gen_or_coroutine(self)
        d = frame.getdebug()
        self.running = True
        try:
            w_result = frame.execute_frame(w_arg_or_err)
        except OperationError as e:
            # errors finish a frame
            try:
                if e.match(space, space.w_StopIteration):
                    self._leak_stopiteration(e)
                elif (isinstance(self, AsyncGenerator) and
                      e.match(space, space.w_StopAsyncIteration)):
                    self._leak_stopasynciteration(e)
            finally:
                self.frame_is_finished()
            raise
        finally:
            frame.f_backref = jit.vref_None
            self.running = False
            ec.pop_gen_or_coroutine(self)
        return w_result

    def get_delegate(self):
        if self.frame is None:
            return None
        return self.frame.w_yielding_from

    def descr_delegate(self, space):
        w_yf = self.get_delegate()
        if w_yf is None:
            return space.w_None
        return w_yf

    def set_delegate(self, w_delegate):
        self.frame.w_yielding_from = w_delegate

    def _leak_stopiteration(self, e):
        # turn a leaking StopIteration into RuntimeError (with its cause set
        # appropriately).
        space = self.space
        e2 = OperationError(space.w_RuntimeError,
                            space.newtext("%s raised StopIteration" %
                                          self.KIND))
        e2.chain_exceptions_from_cause(space, e)
        raise e2

    def _leak_stopasynciteration(self, e):
        space = self.space
        e2 = OperationError(space.w_RuntimeError,
                   space.newtext("async generator raised StopAsyncIteration"))
        e2.chain_exceptions_from_cause(space, e)
        raise e2

    def descr_throw(self, w_type, w_val=None, w_tb=None):
        """throw(typ[,val[,tb]]) -> raise exception in generator/coroutine,
return next yielded value or raise StopIteration."""
        return self.throw(w_type, w_val, w_tb)

    def throw(self, w_type, w_val, w_tb):
        from pypy.interpreter.pytraceback import check_traceback

        space = self.space
        if w_val is None:
            w_val = space.w_None

        msg = "throw() third argument must be a traceback object"
        if space.is_none(w_tb):
            tb = None
        else:
            tb = check_traceback(space, w_tb, msg)

        operr = OperationError(w_type, w_val, tb)
        w_value = operr.normalize_exception(space)

        # note: _w_yielded_from is always None if 'self.running'
        if (self.get_delegate() is not None and
                    operr.match(space, space.w_GeneratorExit)):
            try:
                self._gen_close_iter(space)
            except OperationError as e:
                return self.send_error(e)

        if tb is None:
            tb = space.getattr(w_value,
                               space.newtext('__traceback__'))
            if not space.is_w(tb, space.w_None):
                operr.set_traceback(tb)
        return self.send_error(operr)

    def _gen_close_iter(self, space):
        assert not self.running
        w_yf = self.get_delegate()
        self.set_delegate(None)
        self.running = True
        try:
            gen_close_iter(space, w_yf)
        finally:
            self.running = False

    def descr_close(self):
        """close() -> raise GeneratorExit inside generator/coroutine."""
        if self.frame is None:
            return     # nothing to do in this case
        space = self.space
        operr = None
        # note: _w_yielded_from is always None if 'self.running'
        w_yf = self.get_delegate()
        if w_yf is not None:
            try:
                self._gen_close_iter(space)
            except OperationError as e:
                operr = e
        if operr is None:
            operr = OperationError(space.w_GeneratorExit, space.w_None)
        try:
            self.send_error(operr)
        except OperationError as e:
            if e.match(space, space.w_StopIteration) or \
                    e.match(space, space.w_GeneratorExit):
                return space.w_None
            raise
        else:
            raise oefmt(space.w_RuntimeError,
                        "%s ignored GeneratorExit", self.KIND)

    def descr_gicr_frame(self, space):
        if self.frame is not None and not self.frame.frame_finished_execution:
            return self.frame
        else:
            return space.w_None

    def descr__name__(self, space):
        return space.newtext(self.get_name())

    def descr_set__name__(self, space, w_name):
        if space.isinstance_w(w_name, space.w_unicode):
            self._name = space.text_w(w_name)
        else:
            raise oefmt(space.w_TypeError,
                        "__name__ must be set to a string object")

    def descr__qualname__(self, space):
        return space.newtext(self.get_qualname())

    def descr_set__qualname__(self, space, w_name):
        try:
            self._qualname = space.utf8_w(w_name)
        except OperationError as e:
            if e.match(space, space.w_TypeError):
                raise oefmt(space.w_TypeError,
                            "__qualname__ must be set to a string object")
            raise

    def _finalize_(self):
        # This is only called if the CO_YIELD_INSIDE_TRY flag is set
        # on the code object.  If the frame is still not finished and
        # finally or except blocks are present at the current
        # position, then raise a GeneratorExit.  Otherwise, there is
        # no point.
        if self.frame is not None:
            block = self.frame.lastblock
            if block is not None:
                self.descr_close()

    def frame_is_finished(self):
        self.frame = None
        rgc.may_ignore_finalizer(self)

    def iterator_greenkey(self, space):
        return self.pycode

    def descr_get_suspended(self, space):
        return space.newbool(self.frame is not None and not self.running and self.frame.last_instr >= 0)


class GeneratorIterator(GeneratorOrCoroutine):
    "An iterator created by a generator."
    KIND = "generator"

    def descr__iter__(self):
        """Implement iter(self)."""
        return self

    def descr_next(self):
        """Implement next(self)."""
        return self.send_ex(self.space.w_None)

    # Results can be either an RPython list of W_Root, or it can be an
    # app-level W_ListObject, which also has an append() method, that's why we
    # generate 2 versions of the function and 2 jit drivers.
    def _create_unpack_into():
        jitdriver = jit.JitDriver(greens=['pycode'],
                                  reds='auto',
                                  name='unpack_into')

        def unpack_into(self, results):
            """This is a hack for performance: runs the generator and collects
            all produced items in a list."""
            frame = self.frame
            if frame is None:    # already finished
                return
            pycode = self.pycode
            while True:
                jitdriver.jit_merge_point(pycode=pycode)
                space = self.space
                try:
                    w_result = self._invoke_execute_frame(space.w_None)
                except OperationError as e:
                    if not e.match(space, space.w_StopIteration):
                        raise
                    break
                # if the frame is now marked as finished, it was RETURNed from
                if frame.frame_finished_execution:
                    self.frame_is_finished()
                    break
                results.append(w_result)     # YIELDed
        return unpack_into
    unpack_into = _create_unpack_into()
    unpack_into_w = _create_unpack_into()


GeneratorIterator.typedef = TypeDef("generator",
    __repr__   = interp2app(GeneratorIterator.descr__repr__),
    #__reduce__   = interp2app(GeneratorIterator.descr__reduce__),
    #__setstate__ = interp2app(GeneratorIterator.descr__setstate__),
    __next__   = interp2app(GeneratorIterator.descr_next,
                            descrmismatch='__next__'),
    send       = interp2app(GeneratorIterator.descr_send,
                            descrmismatch='send'),
    throw      = interp2app(GeneratorIterator.descr_throw,
                            descrmismatch='throw'),
    close      = interp2app(GeneratorIterator.descr_close,
                            descrmismatch='close'),
    __iter__   = interp2app(GeneratorIterator.descr__iter__,
                            descrmismatch='__iter__'),
    gi_running = interp_attrproperty('running', cls=GeneratorIterator, wrapfn="newbool"),
    gi_suspended = GetSetProperty(GeneratorIterator.descr_get_suspended),
    gi_frame   = GetSetProperty(GeneratorIterator.descr_gicr_frame),
    gi_code    = interp_attrproperty_w('pycode', cls=GeneratorIterator),
    gi_yieldfrom=GetSetProperty(GeneratorIterator.descr_delegate),
    __name__   = GetSetProperty(GeneratorIterator.descr__name__,
                                GeneratorIterator.descr_set__name__),
    __qualname__ = GetSetProperty(GeneratorIterator.descr__qualname__,
                                  GeneratorIterator.descr_set__qualname__),
    __weakref__ = make_weakref_descr(GeneratorIterator),
)
assert not GeneratorIterator.typedef.acceptable_as_base_class  # no __new__

class Coroutine(GeneratorOrCoroutine):
    "A coroutine object."
    KIND = "coroutine"

    def __init__(self, frame, name=None, qualname=None):
        GeneratorOrCoroutine.__init__(self, frame, name, qualname)
        self.w_cr_origin = self.space.w_None

    def capture_origin(self, ec):
        if not ec.coroutine_origin_tracking_depth:
            return
        self._capture_origin(ec)

    def _capture_origin(self, ec):
        space = self.space
        frames_w = []
        frame = ec.gettopframe_nohidden()
        for i in range(ec.coroutine_origin_tracking_depth):
            frames_w.append(
                space.newtuple([
                    frame.pycode.w_filename,
                    frame.fget_f_lineno(space),
                    space.newtext(frame.pycode.co_name)]))
            frame = ec.getnextframe_nohidden(frame)
            if frame is None:
                break
        self.w_cr_origin = space.newtuple(frames_w)

    def descr__await__(self, space):
        return CoroutineWrapper(self)

    def _finalize_(self):
        # If coroutine was never awaited on issue a RuntimeWarning.
        if (self.pycode is not None and
                self.frame is not None and
                self.frame.last_instr == -1):
            space = self.space
            w_mod = space.getbuiltinmodule("_warnings")
            w_f = space.getattr(w_mod, space.newtext("_warn_unawaited_coroutine"))
            space.call_function(w_f, self)
        GeneratorOrCoroutine._finalize_(self)

Coroutine.typedef = TypeDef("coroutine",
    __repr__   = interp2app(Coroutine.descr__repr__),
    #__reduce__   = interp2app(Coroutine.descr__reduce__),
    #__setstate__ = interp2app(Coroutine.descr__setstate__),
    send       = interp2app(Coroutine.descr_send,
                            descrmismatch='send'),
    throw      = interp2app(Coroutine.descr_throw,
                            descrmismatch='throw'),
    close      = interp2app(Coroutine.descr_close,
                            descrmismatch='close'),
    __await__  = interp2app(Coroutine.descr__await__,
                            descrmismatch='__await__'),
    cr_running = interp_attrproperty('running', cls=Coroutine, wrapfn="newbool"),
    cr_suspended = GetSetProperty(Coroutine.descr_get_suspended),
    cr_frame   = GetSetProperty(Coroutine.descr_gicr_frame),
    cr_code    = interp_attrproperty_w('pycode', cls=Coroutine),
    cr_await=GetSetProperty(Coroutine.descr_delegate),
    cr_origin  = interp_attrproperty_w('w_cr_origin', cls=Coroutine),
    __name__   = GetSetProperty(Coroutine.descr__name__,
                                Coroutine.descr_set__name__,
                                doc="name of the coroutine"),
    __qualname__ = GetSetProperty(Coroutine.descr__qualname__,
                                  Coroutine.descr_set__qualname__,
                                  doc="qualified name of the coroutine"),
    __weakref__ = make_weakref_descr(Coroutine),
)
assert not Coroutine.typedef.acceptable_as_base_class  # no __new__


class CoroutineWrapper(W_Root):
    _immutable_ = True

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

    def descr__iter__(self, space):
        return self

    def descr__next__(self, space):
        return self.coroutine.send_ex(space.w_None)

    def descr_send(self, space, w_arg):
        return self.coroutine.send_ex(w_arg)
    descr_send.__doc__ = Coroutine.descr_send.__doc__

    def descr_throw(self, w_type, w_val=None, w_tb=None):
        return self.coroutine.throw(w_type, w_val, w_tb)
    descr_throw.__doc__ = Coroutine.descr_throw.__doc__

    def descr_close(self):
        return self.coroutine.descr_close()
    descr_close.__doc__ = Coroutine.descr_close.__doc__

CoroutineWrapper.typedef = TypeDef("coroutine_wrapper",
    __iter__     = interp2app(CoroutineWrapper.descr__iter__),
    __next__     = interp2app(CoroutineWrapper.descr__next__),
    send         = interp2app(CoroutineWrapper.descr_send),
    throw        = interp2app(CoroutineWrapper.descr_throw),
    close        = interp2app(CoroutineWrapper.descr_close),
)
assert not CoroutineWrapper.typedef.acceptable_as_base_class  # no __new__


class AIterWrapper(W_Root):
    # NB. this type was added in CPython 3.5.2
    _immutable_ = True

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

    def descr__await__(self, space):
        return self

    def descr__iter__(self, space):
        return self

    def descr__next__(self, space):
        raise stopiteration_value(space, self.w_aiter)

AIterWrapper.typedef = TypeDef("aiter_wrapper",
    __await__    = interp2app(AIterWrapper.descr__await__),
    __iter__     = interp2app(AIterWrapper.descr__iter__),
    __next__     = interp2app(AIterWrapper.descr__next__),
)
assert not AIterWrapper.typedef.acceptable_as_base_class  # no __new__


def stopiteration_value(space, w_value):
    # Mess.  The obvious line, "OperationError(w_StopIteration, w_value)",
    # fails for some types of w_value.  E.g. if it's a subclass of
    # tuple, it will be unpacked as individual arguments.
    raise OperationError(space.w_StopIteration,
                         space.call_function(space.w_StopIteration, w_value))


def gen_close_iter(space, w_yf):
    # This helper function is used by close() and throw() to
    # close a subiterator being delegated to by yield-from.
    if isinstance(w_yf, GeneratorIterator) or isinstance(w_yf, Coroutine):
        w_yf.descr_close()
    else:
        try:
            w_close = space.getattr(w_yf, space.newtext("close"))
        except OperationError as e:
            if not e.match(space, space.w_AttributeError):
                # aaaaaaaah but that's what CPython does too
                e.write_unraisable(space, "generator/coroutine.close()")
        else:
            space.call_function(w_close)

def gen_is_coroutine(w_obj):
    return (isinstance(w_obj, GeneratorIterator) and
            (w_obj.pycode.co_flags & consts.CO_ITERABLE_COROUTINE) != 0)

def get_awaitable_iter(space, w_obj):
    # This helper function returns an awaitable for `o`:
    #    - `o` if `o` is a coroutine-object;
    #    - otherwise, o.__await__()

    if isinstance(w_obj, Coroutine) or gen_is_coroutine(w_obj):
        return w_obj

    w_await = space.lookup(w_obj, "__await__")
    if w_await is None:
        raise oefmt(space.w_TypeError,
                    "object %T can't be used in 'await' expression",
                    w_obj)
    w_res = space.get_and_call_function(w_await, w_obj)
    if isinstance(w_res, Coroutine) or gen_is_coroutine(w_res):
        raise oefmt(space.w_TypeError,
                    "__await__() returned a coroutine (it must return an "
                    "iterator instead, see PEP 492)")
    elif space.lookup(w_res, "__next__") is None:
        raise oefmt(space.w_TypeError,
                "__await__() returned non-iterator "
                "of type '%T'", w_res)
    return w_res


# ----------


def get_printable_location_genentry(bytecode):
    return '%s <generator>' % (bytecode.get_repr(),)
generatorentry_driver = jit.JitDriver(greens=['pycode'],
                                      reds=['gen', 'w_arg'],
                                      get_printable_location =
                                          get_printable_location_genentry,
                                      name='generatorentry')

from pypy.tool.stdlib_opcode import HAVE_ARGUMENT, opmap
YIELD_VALUE = opmap['YIELD_VALUE']
YIELD_FROM = opmap['YIELD_FROM']

@jit.elidable_promote()
def should_not_inline(pycode):
    # Should not inline generators with more than one "yield",
    # as an approximative fix (see issue #1782).  There are cases
    # where it slows things down; for example calls to a simple
    # generator that just produces a few simple values with a few
    # consecutive "yield" statements.  It fixes the near-infinite
    # slow-down in issue #1782, though...
    count_yields = 0
    code = pycode.co_code
    n = len(code)
    i = 0
    while i < n:
        c = code[i]
        op = ord(c)
        if op == YIELD_VALUE:
            count_yields += 1
        i += 1
        if op >= HAVE_ARGUMENT:
            i += 2
    return count_yields >= 2


# ------------------------------------------------
# Python 3.6 async generators


class AsyncGenerator(GeneratorOrCoroutine):
    "An async generator (i.e. a coroutine with a 'yield')"
    KIND = "async generator"

    def __init__(self, frame, name=None, qualname=None):
        self.hooks_inited = False
        self.ag_running = False
        GeneratorOrCoroutine.__init__(self, frame, name, qualname)

    def init_hooks(self):
        if self.hooks_inited:
            return
        self.hooks_inited = True
        ec = self.space.getexecutioncontext()
        self.w_finalizer = ec.w_asyncgen_finalizer_fn
        w_firstiter = ec.w_asyncgen_firstiter_fn
        if w_firstiter is not None:
            self.space.call_function(w_firstiter, self)

    def _finalize_(self):
        if self.frame is not None and self.frame.lastblock is not None:
            if self.w_finalizer is not None:
                # XXX: this is a hack to resurrect the weakref that was cleared
                # before running _finalize_()
                if self.space.config.translation.rweakref:
                    self.frame.f_generator_wref = rweakref.ref(self)
                try:
                    self.space.call_function(self.w_finalizer, self)
                except OperationError as e:
                    e.write_unraisable(self.space, "async generator finalizer")
                return
        GeneratorOrCoroutine._finalize_(self)

    def descr__aiter__(self):
        """Return an asynchronous iterator."""
        return self

    def descr__anext__(self):
        self.init_hooks()
        return AsyncGenASend(self, self.space.w_None)

    def descr_asend(self, w_arg):
        self.init_hooks()
        return AsyncGenASend(self, w_arg)

    def descr_athrow(self, w_type, w_val=None, w_tb=None):
        self.init_hooks()
        return AsyncGenAThrow(self, w_type, w_val, w_tb)

    def descr_aclose(self):
        self.init_hooks()
        return AsyncGenAThrow(self, None, None, None)

AsyncGenerator.typedef = TypeDef("async_generator",
    __repr__   = interp2app(AsyncGenerator.descr__repr__),
    #__reduce__   = interp2app(Coroutine.descr__reduce__),
    #__setstate__ = interp2app(Coroutine.descr__setstate__),
    asend      = interp2app(AsyncGenerator.descr_asend,
                            descrmismatch='asend'),
    athrow     = interp2app(AsyncGenerator.descr_athrow,
                            descrmismatch='athrow'),
    aclose     = interp2app(AsyncGenerator.descr_aclose,
                            descrmismatch='aclose'),
    __aiter__  = interp2app(AsyncGenerator.descr__aiter__,
                            descrmismatch='__aiter__'),
    __anext__  = interp2app(AsyncGenerator.descr__anext__,
                            descrmismatch='__anext__'),
    ag_running = interp_attrproperty('ag_running', cls=AsyncGenerator, wrapfn="newbool"),
    ag_frame   = GetSetProperty(AsyncGenerator.descr_gicr_frame),
    ag_code    = interp_attrproperty_w('pycode', cls=AsyncGenerator),
    ag_await=GetSetProperty(AsyncGenerator.descr_delegate),
    __name__   = GetSetProperty(AsyncGenerator.descr__name__,
                                AsyncGenerator.descr_set__name__,
                                doc="name of the async generator"),
    __qualname__ = GetSetProperty(AsyncGenerator.descr__qualname__,
                                  AsyncGenerator.descr_set__qualname__,
                                  doc="qualified name of the async generator"),
    __weakref__ = make_weakref_descr(AsyncGenerator),
    __class_getitem__ = interp2app(
        generic_alias_class_getitem, as_classmethod=True),
)
assert not AsyncGenerator.typedef.acceptable_as_base_class  # no __new__



class AsyncGenValueWrapper(W_Root):
    def __init__(self, w_value):
        self.w_value = w_value

AsyncGenValueWrapper.typedef = TypeDef("async_generator_wrapped_value")
assert not AsyncGenValueWrapper.typedef.acceptable_as_base_class


class AsyncGenABase(W_Root):
    ST_INIT = 0
    ST_ITER = 1
    ST_CLOSED = 2

    state = ST_INIT

    def __init__(self, async_gen):
        self.space = async_gen.space
        self.async_gen = async_gen

    def descr__iter__(self):
        return self

    def descr__next__(self):
        return self.do_send(self.space.w_None)

    def descr_send(self, w_arg):
        return self.do_send(w_arg)

    def descr_throw(self, w_type, w_val=None, w_tb=None):
        return self.do_throw(w_type, w_val, w_tb)

    def descr_close(self):
        self.state = self.ST_CLOSED

    def unwrap_value(self, w_value):
        if isinstance(w_value, AsyncGenValueWrapper):
            w_value = self.space.call_function(self.space.w_StopIteration,
                                               w_value.w_value)
            self.async_gen.ag_running = False
            raise OperationError(self.space.w_StopIteration, w_value)
        else:
            return w_value


class AsyncGenASend(AsyncGenABase):

    def __init__(self, async_gen, w_value_to_send):
        AsyncGenABase.__init__(self, async_gen)
        self.w_value_to_send = w_value_to_send

    def do_send(self, w_arg_or_err):
        space = self.space
        if self.state == self.ST_CLOSED:
            raise OperationError(
                self.space.w_RuntimeError,
                self.space.newtext("cannot reuse already awaited __anext__()/asend()")
            )

        # We think that the code should look like this:
        #if self.w_value_to_send is not None:
        #    if not space.is_w(w_arg_or_err, space.w_None):
        #        raise ...
        #    w_arg_or_err = self.w_value_to_send
        #    self.w_value_to_send = None

        # But instead, CPython's logic is this, which we think is
        # giving nonsense results for 'g.asend(42).send(43)':
        if self.state == self.ST_INIT:
            if space.is_w(w_arg_or_err, space.w_None):
                w_arg_or_err = self.w_value_to_send
            self.state = self.ST_ITER
            if self.async_gen.ag_running:
                raise oefmt(space.w_RuntimeError,
                    "asynchronous generator is already running")
            self.async_gen.ag_running = True

        try:
            w_value = self.async_gen.send_ex(w_arg_or_err)
            return self.unwrap_value(w_value)
        except OperationError as e:
            self.async_gen.ag_running = False
            self.state = self.ST_CLOSED
            raise

    def do_throw(self, w_type, w_val, w_tb):
        space = self.space
        if self.state == self.ST_CLOSED:
            raise OperationError(
                self.space.w_RuntimeError,
                self.space.newtext("cannot reuse already awaited __anext__()/asend()")
            )
        try:
            w_value = self.async_gen.throw(w_type, w_val, w_tb)
            return self.unwrap_value(w_value)
        except OperationError as e:
            self.async_gen.ag_running = False
            self.state = self.ST_CLOSED
            raise

AsyncGenASend.typedef = TypeDef("async_generator_asend",
    __await__    = interp2app(AsyncGenASend.descr__iter__),
    __iter__     = interp2app(AsyncGenASend.descr__iter__),
    __next__     = interp2app(AsyncGenASend.descr__next__),
    close        = interp2app(AsyncGenASend.descr_close),
    send         = interp2app(AsyncGenASend.descr_send),
    throw        = interp2app(AsyncGenASend.descr_throw),
)


class AsyncGenAThrow(AsyncGenABase):

    def __init__(self, async_gen, w_exc_type, w_exc_value, w_exc_tb):
        AsyncGenABase.__init__(self, async_gen)
        self.w_exc_type = w_exc_type
        self.w_exc_value = w_exc_value
        self.w_exc_tb = w_exc_tb

    def do_send(self, w_arg_or_err):
        # XXX FAR MORE COMPLICATED IN CPYTHON
        space = self.space
        if self.state == self.ST_CLOSED:
            raise OperationError(
                self.space.w_RuntimeError,
                self.space.newtext("cannot reuse already awaited aclose()/athrow()")
            )

        if self.state == self.ST_INIT:
            if not space.is_w(w_arg_or_err, space.w_None):
                raise OperationError(space.w_RuntimeError, space.newtext(
                    "can't send non-None value to a just-started coroutine"))
            self.state = self.ST_ITER
            throwing = True
        else:
            throwing = False

        try:
            if throwing:
                if self.w_exc_type is None:
                    # TODO: add equivalent to CPython's o->agt_gen->ag_closed = 1;
                    w_value = self.async_gen.throw(space.w_GeneratorExit,
                                                   None, None)
                else:
                    w_value = self.async_gen.throw(self.w_exc_type,
                                                   self.w_exc_value,
                                                   self.w_exc_tb)
            else:
                w_value = self.async_gen.send_ex(w_arg_or_err)

            if (self.w_exc_type is None and w_value is not None and
                    isinstance(w_value, AsyncGenValueWrapper)):
                raise oefmt(space.w_RuntimeError,
                            "async generator ignored GeneratorExit")
            return self.unwrap_value(w_value)
        except OperationError as e:
            self.handle_error(e)

    def do_throw(self, w_type, w_val, w_tb):
        space = self.space
        if self.state == self.ST_CLOSED:
            raise OperationError(
                self.space.w_RuntimeError,
                self.space.newtext("cannot reuse already awaited aclose()/athrow()")
            )
        try:
            w_value = self.async_gen.throw(w_type, w_val, w_tb)
            return self.unwrap_value(w_value)
        except OperationError as e:
            self.handle_error(e)

    def handle_error(self, e):
        space = self.space
        self.state = self.ST_CLOSED
        if (e.match(space, space.w_StopAsyncIteration) or
                    e.match(space, space.w_StopIteration)):
            if self.w_exc_type is None:
                # When aclose() is called we don't want to propagate
                # StopAsyncIteration or GeneratorExit; just raise
                # StopIteration, signalling that this 'aclose()' await
                # is done.
                raise OperationError(space.w_StopIteration, space.w_None)
        if e.match(space, space.w_GeneratorExit):
            if self.w_exc_type is None:
                # Ignore this error.
                raise OperationError(space.w_StopIteration, space.w_None)
        raise e

AsyncGenAThrow.typedef = TypeDef("async_generator_athrow",
    __await__    = interp2app(AsyncGenAThrow.descr__iter__),
    __iter__     = interp2app(AsyncGenAThrow.descr__iter__),
    __next__     = interp2app(AsyncGenAThrow.descr__next__),
    close        = interp2app(AsyncGenAThrow.descr_close),
    send         = interp2app(AsyncGenAThrow.descr_send),
    throw        = interp2app(AsyncGenAThrow.descr_throw),
)