[Python] How can I fix Error EOF occurred in violation of protocol (_ssl.c:2393) with AWS IOT...

I am running python to send MQTT messages to an AWS-IOT server. It runs on a Raspberry Pi, and sends a message each time a button is pressed. It works for a while, but soon a message is sent that does not get received by the MQTT Test Client. If I hit the button again, and it tries to send a message I get the error:

ERROR - Error EOF occurred in violation of protocol (_ssl.c:2393) trying to send message:EOF occurred in violation of protocol (_ssl.c:2393)

import os
import sys
import glob
import json
import requests
import threading
import subprocess
from awscrt import mqtt
from awsiot import mqtt_connection_builder

import config
from log import log

cmd = "cat /proc/cpuinfo | grep Serial | cut -d ' ' -f 2 | md5sum | cut -d ' ' -f 1 | tail -c 6"
result = subprocess.check_output(cmd, shell=True)
client_id = result.decode('utf-8').strip('\n')
log.info(f"Serial number: {result} Client ID:{client_id}")


received_count = 0
received_all_event = threading.Event()

# Callback when connection is accidentally lost.
def on_connection_interrupted(connection, error, **kwargs):
print("Connection interrupted. error: {}".format(error))


# Callback when an interrupted connection is re-established.
def on_connection_resumed(connection, return_code, session_present, **kwargs):
print("Connection resumed. return_code: {} session_present: {}".format(return_code, session_present))

if return_code == mqtt.ConnectReturnCode.ACCEPTED and not session_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)


def on_resubscribe_complete(resubscribe_future):
resubscribe_results = resubscribe_future.result()
print("Resubscribe results: {}".format(resubscribe_results))

for topic, qos in resubscribe_results['topics']:
if qos is None:
sys.exit("Server rejected resubscribe to topic: {}".format(topic))


# Callback when the subscribed topic receives a message
def on_message_received(topic, payload, dup, qos, retain, **kwargs):
print("Received message from topic '{}': {}".format(topic, payload))
global received_count
received_count += 1
received_all_event.set()


# Callback when the connection successfully connects
def on_connection_success(connection, callback_data):
print("Connection Successful with return code: {} session present: {}". \
format(callback_data.return_code, callback_data.session_present))


# Callback when a connection attempt fails
def on_connection_failure(connection, callback_data):
print("Connection failed with error code: {}".format(callback_data.error))


# Callback when a connection has been disconnected or shutdown successfully
def on_connection_closed(connection, callback_data):
print("Connection closed")



class AWS:
def __init__(self, certificate_id, client_id, preface='product'):
"""
Initialize MQTT
"""
# Create a MQTT connection from the command line data
ca = os.path.join(config.CERT_DIR, 'AmazonRootCA1.pem')
certfile=os.path.join(config.CERT_DIR, f"{certificate_id}-certificate.pem.crt")
keyfile =os.path.join(config.CERT_DIR, f"{certificate_id}-private.pem.key")

self.mqtt_connection = mqtt_connection_builder.mtls_from_path(
endpoint=config.endpoint,
port=8883,
cert_filepath=certfile,
pri_key_filepath=keyfile,
ca_filepath=ca,
on_connection_interrupted=on_connection_interrupted,
on_connection_resumed=on_connection_resumed,
client_id=f'{preface}_{client_id}',
clean_session=False,
keep_alive_secs=30,
on_connection_success=on_connection_success,
on_connection_failure=on_connection_failure,
on_connection_closed=on_connection_closed)

def connect(self,):
self.connect_future = self.mqtt_connection.connect()
self.connect_future.result()
log.info("Connected!")

def subscribe(self, topic, callback):
self.subscriber, packet_id = self.mqtt_connection.subscribe(topic=topic,
qos=mqtt.QoS.AT_LEAST_ONCE,
callback=callback)

subscribe_result = self.subscriber.result()
log.info("Subscribed with {}".format(str(subscribe_result['qos'])))

def publish(self, topic, message):
result = self.mqtt_connection.publish(topic=topic,
payload=message,
qos=mqtt.QoS.AT_LEAST_ONCE)
log.info(f"Published:{message} Topic:{topic} Result:{result}")
log.info(self.mqtt_connection.get_stats())

def disconnect(self):
disconnect_future = self.mqtt_connection.disconnect()
log.info(f"Disconnect:{disconnect_future.result()}")


I am not sure whats happening or how to fix it. If any one any suggestions I'd appreciate it.

Continue reading...