Python queue.Queue类代码示例

OGeek|极客世界-中国程序员成长平台 › 门户 › 编程› Python›Python编程经验

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

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

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

示例1: test_load_all

 def test_load_all(self):
     self.store.load().AndReturn([(3, 'one'), (5, 'two'), (1, 'three')])
     self.mox.ReplayAll()
     queue = Queue(self.store, self.relay)
     queue._load_all()
     self.assertEqual([(1, 'three'), (3, 'one'), (5, 'two')], queue.queued)
     self.assertTrue(queue.wake.isSet())

开发者ID:RoboticCheese，项目名称:python-slimta，代码行数:7，代码来源:test_slimta_queue.py

示例2: test_check_ready_missing

 def test_check_ready_missing(self):
     self.store.get('1234').AndRaise(KeyError)
     self.mox.ReplayAll()
     queue = Queue(self.store, self.relay, store_pool=Pool(5))
     queue._add_queued((10, '1234'))
     queue._check_ready(20)
     queue.store_pool.join()

开发者ID:you4you4，项目名称:python-slimta，代码行数:7，代码来源:test_slimta_queue.py

示例3: test_load_all_empty

 def test_load_all_empty(self):
     self.store.load().AndReturn([])
     self.mox.ReplayAll()
     queue = Queue(self.store, self.relay)
     queue._load_all()
     self.assertEqual([], queue.queued)
     self.assertFalse(queue.wake.isSet())

开发者ID:RoboticCheese，项目名称:python-slimta，代码行数:7，代码来源:test_slimta_queue.py

示例4: test_enqueue_wait

 def test_enqueue_wait(self):
     self.store.write(self.env, IsA(float)).AndReturn('1234')
     self.relay._attempt(self.env, 0)
     self.store.remove('1234')
     self.mox.ReplayAll()
     queue = Queue(self.store, self.relay, relay_pool=5)
     self.assertEqual([(self.env, '1234')], queue.enqueue(self.env))
     queue.relay_pool.join()

开发者ID:RoboticCheese，项目名称:python-slimta，代码行数:8，代码来源:test_slimta_queue.py

示例5: test_check_ready

 def test_check_ready(self):
     self.store.get('1234').AndReturn((self.env, 0))
     self.relay._attempt(self.env, 0)
     self.store.remove('1234')
     self.mox.ReplayAll()
     queue = Queue(self.store, self.relay, store_pool=Pool(5))
     queue._add_queued((10, '1234'))
     queue._check_ready(20)
     queue.store_pool.join()

开发者ID:RoboticCheese，项目名称:python-slimta，代码行数:9，代码来源:test_slimta_queue.py

示例6: test_wait_ready_nonequeued

 def test_wait_ready_nonequeued(self):
     queue = Queue(self.store, self.relay)
     def wait_func():
         queue._wait_ready(20)
     thread = gevent.spawn(wait_func)
     gevent.sleep(0)
     self.assertFalse(thread.ready())
     queue._add_queued((10, '1234'))
     gevent.sleep(0)
     self.assertTrue(thread.ready())

开发者ID:RoboticCheese，项目名称:python-slimta，代码行数:10，代码来源:test_slimta_queue.py

示例7: test_enqueue_wait_permanentfail

 def test_enqueue_wait_permanentfail(self):
     self.store.write(self.env, IsA(float)).AndReturn('1234')
     self.relay._attempt(self.env, 0).AndRaise(PermanentRelayError('permanent', Reply('550', 'permanent')))
     self.store.remove('1234')
     self.mox.ReplayAll()
     def no_bounce(envelope, reply):
         return None
     queue = Queue(self.store, self.relay, bounce_factory=no_bounce, relay_pool=5)
     queue.enqueue(self.env)
     queue.relay_pool.join()

开发者ID:RoboticCheese，项目名称:python-slimta，代码行数:10，代码来源:test_slimta_queue.py

