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

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第三方库


热门话题
使用jaxrpc的Java eclipse WebService客户端   java编程方式在对象上写入名称   java Spring批处理:重试后跳过   java Android错误:错误:任务执行失败:应用程序:transformClassesWithDexForDebug'   带有清单文件nullPointerException的java Android元数据   spring Java Quartz调度作业停止运行   JavaMockito:如何在不调用实际方法的情况下,模拟带有参数和无效返回类型的静态方法?   java Tomcat连接池问题无法在关闭的连接上调用方法   java如何交换列表中的项目?   java如何停止线程并通过Toast在线程中正确显示文本?   java为什么连续写入OutputStream时偏移量0不会导致重复字节?   java我无法生成头文件   不兼容的返回类型错误java   修改值后键值对的java Jolt转换规范   java有自动更新Javadoc的工具吗?   java线程如何在ints自身实例类中共享变量   java继承一个非gwt模块   java Hibernate xml配置   使用netty4异步调用的java链接HTTP请求响应