- c - 在位数组中找到第一个零
- linux - Unix 显示有关匹配两种模式之一的文件的信息
- 正则表达式替换多个文件
- linux - 隐藏来自 xtrace 的命令
(注意:这个问题的背景非常冗长,但底部有一个 SSCCE 可以跳过)
背景
我正在尝试开发一个基于 Python 的 CLI 来与 Web 服务进行交互。在我的代码库中,我有一个 CommunicationService
类,用于处理与 Web 服务的所有直接通信。它公开了一个 received_response
属性,该属性返回一个 Observable
(来自 RxPY),其他对象可以订阅该代码,以便在从 Web 服务收到响应时收到通知。
我的 CLI 逻辑基于 click
库,其中一个子命令实现如下:
async def enabled(self, request: str, response_handler: Callable[[str], Tuple[bool, str]]) -> None:
self._generate_request(request)
if response_handler is None:
return None
while True:
response = await self.on_response
success, value = response_handler(response)
print(success, value)
if success:
return value
response_handler
不是
None
的情况下)是该子命令表现为一个协程,等待来自 Web 服务(
self.on_response == CommunicationService.received_response
)的响应并从它可以处理的第一个响应中返回一些处理过的值。
CommunicationService
的测试用例来测试我的 CLI 的行为;创建了一个假的
Subject
(它可以充当
Observable
)并且
CommunicationService.received_response
被模拟以返回它。作为测试的一部分,主题的
on_next
方法被调用以将模拟 Web 服务响应传递回生产代码:
@when('the communications service receives a response from TestCube Web Service')
def step_impl(context):
context.mock_received_response_subject.on_next(context.text)
@cli.resultcallback()
def _handle_command_task(task: Coroutine, **_) -> None:
if task:
loop = asyncio.get_event_loop()
result = loop.run_until_complete(task)
loop.close()
print('RESULT:', result)
CliRunner.invoke
来触发整个 shebang。问题是这是一个阻塞调用,它将阻塞线程,直到 CLI 完成并返回结果,如果我需要我的测试线程继续运行,它就可以与它同时生成模拟 Web 服务响应,这没有帮助。
CliRunner.invoke
在新线程上运行
ThreadPoolExecutor
。这允许测试逻辑在原始线程上继续并执行上面发布的
@when
步骤。但是,使用
mock_received_response_subject.on_next
发布的通知似乎不会触发在子命令中继续执行。
AsyncIOScheduler
,但我发现有关此的文档有点稀疏且无用。
import asyncio
import logging
import sys
import time
import click
from click.testing import CliRunner
from rx.subjects import Subject
web_response_subject = Subject()
web_response_observable = web_response_subject.as_observable()
thread_loop = asyncio.new_event_loop()
@click.group()
def cli():
asyncio.set_event_loop(thread_loop)
@cli.resultcallback()
def result_handler(task, **_):
loop = asyncio.get_event_loop()
result = loop.run_until_complete(task) # Should block until subject publishes value
loop.close()
print(result)
@cli.command()
async def get_web_response():
return await web_response_observable
def test():
runner = CliRunner()
future = thread_loop.run_in_executor(None, runner.invoke, cli, ['get_web_response'])
time.sleep(1)
web_response_subject.on_next('foo') # Simulate reception of web response.
time.sleep(1)
result = future.result()
print(result.output)
logging.basicConfig(
level=logging.DEBUG,
format='%(threadName)10s %(name)18s: %(message)s',
stream=sys.stderr,
)
test()
result = loop.run_until_complete(task)
处阻塞。
foo
上打印
stdout
。
Relay.enabled
(等待来自 Web 服务的响应以处理它们的子命令)现在实现如下:
async def enabled(self, request: str, response_handler: Callable[[str], Tuple[bool, str]]) -> None:
self._generate_request(request)
if response_handler is None:
return None
return await self.on_response \
.select(response_handler) \
.where(lambda result, i: result[0]) \
.select(lambda result, index: result[1]) \
.first()
await
将如何处理
RxPY
observables - 它们会在每个生成的元素上将执行返回给调用者,还是仅当 observable 完成(或出错?)。我现在知道是后者,老实说,这感觉是更自然的选择,并且让我使这个函数的实现感觉更加优雅和 react 性。
@when('the communications service receives a response from TestCube Web Service')
def step_impl(context):
loop = asyncio.get_event_loop()
loop.call_soon_threadsafe(context.mock_received_response_subject.on_next, context.text)
@when('the CLI is run with "{arguments}"')
def step_impl(context, arguments):
loop = asyncio.get_event_loop()
if 'async.cli' in context.tags:
context.async_result = loop.run_in_executor(None, context.cli_runner.invoke, testcube.cli, arguments.split())
else:
...
def cli(context, hostname, port):
_initialize_logging(context.meta['click_log.core.logger']['level'])
# Create a new event loop for processing commands asynchronously on.
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
...
@when('the communications service receives a response from TestCube Web Service')
def step_impl(context):
loop = _get_cli_event_loop() # Needs to be implemented.
loop.call_soon_threadsafe(context.mock_received_response_subject.on_next, context.text)
asyncio.new_event_loop
以返回我的测试代码创建和存储的事件循环:
def _apply_mock_event_loop_patch(context):
# Close any already-existing exit stacks.
if hasattr(context, 'mock_event_loop_exit_stack'):
context.mock_event_loop_exit_stack.close()
context.test_loop = asyncio.new_event_loop()
print(context.test_loop)
context.mock_event_loop_exit_stack = ExitStack()
context.mock_event_loop_exit_stack.enter_context(
patch.object(asyncio, 'new_event_loop', spec=True, return_value=context.test_loop))
@when('the communications service receives a response from TestCube Web Service')
def step_impl(context):
loop = context.test_loop
loop.call_soon_threadsafe(context.mock_received_response_subject.on_next, context.text)
Relay.enabled
协程触发!
stdout
上发送:
@then('the CLI should print "{output}"')
def step_impl(context, output):
if 'async.cli' in context.tags:
loop = asyncio.get_event_loop() # main loop, not test loop
result = loop.run_until_complete(context.async_result)
else:
result = context.result
assert_that(result.output, equal_to(output))
context.async_result
(从
loop.run_in_executor
存储 future )很好地转换到
done
并返回结果。使用当前的实现,我得到第一个测试 (
1.1
) 的错误和第二个 (
1.2
) 的无限期挂起:
@mock.comms @async.cli @wip
Scenario Outline: Querying relay enable state -- @1.1 # testcube/tests/features/relay.feature:45
When the user queries the enable state of relay 0 # testcube/tests/features/steps/relay.py:17 0.003s
Then the CLI should query the web service about the enable state of relay 0 # testcube/tests/features/steps/relay.py:48 0.000s
When the communications service receives a response from TestCube Web Service # testcube/tests/features/steps/core.py:58 0.000s
"""
{'module':'relays','path':'relays[0].enabled','data':[True]}'
"""
Then the CLI should print "True" # testcube/tests/features/steps/core.py:94 0.003s
Traceback (most recent call last):
File "/Users/davidfallah/testcube_env/lib/python3.5/site-packages/behave/model.py", line 1456, in run
match.run(runner.context)
File "/Users/davidfallah/testcube_env/lib/python3.5/site-packages/behave/model.py", line 1903, in run
self.func(context, *args, **kwargs)
File "testcube/tests/features/steps/core.py", line 99, in step_impl
result = loop.run_until_complete(context.async_result)
File "/usr/local/Cellar/python3/3.5.2_1/Frameworks/Python.framework/Versions/3.5/lib/python3.5/asyncio/base_events.py", line 387, in run_until_complete
return future.result()
File "/usr/local/Cellar/python3/3.5.2_1/Frameworks/Python.framework/Versions/3.5/lib/python3.5/asyncio/futures.py", line 274, in result
raise self._exception
File "/usr/local/Cellar/python3/3.5.2_1/Frameworks/Python.framework/Versions/3.5/lib/python3.5/concurrent/futures/thread.py", line 55, in run
result = self.fn(*self.args, **self.kwargs)
File "/Users/davidfallah/testcube_env/lib/python3.5/site-packages/click/testing.py", line 299, in invoke
output = out.getvalue()
ValueError: I/O operation on closed file.
Captured stdout:
RECEIVED WEB RESPONSE: {'module':'relays','path':'relays[0].enabled','data':[True]}'
<Future pending cb=[_chain_future.<locals>._call_check_cancel() at /usr/local/Cellar/python3/3.5.2_1/Frameworks/Python.framework/Versions/3.5/lib/python3.5/asyncio/futures.py:431]>
@mock.comms @async.cli @wip
Scenario Outline: Querying relay enable state -- @1.2 # testcube/tests/features/relay.feature:46
When the user queries the enable state of relay 1 # testcube/tests/features/steps/relay.py:17 0.005s
Then the CLI should query the web service about the enable state of relay 1 # testcube/tests/features/steps/relay.py:48 0.001s
When the communications service receives a response from TestCube Web Service # testcube/tests/features/steps/core.py:58 0.000s
"""
{'module':'relays','path':'relays[1].enabled','data':[False]}'
"""
RECEIVED WEB RESPONSE: {'module':'relays','path':'relays[1].enabled','data':[False]}'
Then the CLI should print "False" # testcube/tests/features/steps/core.py:94
When the user queries the enable state of relay <relay_id>
Then the CLI should query the web service about the enable state of relay <relay_id>
When the communications service receives a response from TestCube Web Service:
"""
{"module":"relays","path":"relays[<relay_id>].enabled","data":[<relay_enabled>]}
"""
Then the CLI should print "<relay_enabled>"
Given the communications service will respond to requests:
"""
{"module":"relays","path":"relays[<relay_id>].enabled","data":[<relay_enabled>]}
"""
When the user queries the enable state of relay <relay_id>
Then the CLI should query the web service about the enable state of relay <relay_id>
And the CLI should print "<relay_enabled>"
@given('the communications service will respond to requests')
def step_impl(context):
response = context.text
def publish_mock_response(_):
loop = context.test_loop
loop.call_soon_threadsafe(context.mock_received_response_subject.on_next, response)
# Configure the mock comms service to publish a mock response when a request is made.
instance = context.mock_comms.return_value
instance.send_request.on_next.side_effect = publish_mock_response
2 features passed, 0 failed, 0 skipped
22 scenarios passed, 0 failed, 0 skipped
58 steps passed, 0 failed, 0 skipped, 0 undefined
Took 0m0.111s
最佳答案
我可以看到您的代码有两个问题:
RxPy
不使用 Observable.to_future 中的任何内容,所以你必须访问 RxPy
运行 asyncio 事件循环的同一线程中的对象。 RxPy
当 on_completed
时设置 future 的结果被调用,以便等待 observable 返回最后发出的对象。这意味着您必须同时拨打 on_next
和 on_completed
获取 await
返回。 import click
import asyncio
from rx.subjects import Subject
from click.testing import CliRunner
web_response_subject = Subject()
web_response_observable = web_response_subject.as_observable()
main_loop = asyncio.get_event_loop()
@click.group()
def cli():
pass
@cli.resultcallback()
def result_handler(task, **_):
future = asyncio.run_coroutine_threadsafe(task, main_loop)
print(future.result())
@cli.command()
async def get_web_response():
return await web_response_observable
def test():
runner = CliRunner()
future = main_loop.run_in_executor(
None, runner.invoke, cli, ['get_web_response'])
main_loop.call_later(1, web_response_subject.on_next, 'foo')
main_loop.call_later(2, web_response_subject.on_completed)
result = main_loop.run_until_complete(future)
print(result.output, end='')
if __name__ == '__main__':
test()
关于python - 如何使用 asyncio 在单独的线程上通知 RxPY 观察者?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/39318723/
我有一个应用程序应该在应用程序处于前台和后台(不在历史记录中)时显示提醒通知。 在前景情况下,我通过以下方法实现了这一点。 PendingIntent pendingIntent = PendingI
如何为我的 WPF 应用程序创建通知,例如浏览器上的通知,它们通过浏览器顶部的“工具栏”显示消息或通过在右下角向上/向下滑动的弹出窗口显示“MSN”样式通知屏幕。也许在应用程序中心淡入/淡出的面板可以
关闭。这个问题是opinion-based .它目前不接受答案。 想要改进这个问题? 更新问题,以便 editing this post 可以用事实和引用来回答它. 关闭 9 年前。 Improve
我正在使用 Redis 作为分布式缓存。我有不同的应用程序,它们只听特定的键。例如:App1 听 App1.*App2 监听 App2.* 等等。 我的应用程序使用以下模式接收通知:App1:“ ke
我正在尝试构建一个基于官方节点 docker 镜像的 docker 镜像,我想知道是否有某种方法可以在推送新版本的官方节点镜像时自动重建镜像。这样我的图像就不会基于过时的基础图像。 也许有类似 rss
我在一个项目中工作,我需要在添加或修改文件时在数据库中记录文件信息,以便它们保持同步。这些文件应该存储在 Nextcloud 服务器中,那么 Nextcloud 是否有办法通知这些更改(例如 webh
通知类中的方法via 如何根据用户的偏好动态变化,一个用户可能想通过电子邮件接收,而另一个用户则不想 public function via($notifiable) { return ['d
我有一个应用程序,我正在发送推送通知,如果用户登录到应用程序,这很好 - 但是,如果他们没有/如果他们没有在 X 分钟内阅读通知,我想给他们发送一封电子邮件. 我要解决的方法是使用 Laravel N
我正在使用 Django 的 contrib.comments 并想了解以下内容。 是否有任何实用程序或应用程序可以插入到某个应用程序中,当对某个项目发表评论时向您发送通知? 我并没有真正使用过那么多
我希望用户在启动应用程序之前接受协议(protocol)。所以在 appDelegate.m 中我有以下内容: - (BOOL)application:(UIApplication *)applica
我正在创建一个新指令,我想知道如何在 angular 从 DOM 中删除元素时收到通知。 我的目标是在删除元素时添加 jquery 动画。 最佳答案 如果您尝试对元素的移除进行动画处理,则需要在移除元
我正在编写一个应用程序,其工作方式与Apple的Weather.app非常相似:底部有一个UIPageControl,屏幕中间有一个UIScrollView。在我的代码中,我实现了 - (void)s
如何查明 iPhone 注册了哪些通知? 例如: notify_post("com.apple.springboard/Prefs"); 最佳答案 虽然这个问题的答案已经得到确认,但由于 @Nate
我的 Cocoa 应用程序中有一个 TextField。该文本字段有时会被填充,有时会为空。 我希望当字段为空时按钮被禁用。现在,每当我对 Core Data 执行某些操作时,我都会检查该字段,Tex
我的应用程序在其数据库中包含文档。用户可以打开文档,在这种情况下,文档将保存到临时文件夹并在用户计算机上打开。 我希望在这些临时文件之一发生更改时收到通知,并让用户将更改后的文档保存回数据库。 在 D
我目前正在开发一个网络应用程序,它不断对 php 进行 ajax 调用(轮询),以从数据库中提取新的“任务”,有点像 gmail/facebook 检查新电子邮件和消息的方式。当前的 JavaScri
我正在尝试让通知适用于我使用 Angular 5 和 Electron 制作的 Electron 应用程序。到目前为止,我的 index.html 文件中有以下代码: function doNo
我有一个录音/播放应用程序。它在后台运行。当它进入后台时,如果任何其他音频应用程序打开或开始使用音频资源,我想适本地处理我的应用程序。 iOS 提供了一种发送此类通知的方法,如在 ipod 播放器中看
关闭。这个问题需要多问focused 。目前不接受答案。 想要改进此问题吗?更新问题,使其仅关注一个问题 editing this post . 已关闭 4 年前。 Improve this ques
是否有 Subversion 的工具可以在对某些文件提交更改时自动通知我? 最佳答案 您可以创建一个 post-commit hook script “ Hook ”提交。 在钩子(Hook)脚本中,
我是一名优秀的程序员,十分优秀!