示例8: test_enqueue_wait_unhandledfail

 def test_enqueue_wait_unhandledfail(self):
     self.store.write(self.env, IsA(float)).AndReturn('1234')
     self.relay._attempt(self.env, 0).AndRaise(Exception('unhandled error'))
     self.store.increment_attempts('1234')
     self.store.set_timestamp('1234', IsA(float))
     self.mox.ReplayAll()
     def backoff(envelope, attempts):
         return 0
     queue = Queue(self.store, self.relay, backoff=backoff, relay_pool=5)
     queue.enqueue(self.env)
     queue.relay_pool.join()

开发者ID:rafaelnovello，项目名称:python-slimta，代码行数:11，代码来源:test_slimta_queue.py

示例9: test_enqueue_wait_transientfail

 def test_enqueue_wait_transientfail(self):
     self.store.write(self.env, IsA(float)).AndReturn('1234')
     self.relay._attempt(self.env, 0).AndRaise(TransientRelayError('transient', Reply('450', 'transient')))
     self.store.increment_attempts('1234')
     self.store.set_timestamp('1234', IsA(float))
     self.mox.ReplayAll()
     def backoff(envelope, attempts):
         return 0
     queue = Queue(self.store, self.relay, backoff=backoff, relay_pool=5)
     queue.enqueue(self.env)
     queue.relay_pool.join()

开发者ID:RoboticCheese，项目名称:python-slimta，代码行数:11，代码来源:test_slimta_queue.py

示例10: test_wait_store

 def test_wait_store(self):
     queue = Queue(self.store, self.relay, relay_pool=5)
     queue.wake = self.mox.CreateMock(AsyncResult)
     self.store.wait().AndReturn((1234567890, '1234'))
     queue.wake.set()
     self.store.wait().AndReturn(None)
     self.store.wait().AndReturn((2345678901, '5678'))
     queue.wake.set()
     self.store.wait().AndRaise(NotImplementedError)
     self.mox.ReplayAll()
     queue._wait_store()

开发者ID:pombredanne，项目名称:python-slimta，代码行数:11，代码来源:test_slimta_queue.py

示例11: test_policies

 def test_policies(self):
     p1 = self.mox.CreateMock(QueuePolicy)
     p2 = self.mox.CreateMock(QueuePolicy)
     p1.apply(self.env)
     p2.apply(self.env)
     self.mox.ReplayAll()
     queue = Queue(self.store, self.relay)
     queue.add_policy(p1)
     queue.add_policy(p2)
     self.assertRaises(TypeError, queue.add_policy, None)
     queue._run_policies(self.env)

开发者ID:RoboticCheese，项目名称:python-slimta，代码行数:11，代码来源:test_slimta_queue.py

示例12: test_add_queued

 def test_add_queued(self):
     queue = Queue(self.store, self.relay)
     queue._add_queued((10, 'one'))
     queue._add_queued((5, 'two'))
     queue._add_queued((99, 'one'))
     queue._add_queued((7, 'three'))
     self.assertEqual([(5, 'two'), (7, 'three'), (10, 'one')], queue.queued)
     self.assertTrue(queue.wake.isSet())

开发者ID:pombredanne，项目名称:python-slimta，代码行数:8，代码来源:test_slimta_queue.py

示例13: test_wait_ready_noneready

 def test_wait_ready_noneready(self):
     queue = Queue(self.store, self.relay)
     queue._add_queued((20, '1234'))
     queue.wake.clear()
     def wait_func():
         queue._wait_ready(10)
     thread = gevent.spawn(wait_func)
     gevent.sleep(0)
     self.assertFalse(thread.ready())
     queue._add_queued((5, '5678'))
     gevent.sleep(0)
     self.assertTrue(thread.ready())

开发者ID:pombredanne，项目名称:python-slimta，代码行数:12，代码来源:test_slimta_queue.py

