47
47
# MQTT Commands
48
48
MQTT_PINGREQ = b"\xc0 \0 "
49
49
MQTT_PINGRESP = const (0xD0 )
50
+ MQTT_PUBLISH = const (0x30 )
50
51
MQTT_SUB = b"\x82 "
51
52
MQTT_UNSUB = b"\xA2 "
52
53
MQTT_DISCONNECT = b"\xe0 \0 "
@@ -879,7 +880,9 @@ def loop(self, timeout=0):
879
880
def _wait_for_msg (self , timeout = 0.1 ):
880
881
# pylint: disable = too-many-return-statements
881
882
882
- """Reads and processes network events."""
883
+ """Reads and processes network events.
884
+ Return the packet type or None if there is nothing to be received.
885
+ """
883
886
# CPython socket module contains a timeout attribute
884
887
if hasattr (self ._socket_pool , "timeout" ):
885
888
try :
@@ -909,8 +912,11 @@ def _wait_for_msg(self, timeout=0.1):
909
912
"Unexpected PINGRESP returned from broker: {}." .format (sz )
910
913
)
911
914
return MQTT_PINGRESP
912
- if res [0 ] & 0xF0 != 0x30 :
915
+
916
+ if res [0 ] & 0xF0 != MQTT_PUBLISH :
913
917
return res [0 ]
918
+
919
+ # Handle only the PUBLISH packet type from now on.
914
920
sz = self ._recv_len ()
915
921
# topic length MSB & LSB
916
922
topic_len = self ._sock_exact_recv (2 )
@@ -923,12 +929,13 @@ def _wait_for_msg(self, timeout=0.1):
923
929
pid = self ._sock_exact_recv (2 )
924
930
pid = pid [0 ] << 0x08 | pid [1 ]
925
931
sz -= 0x02
932
+
926
933
# read message contents
927
934
raw_msg = self ._sock_exact_recv (sz )
928
935
msg = raw_msg if self ._use_binary_mode else str (raw_msg , "utf-8" )
929
936
if self .logger is not None :
930
937
self .logger .debug (
931
- "Receiving SUBSCRIBE \n Topic: %s\n Msg: %s\n " , topic , raw_msg
938
+ "Receiving PUBLISH \n Topic: %s\n Msg: %s\n " , topic , raw_msg
932
939
)
933
940
self ._handle_on_message (self , topic , msg )
934
941
if res [0 ] & 0x06 == 0x02 :
@@ -937,6 +944,7 @@ def _wait_for_msg(self, timeout=0.1):
937
944
self ._sock .send (pkt )
938
945
elif res [0 ] & 6 == 4 :
939
946
assert 0
947
+
940
948
return res [0 ]
941
949
942
950
def _recv_len (self ):
0 commit comments