本文整理汇总了Python中mosq_test.packet_matches函数的典型用法代码示例。如果您正苦于以下问题：Python packet_matches函数的具体用法？Python packet_matches怎么用？Python packet_matches使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了packet_matches函数的15个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: dict
client_args = sys.argv[1:]
env = dict(os.environ)
env['LD_LIBRARY_PATH'] = '../../lib:../../lib/cpp'
try:
pp = env['PYTHONPATH']
except KeyError:
pp = ''
env['PYTHONPATH'] = '../../lib/python:'+pp
client = subprocess.Popen(client_args, env=env)
try:
(conn, address) = sock.accept()
conn.settimeout(10)
connect_recvd = conn.recv(len(connect_packet))
if mosq_test.packet_matches("connect", connect_recvd, connect_packet):
conn.send(connack_packet)
subscribe_recvd = conn.recv(len(subscribe_packet))
if mosq_test.packet_matches("subscribe", subscribe_recvd, subscribe_packet):
conn.send(suback_packet)
disconnect_recvd = conn.recv(len(disconnect_packet))
if mosq_test.packet_matches("disconnect", disconnect_recvd, disconnect_packet):
rc = 0
conn.close()
finally:
client.terminate()
client.wait()
sock.close()
开发者ID:Marina1704，项目名称:FotbalRobotic，代码行数:31，代码来源:02-subscribe-qos2.py
示例2: print
rc = 1
keepalive = 10
connect_packet = mosq_test.gen_connect("connect-uname-pwd-test", keepalive=keepalive, username="test-username", password="cnwTICONIURW")
connack_packet = mosq_test.gen_connack(rc=0)
broker = subprocess.Popen(['../../src/mosquitto', '-c', '09-plugin-auth-unpwd-success.conf'], stderr=subprocess.PIPE)
try:
time.sleep(0.5)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(10)
sock.connect(("localhost", 1888))
sock.send(connect_packet)
connack_recvd = sock.recv(len(connack_packet))
sock.close()
if mosq_test.packet_matches("connack", connack_recvd, connack_packet):
rc = 0
finally:
broker.terminate()
broker.wait()
if rc:
(stdo, stde) = broker.communicate()
print(stde)
exit(rc)
开发者ID:Marina1704，项目名称:FotbalRobotic，代码行数:28，代码来源:09-plugin-auth-unpwd-success.py
示例3:
suback_packet = mosq_test.gen_suback(mid, 0)
publish_packet = mosq_test.gen_publish("will/qos0/test", qos=0, payload=struct.pack("!4sB7s", "will", 0, "message"))
broker = subprocess.Popen(['../../src/mosquitto', '-p', '1888'], stderr=subprocess.PIPE)
try:
time.sleep(0.5)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(30)
sock.connect(("localhost", 1888))
sock.send(connect_packet)
connack_recvd = sock.recv(len(connack_packet))
if mosq_test.packet_matches("connack", connack_recvd, connack_packet):
sock.send(subscribe_packet)
suback_recvd = sock.recv(len(suback_packet))
if mosq_test.packet_matches("suback", suback_recvd, suback_packet):
will = subprocess.Popen(['./07-will-null-helper.py'])
will.wait()
publish_recvd = sock.recv(len(publish_packet))
if mosq_test.packet_matches("publish", publish_recvd, publish_packet):
rc = 0
sock.close()
finally:
broker.terminate()
开发者ID:Marina1704，项目名称:FotbalRobotic，代码行数:31，代码来源:07-will-null.py
示例4: dict
client_args = sys.argv[1:]
env = dict(os.environ)
env['LD_LIBRARY_PATH'] = '../../lib:../../lib/cpp'
try:
pp = env['PYTHONPATH']
except KeyError:
pp = ''
env['PYTHONPATH'] = '../../lib/python:'+pp
client = subprocess.Popen(client_args, env=env)
try:
(conn, address) = sock.accept()
conn.settimeout(keepalive+10)
connect_recvd = conn.recv(len(connect_packet))
if mosq_test.packet_matches("connect", connect_recvd, connect_packet):
conn.send(connack_packet)
pingreq_recvd = conn.recv(len(pingreq_packet))
if mosq_test.packet_matches("pingreq", pingreq_recvd, pingreq_packet):
rc = 0
conn.close()
finally:
client.terminate()
client.wait()
sock.close()
exit(rc)
开发者ID:Marina1704，项目名称:FotbalRobotic，代码行数:29，代码来源:01-keepalive-pingreq.py
示例5:
ssock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
ssock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
ssock.settimeout(40)
ssock.bind(('', 1888))
ssock.listen(5)
broker = subprocess.Popen(['../../src/mosquitto', '-c', '06-bridge-b2br-disconnect-qos2.conf'], stderr=subprocess.PIPE)
try:
time.sleep(0.5)
(bridge, address) = ssock.accept()
bridge.settimeout(10)
connect_recvd = bridge.recv(len(connect_packet))
if mosq_test.packet_matches("connect", connect_recvd, connect_packet):
bridge.send(connack_packet)
subscribe_recvd = bridge.recv(len(subscribe_packet))
if mosq_test.packet_matches("subscribe", subscribe_recvd, subscribe_packet):
bridge.send(suback_packet)
bridge.send(publish_packet)
bridge.close()
(bridge, address) = ssock.accept()
bridge.settimeout(10)
connect_recvd = bridge.recv(len(connect_packet))
if mosq_test.packet_matches("connect", connect_recvd, connect_packet):
bridge.send(connack_packet)
subscribe_recvd = bridge.recv(len(subscribe2_packet))
开发者ID:Marina1704，项目名称:FotbalRobotic，代码行数:31，代码来源:06-bridge-b2br-disconnect-qos2.py
示例6: dict
client_args = sys.argv[1:]
env = dict(os.environ)
env["LD_LIBRARY_PATH"] = "../../lib:../../lib/cpp"
try:
pp = env["PYTHONPATH"]
except KeyError:
pp = ""
env["PYTHONPATH"] = "../../lib/python:" + pp
client = subprocess.Popen(client_args, env=env)
try:
(conn, address) = sock.accept()
conn.settimeout(10)
connect_recvd = conn.recv(len(connect_packet))
if mosq_test.packet_matches("connect", connect_recvd, connect_packet):
conn.send(connack_packet)
publish_recvd = conn.recv(len(publish_packet))
if mosq_test.packet_matches("publish", publish_recvd, publish_packet):
disconnect_recvd = conn.recv(len(disconnect_packet))
if mosq_test.packet_matches("disconnect", disconnect_recvd, disconnect_packet):
rc = 0
conn.close()
finally:
client.terminate()
client.wait()
sock.close()
开发者ID:karthikbnd，项目名称:BBB-codes，代码行数:30，代码来源:03-publish-qos0.py
示例7:
keepalive = 60
connect_packet = mosq_test.gen_connect("test-helper", keepalive=keepalive)
connack_packet = mosq_test.gen_connack(rc=0)
mid = 312
publish_packet = mosq_test.gen_publish("qos2/timeout/test", qos=2, mid=mid, payload="timeout-message")
pubrec_packet = mosq_test.gen_pubrec(mid)
pubrel_packet = mosq_test.gen_pubrel(mid)
pubcomp_packet = mosq_test.gen_pubcomp(mid)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect(("localhost", 1888))
sock.send(connect_packet)
connack_recvd = sock.recv(len(connack_packet))
if mosq_test.packet_matches("helper connack", connack_recvd, connack_packet):
sock.send(publish_packet)
pubrec_recvd = sock.recv(len(pubrec_packet))
if mosq_test.packet_matches("helper pubrec", pubrec_recvd, pubrec_packet):
sock.send(pubrel_packet)
pubcomp_recvd = sock.recv(len(pubcomp_packet))
if mosq_test.packet_matches("helper pubcomp", pubcomp_recvd, pubcomp_packet):
rc = 0
sock.close()
exit(rc)
开发者ID:Marina1704，项目名称:FotbalRobotic，代码行数:29，代码来源:03-publish-b2c-timeout-qos2-helper.py
示例8:
if cmd_subfolder not in sys.path:
sys.path.insert(0, cmd_subfolder)
import mosq_test
rc = 1
keepalive = 60
connect_packet = mosq_test.gen_connect("test-helper", keepalive=keepalive)
connack_packet = mosq_test.gen_connack(rc=0)
mid = 128
publish_packet = mosq_test.gen_publish("bridge/disconnect/test", qos=1, mid=mid, payload="disconnect-message")
puback_packet = mosq_test.gen_puback(mid)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect(("localhost", 1889))
sock.send(connect_packet)
connack_recvd = sock.recv(len(connack_packet))
if mosq_test.packet_matches("helper connack", connack_recvd, connack_packet):
sock.send(publish_packet)
puback_recvd = sock.recv(len(puback_packet))
if mosq_test.packet_matches("helper puback", puback_recvd, puback_packet):
rc = 0
sock.close()
exit(rc)
开发者ID:Marina1704，项目名称:FotbalRobotic，代码行数:29，代码来源:06-bridge-br2b-disconnect-qos1-helper.py
示例9: dict
sock.settimeout(10)
sock.bind(('', 1888))
sock.listen(5)
client_args = sys.argv[1:]
env = dict(os.environ)
env['LD_LIBRARY_PATH'] = '../../lib:../../lib/cpp'
try:
pp = env['PYTHONPATH']
except KeyError:
pp = ''
env['PYTHONPATH'] = '../../lib/python:'+pp
client = subprocess.Popen(client_args, env=env)
try:
(conn, address) = sock.accept()
conn.settimeout(10)
connect_recvd = conn.recv(len(connect_packet))
if mosq_test.packet_matches("connect", connect_recvd, connect_packet):
rc = 0
conn.close()
finally:
client.terminate()
client.wait()
sock.close()
exit(rc)
开发者ID:Marina1704，项目名称:FotbalRobotic，代码行数:29，代码来源:01-unpwd-set.py
示例10:
ssock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
ssock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
ssock.settimeout(40)
ssock.bind(('', 1888))
ssock.listen(5)
broker = subprocess.Popen(['../../src/mosquitto', '-c', '06-bridge-b2br-disconnect-qos1.conf'], stderr=subprocess.PIPE)
try:
time.sleep(0.5)
(bridge, address) = ssock.accept()
bridge.settimeout(10)
connect_recvd = bridge.recv(len(connect_packet))
if mosq_test.packet_matches("connect", connect_recvd, connect_packet):
bridge.send(connack_packet)
subscribe_recvd = bridge.recv(len(subscribe_packet))
if mosq_test.packet_matches("subscribe", subscribe_recvd, subscribe_packet):
bridge.send(suback_packet)
bridge.send(publish_packet)
# Bridge doesn't have time to respond but should expect us to retry
# and so remove PUBACK.
bridge.close()
(bridge, address) = ssock.accept()
bridge.settimeout(10)
connect_recvd = bridge.recv(len(connect_packet))
if mosq_test.packet_matches("connect", connect_recvd, connect_packet):
开发者ID:Marina1704，项目名称:FotbalRobotic，代码行数:31，代码来源:06-bridge-b2br-disconnect-qos1.py
示例11: exit
ssock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
ssock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
ssock.settimeout(40)
ssock.bind(('', 1888))
ssock.listen(5)
broker = subprocess.Popen(['../../src/mosquitto', '-c', '06-bridge-br2b-disconnect-qos2.conf'], stderr=subprocess.PIPE)
try:
time.sleep(0.5)
(bridge, address) = ssock.accept()
bridge.settimeout(10)
connect_recvd = bridge.recv(len(connect_packet))
if mosq_test.packet_matches("connect", connect_recvd, connect_packet):
bridge.send(connack_packet)
subscribe_recvd = bridge.recv(len(subscribe_packet))
if mosq_test.packet_matches("subscribe", subscribe_recvd, subscribe_packet):
bridge.send(suback_packet)
pub = subprocess.Popen(['./06-bridge-br2b-disconnect-qos2-helper.py'])
if pub.wait():
exit(1)
publish_recvd = bridge.recv(len(publish_packet))
if mosq_test.packet_matches("publish", publish_recvd, publish_packet):
bridge.close()
(bridge, address) = ssock.accept()
bridge.settimeout(10)
开发者ID:Marina1704，项目名称:FotbalRobotic，代码行数:31，代码来源:06-bridge-br2b-disconnect-qos2.py
示例12: dict
client_args = sys.argv[1:]
env = dict(os.environ)
env['LD_LIBRARY_PATH'] = '../../lib:../../lib/cpp'
try:
pp = env['PYTHONPATH']
except KeyError:
pp = ''
env['PYTHONPATH'] = '../../lib/python:'+pp
client = subprocess.Popen(client_args, env=env)
try:
(conn, address) = sock.accept()
conn.settimeout(10)
connect_recvd = conn.recv(len(connect_packet))
if mosq_test.packet_matches("connect", connect_recvd, connect_packet):
conn.send(connack_packet)
conn.send(publish_packet)
pubrec_recvd = conn.recv(len(pubrec_packet))
if mosq_test.packet_matches("pubrec", pubrec_recvd, pubrec_packet):
# Should be repeated due to timeout
pubrec_recvd = conn.recv(len(pubrec_packet))
if mosq_test.packet_matches("pubrec", pubrec_recvd, pubrec_packet):
conn.send(pubrel_packet)
pubcomp_recvd = conn.recv(len(pubcomp_packet))
if mosq_test.packet_matches("pubcomp", pubcomp_recvd, pubcomp_packet):
rc = 0
conn.close()
finally:
开发者ID:Marina1704，项目名称:FotbalRobotic，代码行数:31，代码来源:03-publish-b2c-qos2.py
示例13: pattern_test
def pattern_test(sub_topic, pub_topic):
rc = 1
keepalive = 60
connect_packet = mosq_test.gen_connect("pattern-sub-test", keepalive=keepalive)
connack_packet = mosq_test.gen_connack(rc=0)
publish_packet = mosq_test.gen_publish(pub_topic, qos=0, payload="message")
publish_retained_packet = mosq_test.gen_publish(pub_topic, qos=0, retain=True, payload="message")
mid = 312
subscribe_packet = mosq_test.gen_subscribe(mid, sub_topic, 0)
suback_packet = mosq_test.gen_suback(mid, 0)
mid = 234;
unsubscribe_packet = mosq_test.gen_unsubscribe(mid, sub_topic)
unsuback_packet = mosq_test.gen_unsuback(mid)
broker = subprocess.Popen(['../../src/mosquitto', '-p', '1888'], stderr=subprocess.PIPE)
try:
time.sleep(0.5)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect(("localhost", 1888))
sock.settimeout(5)
sock.send(connect_packet)
connack_recvd = sock.recv(len(connack_packet))
if mosq_test.packet_matches("connack", connack_recvd, connack_packet):
sock.send(subscribe_packet)
suback_recvd = sock.recv(len(suback_packet))
if mosq_test.packet_matches("suback", suback_recvd, suback_packet):
pub = subprocess.Popen(['./03-pattern-matching-helper.py', pub_topic])
pub.wait()
publish_recvd = sock.recv(len(publish_packet))
if mosq_test.packet_matches("publish", publish_recvd, publish_packet):
sock.send(unsubscribe_packet)
unsuback_recvd = sock.recv(len(unsuback_packet))
if mosq_test.packet_matches("unsuback", unsuback_recvd, unsuback_packet):
sock.send(subscribe_packet)
suback_recvd = sock.recv(len(suback_packet))
if mosq_test.packet_matches("suback", suback_recvd, suback_packet):
publish_retained_recvd = sock.recv(len(publish_retained_packet))
if mosq_test.packet_matches("publish retained", publish_retained_recvd, publish_retained_packet):
rc = 0
sock.close()
finally:
broker.terminate()
broker.wait()
if rc:
(stdo, stde) = broker.communicate()
print(stde)
return rc
开发者ID:Marina1704，项目名称:FotbalRobotic，代码行数:61，代码来源:03-pattern-matching.py
示例14: dict
client_args = sys.argv[1:]
env = dict(os.environ)
env['LD_LIBRARY_PATH'] = '../../lib:../../lib/cpp'
try:
pp = env['PYTHONPATH']
except KeyError:
pp = ''
env['PYTHONPATH'] = '../../lib/python:'+pp
client = subprocess.Popen(client_args, env=env)
try:
(conn, address) = sock.accept()
conn.settimeout(10)
connect_recvd = conn.recv(len(connect_packet))
if mosq_test.packet_matches("connect", connect_recvd, connect_packet):
conn.send(connack_packet)
conn.send(publish_packet)
puback_recvd = conn.recv(len(puback_packet))
if mosq_test.packet_matches("puback", puback_recvd, puback_packet):
rc = 0
conn.close()
finally:
for i in range(0, 5):
if client.returncode != None:
break
time.sleep(0.1)
client.terminate()
开发者ID:Marina1704，项目名称:FotbalRobotic，代码行数:31，代码来源:03-publish-b2c-qos1.py
示例15:
pubrec_packet2 = mosq_test.gen_pubrec(mid)
pubrel_packet2 = mosq_test.gen_pubrel(mid)
pubcomp_packet2 = mosq_test.gen_pubcomp(mid)
broker = subprocess.Popen(["../../src/mosquitto", "-p", "1888"], stderr=subprocess.PIPE)
try:
time.sleep(0.5)
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(10)
sock.connect(("localhost", 1888))
sock.send(connect_packet)
connack_recvd = sock.recv(len(connack_packet))
if mosq_test.packet_matches("connack", connack_recvd, connack_packet):
sock.send(subscribe_packet)
suback_recvd = sock.recv(len(suback_packet))
if mosq_test.packet_matches("suback", suback_recvd, suback_packet):
sock.send(publish_packet)
pubrec_recvd = sock.recv(len(pubrec_packet))
if mosq_test.packet_matches("pubrec", pubrec_recvd, pubrec_packet):
sock.send(pubrel_packet)
pubcomp_recvd = sock.recv(len(pubcomp_packet))
if mosq_test.packet_matches("pubcomp", pubcomp_recvd, pubcomp_packet):
publish_recvd = sock.recv(len(publish_packet2))
if mosq_test.packet_matches("publish2", publish_recvd, publish_packet2):
开发者ID:karthikbnd，项目名称:BBB-codes，代码行数:31，代码来源:02-subpub-qos2.py
注：本文中的mosq_test.packet_matches函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论