ConnectionSample.py

import logging
from enos.core.MqttClient import MqttClient
from enos.core.internal.Profile import Profile
from enos.sample.SampleHelper import SampleHelper

# these file are generated by get_cert.py via EnOS cert tool, for tls connection
ca_file = 'edge_ca.pem'
key_file = 'edge.key'
cer_file = 'edge.pem'
key_file_password = 'PRIVATE_KEY_PASSWORD'


def on_connect():
    """ Called when the connection to the server is completed."""
    print('connect success')


def on_disconnect():
    """ Called when the client connection lost."""
    print('connect lost')


def on_connect_failed():
    """ Called when the client connect failed"""
    print('connect failed...')


def on_device_dynamic_activation(device_secret):
    """ Record the device_secret for further connection."""
    print('device dynamic activate success, the device secret is:' + device_secret)


def static_device_login():
    static_profile = Profile.create_instance(SampleHelper.get_device_static_login())
    client = MqttClient(profile=static_profile)
    client.get_profile().set_auto_reconnect(True)

    # register connection callback
    client.on_connect = on_connect
    client.on_disconnect = on_disconnect
    client.on_connected_failed = on_connect_failed

    client.connect()  # connect in sync
    client.close()


def dynamic_device_login():
    dynamic_activation_profile = Profile.create_instance(SampleHelper.get_device_dynamic_login())
    client = MqttClient(profile=dynamic_activation_profile)
    client.get_profile().set_auto_reconnect(True)
    # register dynamic activation callback to acquire device_secret
    client.on_device_dynamic_active = on_device_dynamic_activation

    # register connection callback
    client.on_connect = on_connect
    client.on_disconnect = on_disconnect
    client.on_connected_failed = on_connect_failed

    client.connect()  # connect in sync
    client.close()


def secure_login():
    static_ssl_profile = Profile.create_instance(SampleHelper.get_device_static_ssl_login())
    client = MqttClient(profile=static_ssl_profile)
    client.get_profile().set_auto_reconnect(True)
    # set the certificate files for bi-directional certification
    client.get_profile().set_ssl_context(ca_file, cer_file, key_file, key_file_password)
    client.setup_basic_logger(logging.INFO)

    # register connection callback
    client.on_connect = on_connect
    client.on_disconnect = on_disconnect
    client.on_connected_failed = on_connect_failed

    client.connect()  # connect in sync
    client.close()


if __name__ == "__main__":
    # static_device_login()
    # dynamic_device_login()
    secure_login()