python - 如何获得对生成器发送方法的弱引用？

问题描述

该weakref文档似乎没有提供一种方法来创建对send生成器方法的弱引用：

import weakref

def gen(): yield

g=gen()
w_send=weakref.ref(g.send)
w_send() # <- this is None; the g.send object is ephemeral

我不认为它会起作用，但我也尝试了weakref.WeakMethod以防万一：

>>> w_send=weakref.WeakMethod(g.send)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "C:\Users\ricky\AppData\Local\Programs\Python\Python37\lib\weakref.py", line 50, in __new__
    .format(type(meth))) from None
TypeError: argument should be a bound method, not <class 'builtin_function_or_method'>

如果不将生成器包装在自定义类中，如何做到这一点？像这样：

import weakref

def gen(): yield

class MyGenerator:
    def __init__(self):
        self._generator = gen()
    def send(self, arg):
        return self._generator.send(arg)

g = MyGenerator()
ref = weakref.WeakMethod(g.send)

我不想这样做。有没有更好的办法？

我想这样做的原因是我正在为我可能构建的应用程序制定一个简单的消息传递协议的想法。消息看起来像这样：

# messaging module

from typing import Generator
from functools import wraps
from collections import NamedTuple
import weakref

class Message(NamedTuple):
    channel: int
    content: str

_REGISTRY = {}

def _do_register(channel, route):
    # handle bound methods
    if hasattr(route,"__self__") and hasattr(route,"__func__"):
        route_ref = weakref.WeakMethod(route)
    # handle generators
    elif isinstance(route, Generator):
        route_ref = weakref.ref(route.send) # get weak ref to route.send here
    # all other callables
    else:
        route_ref = weakref.ref(route)
    try:
        _REGISTRY[channel].add(route_ref)
    except KeyError:
        _REGISTRY[channel] = {route_ref}

def register(obj=None, *, channel, route=None):
    """Decorator for registering callable objects for messaging."""
    if obj is None:
        def wrapper(callable):
            @wraps(callable)
            def wrapped(*args, **kwargs):
                nonlocal route
                obj_ = callable(*args, **kwargs)
                route_ = obj_ if route is None else route
                _do_register(channel, route_)
                return obj_
            return wrapped
        return wrapper
    else:
        if route is None:
            route = obj
        _do_register(channel, route)

def manager():
    msg_obj = None
    while True:
        msg_obj = yield _broadcast(msg_obj)

def _broadcast(msg_obj):
    count = 0
    if msg_obj:
        for route_ref in _REGISTRY[msg_obj.channel]:
            route = route_ref()
            if route is not None:
                count += 1
                route(msg_obj)
    return count

...这样使用：

@register(channel=1)
def listening_gen(name):
    while True:
        msg = yield
        print(f"{name} received message {msg.content} on channel {msg.channel}")


a = listening_gen("a")
b = listening_gen("b")
next(a)
next(b)
register(a, channel=2)
register(b, channel=3)

msg1 = Message(channel=1, content="foo")
msg2 = Message(channel=2, content="bar")
msg3 = Message(channel=3, content="baz")

m = manager()
next(m)
m.send(msg1)
m.send(msg2)
m.send(msg3)

a收听频道 1 和 2 上的b消息，收听频道 1 和 3 上的消息。

标签： pythonpython-3.xgeneratorweak-references

python - 如何获得对生成器发送方法的弱引用？

问题描述

解决方案

推荐阅读