- Notifications
You must be signed in to change notification settings - Fork 219
/
Copy pathpubsub.py
157 lines (128 loc) · 6.33 KB
/
pubsub.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0.
fromawscrtimportmqtt, http
fromawsiotimportmqtt_connection_builder
importsys
importthreading
importtime
importjson
fromutils.command_line_utilsimportCommandLineUtils
# This sample uses the Message Broker for AWS IoT to send and receive messages
# through an MQTT connection. On startup, the device connects to the server,
# subscribes to a topic, and begins publishing messages to that topic.
# The device should receive those same messages back from the message broker,
# since it is subscribed to that same topic.
# cmdData is the arguments/input from the command line placed into a single struct for
# use in this sample. This handles all of the command line parsing, validating, etc.
# See the Utils/CommandLineUtils for more information.
cmdData=CommandLineUtils.parse_sample_input_pubsub()
received_count=0
received_all_event=threading.Event()
# Callback when connection is accidentally lost.
defon_connection_interrupted(connection, error, **kwargs):
print("Connection interrupted. error: {}".format(error))
# Callback when an interrupted connection is re-established.
defon_connection_resumed(connection, return_code, session_present, **kwargs):
print("Connection resumed. return_code: {} session_present: {}".format(return_code, session_present))
ifreturn_code==mqtt.ConnectReturnCode.ACCEPTEDandnotsession_present:
print("Session did not persist. Resubscribing to existing topics...")
resubscribe_future, _=connection.resubscribe_existing_topics()
# Cannot synchronously wait for resubscribe result because we're on the connection's event-loop thread,
# evaluate result with a callback instead.
resubscribe_future.add_done_callback(on_resubscribe_complete)
defon_resubscribe_complete(resubscribe_future):
resubscribe_results=resubscribe_future.result()
print("Resubscribe results: {}".format(resubscribe_results))
fortopic, qosinresubscribe_results['topics']:
ifqosisNone:
sys.exit("Server rejected resubscribe to topic: {}".format(topic))
# Callback when the subscribed topic receives a message
defon_message_received(topic, payload, dup, qos, retain, **kwargs):
print("Received message from topic '{}': {}".format(topic, payload))
globalreceived_count
received_count+=1
ifreceived_count==cmdData.input_count:
received_all_event.set()
# Callback when the connection successfully connects
defon_connection_success(connection, callback_data):
assertisinstance(callback_data, mqtt.OnConnectionSuccessData)
print("Connection Successful with return code: {} session present: {}".format(callback_data.return_code, callback_data.session_present))
# Callback when a connection attempt fails
defon_connection_failure(connection, callback_data):
assertisinstance(callback_data, mqtt.OnConnectionFailureData)
print("Connection failed with error code: {}".format(callback_data.error))
# Callback when a connection has been disconnected or shutdown successfully
defon_connection_closed(connection, callback_data):
print("Connection closed")
if__name__=='__main__':
# Create the proxy options if the data is present in cmdData
proxy_options=None
ifcmdData.input_proxy_hostisnotNoneandcmdData.input_proxy_port!=0:
proxy_options=http.HttpProxyOptions(
host_name=cmdData.input_proxy_host,
port=cmdData.input_proxy_port)
# Create a MQTT connection from the command line data
mqtt_connection=mqtt_connection_builder.mtls_from_path(
endpoint=cmdData.input_endpoint,
port=cmdData.input_port,
cert_filepath=cmdData.input_cert,
pri_key_filepath=cmdData.input_key,
ca_filepath=cmdData.input_ca,
on_connection_interrupted=on_connection_interrupted,
on_connection_resumed=on_connection_resumed,
client_id=cmdData.input_clientId,
clean_session=False,
keep_alive_secs=30,
http_proxy_options=proxy_options,
on_connection_success=on_connection_success,
on_connection_failure=on_connection_failure,
on_connection_closed=on_connection_closed)
ifnotcmdData.input_is_ci:
print(f"Connecting to {cmdData.input_endpoint} with client ID '{cmdData.input_clientId}'...")
else:
print("Connecting to endpoint with client ID")
connect_future=mqtt_connection.connect()
# Future.result() waits until a result is available
connect_future.result()
print("Connected!")
message_count=cmdData.input_count
message_topic=cmdData.input_topic
message_string=cmdData.input_message
# Subscribe
print("Subscribing to topic '{}'...".format(message_topic))
subscribe_future, packet_id=mqtt_connection.subscribe(
topic=message_topic,
qos=mqtt.QoS.AT_LEAST_ONCE,
callback=on_message_received)
subscribe_result=subscribe_future.result()
print("Subscribed with {}".format(str(subscribe_result['qos'])))
# Publish message to server desired number of times.
# This step is skipped if message is blank.
# This step loops forever if count was set to 0.
ifmessage_string:
ifmessage_count==0:
print("Sending messages until program killed")
else:
print("Sending {} message(s)".format(message_count))
publish_count=1
while (publish_count<=message_count) or (message_count==0):
message="{} [{}]".format(message_string, publish_count)
print("Publishing message to topic '{}': {}".format(message_topic, message))
message_json=json.dumps(message)
mqtt_connection.publish(
topic=message_topic,
payload=message_json,
qos=mqtt.QoS.AT_LEAST_ONCE)
time.sleep(1)
publish_count+=1
# Wait for all messages to be received.
# This waits forever if count was set to 0.
ifmessage_count!=0andnotreceived_all_event.is_set():
print("Waiting for all messages to be received...")
received_all_event.wait()
print("{} message(s) received.".format(received_count))
# Disconnect
print("Disconnecting...")
disconnect_future=mqtt_connection.disconnect()
disconnect_future.result()
print("Disconnected!")