Permalink
Cannot retrieve contributors at this time
Name already in use
A tag already exists with the provided branch name. Many Git commands accept both tag and branch names, so creating this branch may cause unexpected behavior. Are you sure you want to create this branch?
Handling_connection_Multiple_clients/demo1_recording
Go to fileThis commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
122 lines (95 sloc)
2.79 KB
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
""" | |
Creates multiple Connections to a broker | |
and sends and receives messages. Support SSL and Normal connections | |
uses the loop_start and stop functions just like a single client | |
Shows number of thread used | |
""" | |
import paho.mqtt.client as mqtt | |
import time | |
import json | |
import threading | |
import logging | |
broker="192.168.1.159" | |
port =1883 | |
ssl_port=8883 #ssl | |
Normal_connetions=1 | |
SSL_Connections=0 #no SSL connections, illustration-purposes | |
message="test message" | |
topic="broker-test/test1' | |
out_queue=[] #use simple array to get printed message in some form | |
def on_log(client,userdata,level,buf): | |
print(buf) | |
def on_message(client,userdata,message): | |
time.sleep(1) | |
msg="message received",str(message.payload.decode("utf-8")) | |
out_queue.append(msg) | |
def on_connect(client,userdata,flags,rc): | |
if rc==0: | |
client.connected_flag=True #set flags | |
client.subscribe(topic) | |
else: | |
print("Bad connection Retured code" ,rc) | |
client.loop_stop() | |
def on_disconnect(client,userdata,mid): | |
pass | |
def on_publish(client,userdata,mid): | |
time.sleep(1) | |
print("In on pub callback mid=" ,mid) | |
def Create_connections(nclients,port,SSL_flag=False): | |
for i in range(nclients): | |
if SSL_Flag: | |
cname="python-ssl"+str(i) | |
else: | |
cname="python-"+str(i) | |
client=mqtt.Client(cname) #create new instance | |
if SSL_Flag: | |
client.tls_set("ca-pi.crt') | |
client.connect(broker,port) #establish connection | |
#client.on_log=on_log #give detailed logging | |
client.on_connect=on_connect | |
client.on_disconnect=on_disconnect | |
#client.on_publish=on_publish | |
client.on_message=on_message | |
clients.append(client) | |
client.loop_start() | |
while not client.connected_flag: | |
time.sleep(0.05) | |
mqtt.Client.connected_flag=False #create flag in class | |
clients=[] | |
no_threads=threading.active_count() | |
print("current threads=",no_threads) | |
print("Creating Normal Connections",Normal_connections,"clients") | |
Create_connections(Normal_connections,port,False) | |
if SSL_Connections!=0: | |
print("Creating SSL_Connections",SSL_Connections,"clients") | |
Create_connections(SSL_Connections,ssl_port,True) | |
print("All clients connected") | |
time.sleep(5) | |
# | |
count=0 | |
no_threads=threading.active_count() | |
print("current_threads=",no_threads) | |
print("Publishing") | |
Run_Flag=True | |
try: | |
while Run_Flag: | |
i=0 | |
for client in clients: | |
counter=str(count).rjust(6,"0") | |
msg="client"+str(i)+""+counter+"XXXXXX"+message | |
client.publish(topic,msg) | |
time.sleep(0.1) | |
print("publishing client" +str(i)) | |
i+=1 | |
time.sleep(10) #now print messages | |
print("queue length=",len(out_queue)) | |
for x in range(len(out_queue)): | |
print(out_queue.pop()) | |
count+=1 | |
except KeyboardInterrupt: | |
print("interrupted by keyboard") | |
for client in clients: | |
client.disconnect() | |
client.loop_stop() | |
#allow time for allthreads to stop before exiting | |
time.sleep(10) |