• 设为首页
  • 点击收藏
  • 手机版
    手机扫一扫访问
    迪恩网络手机版
  • 关注官方公众号
    微信扫一扫关注
    迪恩网络公众号

Python endpoint.ListeningBaseEndpoint类代码示例

原作者: [db:作者] 来自: [db:来源] 收藏 邀请

本文整理汇总了Python中pyon.net.endpoint.ListeningBaseEndpoint的典型用法代码示例。如果您正苦于以下问题:Python ListeningBaseEndpoint类的具体用法?Python ListeningBaseEndpoint怎么用?Python ListeningBaseEndpoint使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。



在下文中一共展示了ListeningBaseEndpoint类的9个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。

示例1: test_get_stats

    def test_get_stats(self):
        ep = ListeningBaseEndpoint()
        ep._chan = Mock(spec=ListenChannel)

        ep.get_stats()

        ep._chan.get_stats.assert_called_once_with()
开发者ID:daf,项目名称:pyon,代码行数:7,代码来源:test_endpoint.py


示例2: _setup_listener

    def _setup_listener(self, name, binding=None):
        """
        Override Subscriber - we may have specified a queue name.

        XXX BASECLASS SEEMS WRONG
        """
        ListeningBaseEndpoint._setup_listener(self, name, binding=binding)
开发者ID:blazetopher,项目名称:pyon,代码行数:7,代码来源:event.py


示例3: test_listen_exception_in_handling

    def test_listen_exception_in_handling(self):

        # make a listen loop that will return one message (to blow up in processing)
        chmock = MagicMock(spec=ListenChannel)
        chmock.accept.return_value = Mock()
        chmock.accept.return_value.recv = Mock(return_value=(sentinel.msg, sentinel.headers, sentinel.delivery_tag))
        chmock.accept.return_value._recv_queue.qsize.return_value = 1

        nodemock = Mock(spec=NodeB)
        nodemock.channel.return_value = chmock

        recv_name = NameTrio(sentinel.ex, sentinel.queue)

        ep = ListeningBaseEndpoint(node=nodemock, from_name=recv_name)

        # make msg received error out!
        ep.create_endpoint = Mock(return_value=Mock(spec=EndpointUnit))
        ep.create_endpoint.return_value._message_received.side_effect = TestError
        ep.create_endpoint.return_value.intercept_in.return_value = (sentinel.msg, sentinel.headers)

        self.assertRaises(TestError, ep.listen)
        chmock.setup_listener.assert_called_once_with(recv_name, binding=sentinel.queue)
        chmock.start_consume.assert_called_once_with()

        chmock.accept.assert_called_once_with(n=1, timeout=None)
        chmock.accept.return_value.recv.assert_called_once_with()
        ep.create_endpoint.assert_called_once_with(existing_channel=chmock.accept.return_value)
开发者ID:daf,项目名称:pyon,代码行数:27,代码来源:test_endpoint.py


示例4: test__create_channel_sets_transport_kwarg

    def test__create_channel_sets_transport_kwarg(self):
        # if send_name is a transport, it makes sure that kwarg is passed in to node's channel (and therefore the channel)
        class FakeSendName(NameTrio, BaseTransport):
            pass

        fn = FakeSendName()
        ep = ListeningBaseEndpoint(node=Mock(spec=NodeB), from_name=fn)

        ch = ep._create_channel()
        self.assertIn('transport', ep.node.channel.call_args[1])
        self.assertIn(fn, ep.node.channel.call_args[1].itervalues())
开发者ID:daf,项目名称:pyon,代码行数:11,代码来源:test_endpoint.py


示例5: test_listen

    def test_listen(self):
        # make a listen loop that will exit right away
        chmock = Mock(spec=ListenChannel)
        chmock.accept.side_effect = ChannelClosedError

        nodemock = Mock(spec=NodeB)
        nodemock.channel.return_value = chmock

        ep = ListeningBaseEndpoint(node=nodemock, from_name=NameTrio(sentinel.ex, sentinel.queue))
        ep.listen()

        chmock.setup_listener.assert_called_once_with(ep._recv_name, binding=sentinel.queue)
开发者ID:daf,项目名称:pyon,代码行数:12,代码来源:test_endpoint.py


示例6: test_listen_with_base_transport_for_name

    def test_listen_with_base_transport_for_name(self):

        # make a listen loop that will exit right away
        chmock = Mock(spec=ListenChannel)
        chmock.accept.side_effect = ChannelClosedError

        nodemock = Mock(spec=NodeB)
        nodemock.channel.return_value = chmock

        class FakeRecvName(BaseTransport, NameTrio):
            pass
        recv_name = FakeRecvName()
        recv_name.setup_listener = Mock()

        ep = ListeningBaseEndpoint(node=nodemock, from_name=recv_name)
        ep.listen(binding=sentinel.binding)

        self.assertTrue(ep.get_ready_event().is_set())
        self.assertIn('transport', nodemock.channel.call_args[1])
        self.assertIn(recv_name, nodemock.channel.call_args[1].itervalues())
开发者ID:daf,项目名称:pyon,代码行数:20,代码来源:test_endpoint.py


示例7: test_get_stats_multiple_on_named_queue

    def test_get_stats_multiple_on_named_queue(self):
        ep1 = ListeningBaseEndpoint(node=self.container.node, from_name="test_get_stats_multi")
        gl1 = spawn(ep1.listen)

        ep2 = ListeningBaseEndpoint(node=self.container.node, from_name="test_get_stats_multi")
        gl2 = spawn(ep2.listen)

        ep1.get_ready_event().wait(timeout=5)
        ep2.get_ready_event().wait(timeout=5)

        gs_res1 = ep1.get_stats()
        self.assertEquals(gs_res1, (0, 2)) # num of messages, num listeners

        gs_res2 = ep2.get_stats()
        self.assertEquals(gs_res2, (0, 2)) # num of messages, num listeners

        ep1.close()
        ep2.close()
        gl1.join(timeout=5)
        gl2.join(timeout=5)
开发者ID:daf,项目名称:pyon,代码行数:20,代码来源:test_endpoint.py


示例8: test_close

    def test_close(self):
        ep = ListeningBaseEndpoint(node=Mock(soec=NodeB))
        ep._chan = Mock()
        ep.close()

        ep._chan.close.assert_called_once_with()
开发者ID:daf,项目名称:pyon,代码行数:6,代码来源:test_endpoint.py


示例9: test_get_ready_event

 def test_get_ready_event(self):
     ep = ListeningBaseEndpoint(node=Mock(spec=NodeB))
     self.assertEquals(ep.get_ready_event(), ep._ready_event)
开发者ID:daf,项目名称:pyon,代码行数:3,代码来源:test_endpoint.py



注:本文中的pyon.net.endpoint.ListeningBaseEndpoint类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。


鲜花

握手

雷人

路过

鸡蛋
该文章已有0人参与评论

请发表评论

全部评论

专题导读
上一篇:
Python endpoint.Publisher类代码示例发布时间:2022-05-27
下一篇:
Python endpoint.EndpointUnit类代码示例发布时间:2022-05-27
热门推荐
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

在线客服(服务时间 9:00~18:00)

在线QQ客服
地址:深圳市南山区西丽大学城创智工业园
电邮:jeky_zhao#qq.com
移动电话:139-2527-9053

Powered by 互联科技 X3.4© 2001-2213 极客世界.|Sitemap