示例14: test_enqueue_wait_splitpolicy

 def test_enqueue_wait_splitpolicy(self):
     splitpolicy1 = self.mox.CreateMock(QueuePolicy)
     splitpolicy2 = self.mox.CreateMock(QueuePolicy)
     regpolicy = self.mox.CreateMock(QueuePolicy)
     env1 = Envelope('[email protected]', ['[email protected]'])
     env2 = Envelope('[email protected]', ['[email protected]'])
     env3 = Envelope('[email protected]', ['[email protected]'])
     splitpolicy1.apply(self.env).AndReturn([env1, env2])
     regpolicy.apply(env1)
     splitpolicy2.apply(env1)
     regpolicy.apply(env2)
     splitpolicy2.apply(env2).AndReturn([env2, env3])
     self.store.write(env1, IsA(float)).AndReturn('1234')
     self.store.write(env2, IsA(float)).AndReturn('5678')
     self.store.write(env3, IsA(float)).AndReturn('90AB')
     self.relay._attempt(env1, 0).InAnyOrder('relay')
     self.relay._attempt(env2, 0).InAnyOrder('relay')
     self.relay._attempt(env3, 0).InAnyOrder('relay')
     self.store.remove('1234').InAnyOrder('relay')
     self.store.remove('5678').InAnyOrder('relay')
     self.store.remove('90AB').InAnyOrder('relay')
     self.mox.ReplayAll()
     queue = Queue(self.store, self.relay, relay_pool=5)
     queue.add_policy(splitpolicy1)
     queue.add_policy(regpolicy)
     queue.add_policy(splitpolicy2)
     self.assertEqual([(env1, '1234'), (env2, '5678'), (env3, '90AB')],
                      queue.enqueue(self.env))
     queue.relay_pool.join()

开发者ID:RoboticCheese，项目名称:python-slimta，代码行数:29，代码来源:test_slimta_queue.py

示例15: test_enqueue_wait_partial_relay_expired

 def test_enqueue_wait_partial_relay_expired(self):
     env = Envelope('[email protected]', ['[email protected]',
                                           '[email protected]',
                                           '[email protected]'])
     bounce_mock = self.mox.CreateMockAnything()
     bounce_mock(IsA(Envelope), IsA(Reply)).AndReturn(None)
     bounce_mock(IsA(Envelope), IsA(Reply)).AndReturn(None)
     self.store.write(env, IsA(float)).AndReturn('1234')
     self.relay._attempt(env, 0).AndReturn([TransientRelayError('transient', Reply('450', 'transient 1')),
                                            TransientRelayError('transient', Reply('450', 'transient 1')),
                                            TransientRelayError('transient', Reply('450', 'transient 2'))])
     self.store.increment_attempts('1234')
     self.store.remove('1234')
     self.mox.ReplayAll()
     queue = Queue(self.store, self.relay, bounce_factory=bounce_mock, relay_pool=5)
     queue.enqueue(env)
     queue.relay_pool.join()

开发者ID:sk1p，项目名称:python-slimta，代码行数:17，代码来源:test_slimta_queue.py

示例16: test_enqueue_wait_partial_relay

 def test_enqueue_wait_partial_relay(self):
     env = Envelope('[email protected]', ['[email protected]',
                                           '[email protected]',
                                           '[email protected]'])
     self.store.write(env, IsA(float)).AndReturn('1234')
     self.relay._attempt(env, 0).AndReturn([None,
                                            TransientRelayError('transient', Reply('450', 'transient')),
                                            PermanentRelayError('permanent', Reply('550', 'permanent'))])
     self.store.increment_attempts('1234')
     self.store.set_timestamp('1234', IsA(float))
     self.store.set_recipients_delivered('1234', [0, 2])
     self.mox.ReplayAll()
     def backoff(envelope, attempts):
         return 0
     def no_bounce(envelope, reply):
         return None
     queue = Queue(self.store, self.relay, backoff=backoff, bounce_factory=no_bounce, relay_pool=5)
     queue.enqueue(env)
     queue.relay_pool.join()

开发者ID:sk1p，项目名称:python-slimta，代码行数:19，代码来源:test_slimta_queue.py

