nessar/projet-reseau-mqtt
Archived
1
0 Fork 0
Code Issues Packages Projects Releases Wiki Activity

finished

Browse Source
This commit is contained in:
nessar committed 2021-12-18 17:13:58 +01:00
1 parent 84cb2d6643
commit 1f51a8015d
3 files changed
+123 -53
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.vscode/launch.json
Vendored
+19 -2
View File
@@ -9,7 +9,16 @@
"type": "python",
"request": "launch",
"program": "./mqtt-client.py",
"args": ["-H","localhost","-p","1887","-t","temp","pub"],
"args": [
"-H",
"localhost",
"-p",
"1887",
"-t",
"temp",
"pub",
"-r"
],
"console": "integratedTerminal"
},
{
@@ -17,7 +26,15 @@
"type": "python",
"request": "launch",
"program": "./mqtt-client.py",
"args": ["-H","localhost","-p","1887","-t","temp","sub"],
"args": [
"-H",
"localhost",
"-p",
"1887",
"-t",
"TOTO",
"sub"
],
"console": "integratedTerminal"
},
{
__pycache__/mqtt.cpython-37.pyc
BIN
View File
Binary file not shown.
mqtt.py
+104 -51
View File
@@ -6,7 +6,7 @@ import sys
import os
import traceback
import select
from sys import byteorder, stdin
from sys import byteorder, exc_info, stdin
from typing import BinaryIO, Tuple
PORT = 1883
@@ -15,6 +15,7 @@ TYPE_CONNACK = 0x20
TYPE_PUBLISH = 0x30
TYPE_SUBREQ = 0x82
TYPE_SUBACK = 0x90
TYPE_DISCONNECT = 0xe0
def create_mqtt_publish_msg(topic, value, retain=False):
@@ -82,40 +83,64 @@ def create_mqtt_connack_msg(accepted=True):
return header + msglen + flags + returncode
def create_mqtt_disconnect_msg():
header = (TYPE_DISCONNECT).to_bytes(1, byteorder="big")
msglen = (0).to_bytes(1, byteorder='big')
return header + msglen
def decode_msg(msg):
finished = True
if msg[:1] == TYPE_CONNECT.to_bytes(1, byteorder="big"):
typemsg = "CONNECT"
keepalive = int.from_bytes(msg[11:12], byteorder="big")
clientid = msg[14:].decode("ascii")
return (typemsg, keepalive, clientid)
return [(typemsg, keepalive, clientid)]
if msg[:1] == TYPE_CONNACK.to_bytes(1, byteorder='big'):
typemsg = "CONNACK"
ackflags = msg[2:3].decode('ascii')
code = int.from_bytes(msg[3:4], byteorder="big")
return (typemsg, ackflags, code)
return [(typemsg, ackflags, code)]
if msg[:1] == TYPE_SUBREQ.to_bytes(1, byteorder="big"):
typemsg = "SUBREQ"
msgid = int.from_bytes(msg[2:4], byteorder="big")
topiclenght = int.from_bytes(msg[4:6], byteorder="big")
topic = msg[6:6 + topiclenght].decode('ascii')
return (typemsg, msgid, topic)
return [(typemsg, msgid, topic)]
if msg[:1] == TYPE_SUBACK.to_bytes(1, byteorder="big"):
typemsg = "SUBACK"
msg_lenght = int.from_bytes(msg[1:2], byteorder='big')
msgid = int.from_bytes(msg[2:4], byteorder="big")
return (typemsg, msgid)
if (len(msg[2 + msg_lenght:]) != 0):
finished = False
if not finished:
return [(typemsg, msgid)] + decode_msg(msg[2 + msg_lenght:])
return [(typemsg, msgid)]
if msg[:1] == TYPE_DISCONNECT.to_bytes(1, byteorder="big"):
typemsg = "DISCONNECT"
msglen = msg[1:]
return [(typemsg, msglen)]
if msg[:1] == TYPE_PUBLISH.to_bytes(1, byteorder="big") or msg[:1] == (TYPE_PUBLISH + 1).to_bytes(1, byteorder="big"):
typemsg = "PUBLISH"
messagelenght = int.from_bytes(msg[1:2], byteorder="big")
topiclenght = int.from_bytes(msg[2:4], byteorder="big")
topic = msg[4:4 + topiclenght].decode('ascii')
message = msg[4 + topiclenght:].decode('ascii')
message = msg[4 + topiclenght:2 + messagelenght].decode('ascii')
if (len(msg[2 + messagelenght:]) != 0):
finished = False
if msg[:1] == TYPE_PUBLISH.to_bytes(1, byteorder="big"):
return (typemsg, topic, message, False)
return (typemsg, topic, message, True)
return tuple("ERROR")
if not finished:
return [(typemsg, topic, message, False)] + decode_msg(msg[2 + messagelenght:])
return [(typemsg, topic, message, False)]
if not finished:
return [(typemsg, topic, message, True)] + decode_msg(msg[2 + messagelenght:])
return [(typemsg, topic, message, True)]
return [tuple("ERROR")]
# print(create_mqtt_connect_msg("mosq-6F4yNCdkrVx80t8BVp"))
@@ -125,32 +150,44 @@ def decode_msg(msg):
def run_publisher(addr, topic, pub_id, retain=False):
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect(addr)
s.sendall(create_mqtt_connect_msg(pub_id))
connack = s.recv(127)
if decode_msg(connack)[0] == "CONNACK" and decode_msg(connack)[2] == 0:
while True:
msg = sys.stdin.readline()
sys.stdin.flush()
s.sendall(create_mqtt_publish_msg(topic, msg, retain))
pass
try:
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect(addr)
s.sendall(create_mqtt_connect_msg(pub_id))
connack = s.recv(127)
if decode_msg(connack)[0][0] == "CONNACK" and decode_msg(connack)[0][2] == 0:
for line in stdin:
s.sendall(create_mqtt_publish_msg(topic, line[:-1], retain))
s.sendall(create_mqtt_disconnect_msg())
except KeyboardInterrupt:
s.sendall(create_mqtt_disconnect_msg())
def run_subscriber(addr, topic, sub_id):
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect(addr)
s.sendall(create_mqtt_connect_msg(sub_id))
connack = s.recv(127)
if decode_msg(connack)[0] == "CONNACK" and decode_msg(connack)[2] == 0:
s.sendall(create_mqtt_subscriber_msg(topic))
suback = s.recv(127)
if decode_msg(suback)[0] == 'SUBACK':
while True:
tmp = decode_msg(s.recv(127))
if tmp[0] == "PUBLISH":
print(tmp[1] + ' : ' + tmp[2], end="")
pass
try:
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect(addr)
s.sendall(create_mqtt_connect_msg(sub_id))
connack = s.recv(127)
data = decode_msg(connack)
if data[0][0] == "CONNACK" and data[0][2] == 0:
s.sendall(create_mqtt_subscriber_msg(topic))
suback = s.recv(127)
data = decode_msg(suback)
if data[0][0] == 'SUBACK':
if len(data) > 1:
for d in data:
if d[0] == "PUBLISH":
print(d[1] + ' : ' + d[2])
while True:
pkt = s.recv(1500)
tmp = decode_msg(pkt)
for d in tmp:
if d[0] == "PUBLISH":
print(d[1] + ' : ' + d[2])
pass
except KeyboardInterrupt:
s.sendall(create_mqtt_disconnect_msg())
def run_server(addr):
@@ -160,7 +197,7 @@ def run_server(addr):
s.listen(1)
l = [s]
listconnected = []
retainmsg = [False, "", ""]
retainmsg = []
while True:
l3, _, _ = select.select(l, [], [])
for i in l3:
@@ -170,21 +207,37 @@ def run_server(addr):
else:
data = i.recv(1500)
data = decode_msg(data)
if data[0] == "CONNECT":
i.sendall(create_mqtt_connack_msg(True))
listconnected.append([i, None])
elif data[0] == "SUBREQ":
for o in listconnected:
if i in o:
o[1] = data[2]
i.sendall(create_mqtt_suback_msg())
if retainmsg[0]:
if retainmsg[1] == data[2]:
i.sendall(create_mqtt_publish_msg(data[2], retainmsg[2]))
elif data[0] == "PUBLISH":
if data[3]:
retainmsg = (True, data[1], data[2])
for o in listconnected:
if data[1] in o:
o[0].sendall(create_mqtt_publish_msg(data[1], data[2], data[3]))
pass
for i_data in data:
if i_data[0] == "CONNECT":
i.sendall(create_mqtt_connack_msg(True))
listconnected.append([i, None])
elif i_data[0] == "SUBREQ":
for o in listconnected:
if i in o:
o[1] = i_data[2]
i.sendall(create_mqtt_suback_msg())
for i_retainmsg in retainmsg:
if i_retainmsg[0] == i_data[2]:
i.sendall(create_mqtt_publish_msg(i_data[2], i_retainmsg[1]))
break
elif i_data[0] == "DISCONNECT":
i.close()
l.remove(i)
for o in listconnected:
if i in o:
listconnected.remove(o)
elif i_data[0] == "PUBLISH":
if i_data[3]:
topic_already_on_retain = False
for i_retainmsg in range(len(retainmsg)):
if i_data[1] == retainmsg[i_retainmsg][0]:
topic_already_on_retain = True
retainmsg[i_retainmsg][1] = i_data[2]
break
if not topic_already_on_retain:
retainmsg.append([i_data[1], i_data[2]])
for o in listconnected:
if i_data[1] in o:
pkt = create_mqtt_publish_msg(i_data[1], i_data[2], i_data[3])
o[0].sendall(pkt)
pass