观测器设计模式的一种简洁实现

GreenRocket的Python项目详细描述


绿色火箭

绿色火箭是一种简单紧凑的观测器实现 (或发布/订阅)通过信号的设计模式。

使用Base One创建特定信号:

>>> fromgreenrocketimportSignal>>> classMySignal(Signal):... pass...

订阅处理程序:

>>> @MySignal.subscribe... defhandler(signal):... print('handler: '+repr(signal))...

火警信号:

>>> MySignal().fire()handler: MySignal()

如果使用的是asyncio,则还可以使用协程作为处理程序 并使用await Signal.afire()异步触发信号 yield from Signal().afire()。方法afire()与 同步处理程序。

注意,该信号在继承上传播,即基址的所有订户 启动子一时将调用信号:

>>> @Signal.subscribe... defbase_handler(signal):... print('base_handler: '+repr(signal))...>>> MySignal().fire()handler: MySignal()
base_handler: MySignal()

取消订阅处理程序:

>>> MySignal.unsubscribe(handler)>>> MySignal().fire()base_handler: MySignal()

处理程序是使用弱引用订阅的。所以如果你创建并订阅 本地作用域中的处理程序(例如在生成器中),它将 自动取消订阅。

>>> defgen():... @MySignal.subscribe... deflocal_handler(signal):... print('local_handler: '+repr(signal))... yield1...>>> forvalueingen():... MySignal(value=value).fire()...local_handler: MySignal(value=1)
base_handler: MySignal(value=1)
>>> importgc# PyPy fails the following test without>>> _=gc.collect()# explicit call of garbage collector.>>> MySignal(value=2).fire()base_handler: MySignal(value=2)
>>> Signal.unsubscribe(base_handler)

如上所示,信号构造函数接受关键字参数。这些 参数可用作信号的属性:

>>> s=MySignal(a=1,b=2)>>> s.a1
>>> s.b2

信号抑制处理程序调用时引发的任何异常。它使用 名为greenrocket的记录器,来自标准logging模块,用于记录错误和 调试信息。

库还提供了Watchman类作为一种方便的测试方法 信号。

为特定信号创建监视人:

>>> fromgreenrocketimportWatchman>>> watchman=Watchman(MySignal)

火警信号:

>>> MySignal(x=1).fire()

测试信号:

>>> watchman.assert_fired_with(x=1)>>> watchman.assert_fired_with(x=2)# DOCTEST: +ellipsisTraceback (most recent call last):
...AssertionError: Failed assertion on MySignal.x: 1 != 2>>> watchman.assert_fired_with(x=1,y=2)# DOCTEST: +ellipsisTraceback (most recent call last):
...AssertionError: MySignal has no attribute y

watchman对象将每个激发的信号保存到其日志:

>>> watchman.log[MySignal(x=1)]
>>> MySignal(x=2).fire()>>> watchman.log[MySignal(x=1), MySignal(x=2)]

方法assert_fired_with测试来自 默认日志:

>>> watchman.assert_fired_with(x=2)

但您可以指定要测试的对象:

>>> watchman.assert_fired_with(-2,x=1)

更改

0.30

  • 添加了返回awaitable to support的Signal.afire()方法 基于协程的信号处理程序
  • 放弃了对Python2.6和3.2的支持

0.22

  • 添加了作为测试助手的watchman类

0.21

  • 已删除分发依赖项
  • 改进的测试

0.20

  • 已通过引用将处理程序订阅机制从订阅更改为 弱引用订阅

0.11

  • 修复了程序终止时记录器松动的问题

0.1

  • 初始版本

欢迎加入QQ群-->: 979659372 Python中文网_新手群

推荐PyPI第三方库


热门话题
java Hibernate会话/模板返回null   java Bluetooth无法连接到OBD2   smartGWT中的行分隔符的java等价物?   java如何删除可执行路径错误   java使用ICE Tar库对大文件进行Tar   java Spring Boot/JavaFX:添加一个仍然可以访问JPA的关机挂钩(ctrlc)   使用输入流读取java属性文件   如何在Java中迭代HashSet元素时覆盖该元素   内容类型中的javascript set charset=“utf8”   使用cassandra触发器的java   用户界面什么是适合我的GUI(JAVA)的update()方法   Java实例控制   java前置摄像头无法在不同设备上工作   java方法组织。postgresql。jdbc。连接。createClob()尚未实现