示例17: test_flush

 def test_flush(self):
     self.store.get('three').AndReturn((self.env, 1))
     self.store.get('two').AndReturn((self.env, 2))
     self.store.get('one').AndReturn((self.env, 3))
     self.relay._attempt(self.env, 1)
     self.store.remove('three')
     self.relay._attempt(self.env, 2)
     self.store.remove('two')
     self.relay._attempt(self.env, 3)
     self.store.remove('one')
     self.mox.ReplayAll()
     queue = Queue(self.store, self.relay, store_pool=5, relay_pool=5)
     queue._add_queued((float('inf'), 'one'))
     queue._add_queued((0, 'two'))
     queue._add_queued((float('-inf'), 'three'))
     queue.flush()
     queue.store_pool.join()
     queue.relay_pool.join()

开发者ID:RoboticCheese，项目名称:python-slimta，代码行数:18，代码来源:test_slimta_queue.py

示例18: test_kill

 def test_kill(self):
     self.mox.ReplayAll()
     queue = Queue(self.store, self.relay)
     self.assertFalse(queue.ready())
     queue.kill()
     self.assertTrue(queue.ready())

开发者ID:you4you4，项目名称:python-slimta，代码行数:6，代码来源:test_slimta_queue.py

示例19: test_wait_ready_nowait

 def test_wait_ready_nowait(self):
     queue = Queue(self.store, self.relay)
     queue._add_queued((10, '1234'))
     with gevent.Timeout(1.0):
         queue._wait_ready(20)

开发者ID:RoboticCheese，项目名称:python-slimta，代码行数:5，代码来源:test_slimta_queue.py

示例20: MxSmtpRelay

from slimta.edge.smtp import SmtpEdge
from slimta.queue import Queue
from slimta.queue.dict import DictStorage
from slimta.relay.smtp.mx import MxSmtpRelay
from slimta.policy.headers import *
from slimta.policy.split import RecipientDomainSplit

logging.basicConfig(level=logging.DEBUG)

relay = MxSmtpRelay()

env_db = shelve.open('envelope.db')
meta_db = shelve.open('meta.db')
queue_storage = DictStorage(env_db, meta_db)
queue = Queue(queue_storage, relay)

queue.add_prequeue_policy(AddDateHeader())
queue.add_prequeue_policy(AddMessageIdHeader())
queue.add_prequeue_policy(RecipientDomainSplit())
queue.add_prequeue_policy(AddReceivedHeader())

edge = SmtpEdge(('127.0.0.1', 1337), queue)
edge.start()
queue.start()
try:
    edge.get()
except KeyboardInterrupt:
    print
finally:
    for key in env_db.keys():

开发者ID:RoboticCheese，项目名称:python-slimta，代码行数:30，代码来源:smtpmx.py

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

鲜花

握手

雷人

路过

鸡蛋

该文章已有0人参与评论

请发表评论

全部评论

专题导读

More+

10-27 六六分期app的软件客服如何联系？(六六分期

11-06 可心卡盟:win10系统火狐flash插件崩溃怎么

11-06 亲亲特价:怎么删除回收站图标

11-06 济南大学虚拟社区:鲁大师节能降温的具体办

11-06 xlueops.exe:无线网络安装向导

11-06 女斗合众国:win7系统cf与主机连接不稳定怎

11-06 0xc000022-[cf烟雾头]cf怎么调烟雾头

11-06 qizideyouhuo:应用程序无法正常启动0xc0000

11-06 ipz-185:win7系统vcf文件怎么打开

11-06 傻哥蹦迪:win10系统s4怎么打开usb调试

11-06 八神浩树gtaste:回收站清空了怎么恢复

11-06 妖尾之黑色守护:win10系统电脑没有1440x900

11-06 校园至尊魔王小说:win7系统浏览网页时字体

11-06 女斗合众国:win10系统访问共享文件夹提示请

11-06 tokyo hot n0654:恢复win7系统默认字体一招

11-06 雨酷仙境:设置win7系统转移临时文件夹腾出

11-06 阿穆纳伊之杖:win7系统开始菜单在右边还原

11-06 tunespotting:win10系统火狐flash插件总是

11-06 甘尔葛分析师：计谋网站seo关键词暴涨有什

11-06 蔡贵霖: 计谋网站seo关键词暴涨有什么秘密

11-06 博益网首页:ao3网页版进入不了解决方法

11-06 漏斗子专栏: 网站数据分析小白易懂精华篇

11-06 见证双虹怎么做:win7系统开启telnet命令的

11-06 颾狐蝶蜋:系统资源不足无法完成请求的服务

11-06 国光中学校歌:提交网站到alexa查询详细步骤

11-06 西安有情天:静态网页和动态网页的区别

11-06 红木雅尚斋:外部链接构造对网站的好处

11-06 前官礼遇：防止域名劫持–增强域安全性的10

11-06 密传二转答案: 中文分词算法有哪些

11-06 金泉家园邮编:百度快照劫持的表现及应对方

Python client.SmtpRelayClient类代码示例发布时间：2022-05-27

Python envelope.Envelope类代码示例发布时间：2022-05-27

Python util.grid_equal函数代码示例

1 Python 入门教程

Python入门教程 Python 是一种解释型、面向对象、动态数据类型的高级程序设计语言。 P

阅读：13807|2022-01-22

2 Python wikiutil.getFrontPage函数代码示例

Python wikiutil.getFrontPage函数代码示例

阅读：10195|2022-05-24

3 Python 简介

Python 简介 Python 是一个高层次的结合了解释性、编译性、互动性和面向对象的脚本

阅读：4091|2022-01-22

4 Python tests.group函数代码示例

Python tests.group函数代码示例

阅读：4043|2022-05-27

5 Python util.check_if_user_has_permission

Python util.check_if_user_has_permission函数代码示例

阅读：3845|2022-05-27

6 Python 操练实例98

Python 练习实例98 Python 100例题目：从键盘输入一个字符串，将小写字母全部转换成大

阅读：3514|2022-01-22

7 Python 环境搭建

Python 环境搭建本章节我们将向大家介绍如何在本地搭建 Python 开发环境。 Py

阅读：3031|2022-01-22

8 Python output.darkgreen函数代码示例

Python output.darkgreen函数代码示例

阅读：2653|2022-05-25

9 Python 基础语法

Python 基础语法 Python 语言与 Perl，C 和 Java 等语言有许多相似之处。但是，也

阅读：2649|2022-01-22

10 Python 中文编码

Python 中文编码前面章节中我们已经学会了如何用 Python 输出 Hello, World!，英文没

阅读：2302|2022-01-22

客服电话

电子邮件

Python queue.Queue类代码示例

示例1: test_load_all

示例2: test_check_ready_missing

示例3: test_load_all_empty

示例4: test_enqueue_wait

示例5: test_check_ready

示例6: test_wait_ready_nonequeued

示例7: test_enqueue_wait_permanentfail

示例8: test_enqueue_wait_unhandledfail

示例9: test_enqueue_wait_transientfail

示例10: test_wait_store

示例11: test_policies

示例12: test_add_queued

示例13: test_wait_ready_noneready

示例14: test_enqueue_wait_splitpolicy

示例15: test_enqueue_wait_partial_relay_expired

示例16: test_enqueue_wait_partial_relay

示例17: test_flush

示例18: test_kill

示例19: test_wait_ready_nowait

示例20: MxSmtpRelay

请发表评论

全部评论

上一篇：

下一篇：

Python util.grid_equal函数代码示例

Python util.get_worker_name函数代码示例

Python util.get_webmention_target函数代

Python util.get_uuid函数代码示例

Python util.get_type_by_name函数代码示例

Python util.grid_equal函数代码示例

Python util.get_worker_name函数代码示例

Python util.get_webmention_target函数代

Python util.get_uuid函数代码示例

Python util.get_type_by_name函数代码示例

Python util.get_stdout函数代码示例

关于我们

产品与服务

解决方案

139-2527-9053