<jupyter_start><jupyter_text>Multi-Region IoT - Persist DataPersist data accross regions by using [Amazon Simple Queue Service](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/welcome.html) (SQS) or DynamoDB global tables.  Libraries<jupyter_code>import boto3
import datetime
import json
import logging
import time

from AWSIoTPythonSDK.MQTTLib import AWSIoTMQTTClient
from os.path import join<jupyter_output><empty_output><jupyter_text>Note: If you get an error that the AWSIoTPythonSDK is not installed, install the SDK with the command below and import the libraries again!<jupyter_code>!pip install AWSIoTPythonSDK -t .<jupyter_output><empty_output><jupyter_text>Restore variable that have been defined in the notebook which has been used to create the ACM PCA setup.<jupyter_code>%store -r config
print("config: {}".format(json.dumps(config, indent=4, default=str)))<jupyter_output><empty_output><jupyter_text>Global VarsFor several actions in AWS IoT topic rules a service role is required. This roles allows the IoT Core to access other services. A role has already been created by the CloudFormation stack for the master region. The arn for this role can be found in the outputs section of your CloudFormation stack next to the key **IoTAccessServicesRoleArn**.Set the variable **topic_rule_role_arn** to contain this arn.<jupyter_code>topic_rule_role_arn = 'YOUR_ROLE_ARN_HERE'

topic_rule_name_sqs = 'IoTMRCrossRegionSQSRule'
topic_rule_name_sns = 'IoTMRCrossRegionSNSRule'

queue_name = 'IoTCrossRegion'
topic_name = 'IoTCrossRegion'<jupyter_output><empty_output><jupyter_text>IoT EndpointTo connect a device to the AWS IoT master region we need to get the iot endpoint.<jupyter_code>c_iot = boto3.client('iot', region_name = config['aws_region_master'])<jupyter_output><empty_output><jupyter_text>Get the iot endpoint.<jupyter_code>response = c_iot.describe_endpoint(endpointType='iot:Data-ATS')
iot_endpoint = response['endpointAddress']
print("iot_endpoint: {}".format(iot_endpoint))<jupyter_output><empty_output><jupyter_text>Transfer data with Amazon Simple Queue Service (SQS)By using SQS in a topic rule data can be send across regions. An SQS queue in the slave region will be created and a topic rule in the master region which transfers incoming messages to the SQS queue in the slave region. SQS client in slave region.<jupyter_code>c_sqs = boto3.client('sqs', region_name = config['aws_region_slave'])<jupyter_output><empty_output><jupyter_text>Queue operationsCreate the queue and get the queue url. The queue url is required to create the IoT topic rule. Create the queue<jupyter_code>response = c_sqs.create_queue(QueueName=queue_name)
print("response: {}".format(json.dumps(response, indent=4, default=str)))<jupyter_output><empty_output><jupyter_text>Get the queue url<jupyter_code>response = c_sqs.get_queue_url(QueueName=queue_name)
print("response: {}\n".format(json.dumps(response, indent=4, default=str)))
queue_url = response['QueueUrl']
print("queue_url: {}".format(queue_url))<jupyter_output><empty_output><jupyter_text>Topic ruleCreate the rule the topic rule. Create the topic rule<jupyter_code>response = c_iot.create_topic_rule(
    ruleName=topic_rule_name_sqs,
    topicRulePayload={
        'awsIotSqlVersion': '2016-03-23',
        'sql': 'SELECT * FROM \'cmd/+/cross/region\'',
        'actions': [{
                'sqs': {
                        'roleArn': topic_rule_role_arn,
                        'queueUrl': queue_url,
                        'useBase64': False
                }
            }],
        'ruleDisabled': False
    }
)

print("response: {}".format(json.dumps(response, indent=4, default=str)))<jupyter_output><empty_output><jupyter_text>VerifyGet the topic rule to verify that it has been created successfully.<jupyter_code>response = c_iot.get_topic_rule(
    ruleName=topic_rule_name_sqs
)
print("response: {}".format(json.dumps(response, indent=4, default=str)))<jupyter_output><empty_output><jupyter_text>Transfer data with Amazon Simple Notification Service (SNS) and AWS LambdaAnother option to transfer data across regions is the use of SNS in combination with a Lambda function. In this example an SNS topic in the master region will be created. A Lambda function in the slave region was already created by CloudFormation. The Lambda endpoint from the slave region will be subscribed to the SNS topic.<jupyter_code>c_sns = boto3.client('sns', region_name = config['aws_region_master'])<jupyter_output><empty_output><jupyter_text>Create the topic.<jupyter_code>response = c_sns.create_topic(Name=topic_name)
print("response: {}".format(json.dumps(response, indent=4, default=str)))<jupyter_output><empty_output><jupyter_text>Get the topic arn.<jupyter_code>topic_arn = response['TopicArn']
print("topic_arn: {}".format(topic_arn))<jupyter_output><empty_output><jupyter_text>Subscribe to SNS topicThe Lambda function in the slave region has been created by the slave CloudFormation stack. Set the variable **lambda_arn** to this arn.The arn for the Lambda can be found in the outputs section of your CloudFormation stack next to the key **CrossRegionLambdaFunctionArn**.<jupyter_code>lambda_arn = 'YOUR_LAMBDA_ARN_HERE'
lambda_name = lambda_arn.split(':')[-1]
statement_id = str(int(time.time()))
print("lambda_name: {} statement_id: {}".format(lambda_name, statement_id))<jupyter_output><empty_output><jupyter_text>Add permission to the Lambda functionTo allow SNS to invoke the lambda function a permission must be added to the function.<jupyter_code>c_lambda = boto3.client('lambda', region_name = config['aws_region_slave'])
response = c_lambda.add_permission(
    FunctionName=lambda_name,
    StatementId=statement_id,
    Action='lambda:invokeFunction',
    Principal='sns.amazonaws.com',
    SourceArn=topic_arn
)

print("response: {}".format(json.dumps(response, indent=4, default=str)))
response = c_sns.subscribe(
    TopicArn=topic_arn,
    Protocol='lambda',
    Endpoint=lambda_arn,
    ReturnSubscriptionArn=True
)

print("response: {}\n".format(json.dumps(response, indent=4, default=str)))
subscription_arn = response['SubscriptionArn']
print("subscription_arn: {}".format(subscription_arn))<jupyter_output><empty_output><jupyter_text>Create topic ruleCreate a topic rule to forward messages to the SNS topic.<jupyter_code>response = c_iot.create_topic_rule(
    ruleName=topic_rule_name_sns,
    topicRulePayload={
        'awsIotSqlVersion': '2016-03-23',
        'sql': 'SELECT * FROM \'cmd/+/cross/region\'',
        'actions': [{
                'sns': {
                    'targetArn': topic_arn,
                    'roleArn': topic_rule_role_arn,
                    'messageFormat': 'RAW'
                }
            }],
        'ruleDisabled': False
    }
)

print("response: {}".format(json.dumps(response, indent=4, default=str)))<jupyter_output><empty_output><jupyter_text>VerifyGet the topic rule to verify that it has been created successfully.<jupyter_code>response = c_iot.get_topic_rule(ruleName=topic_rule_name_sns)
print("response: {}".format(json.dumps(response, indent=4, default=str)))<jupyter_output><empty_output><jupyter_text>Connect a DeviceConnect a device that you created earlier to the message broker from AWS IoT and send some messages. These messages will be forwarded by the topic rules that you created to SQS as well as SNS.<jupyter_code>thing_name = 'thing-mr04'
root_ca = 'AmazonRootCA1.pem'

device_key_file = '{}.device.key.pem'.format(thing_name)
device_cert_file = '{}.device.cert.pem'.format(thing_name)

# AWS IoT Python SDK needs logging
logger = logging.getLogger("AWSIoTPythonSDK.core")
#logger.setLevel(logging.DEBUG)
logger.setLevel(logging.INFO)
streamHandler = logging.StreamHandler()
formatter = logging.Formatter("[%(asctime)s - %(levelname)s - %(filename)s:%(lineno)s - %(funcName)s - %(message)s")
streamHandler.setFormatter(formatter)
logger.addHandler(streamHandler)

myAWSIoTMQTTClient = None
myAWSIoTMQTTClient = AWSIoTMQTTClient(thing_name)
myAWSIoTMQTTClient.configureEndpoint(iot_endpoint, 8883)
myAWSIoTMQTTClient.configureCredentials(root_ca, 
                                        join(config['PCA_directory'], device_key_file), 
                                        join(config['PCA_directory'], device_cert_file))

# AWSIoTMQTTClient connection configuration
myAWSIoTMQTTClient.configureAutoReconnectBackoffTime(1, 32, 20)
myAWSIoTMQTTClient.configureOfflinePublishQueueing(-1)  # Infinite offline Publish queueing
myAWSIoTMQTTClient.configureDrainingFrequency(2)  # Draining: 2 Hz
myAWSIoTMQTTClient.configureConnectDisconnectTimeout(10)  # 10 sec
myAWSIoTMQTTClient.configureMQTTOperationTimeout(5)  # 5 sec

# Connect and reconnect to AWS IoT
try:
    myAWSIoTMQTTClient.connect()
except Exception as e:
    logger.error('{}'.format(e))
    time.sleep(5)
    myAWSIoTMQTTClient.connect()<jupyter_output><empty_output><jupyter_text>Publish messagesPublish some messages that should be transferred to the SQS queue in the other region.**Hint:** Before publishing message subscribe in the test client in the master region in the AWS IoT Core console to the topic `cmd/+/cross/region`. By doing so you can verify that your messages are reaching the IoT Core.<jupyter_code>topic = 'cmd/{}/cross/region'.format(thing_name)
print("topic: {}".format(topic))

for i in range(5):
    date_time = datetime.datetime.now().isoformat()
    message = {"thing_name": "{}".format(thing_name), "date_time": date_time, "i": i}
    print("publish: message: {}".format(message))
    myAWSIoTMQTTClient.publish(topic, json.dumps(message), 0)
    time.sleep(1)<jupyter_output><empty_output><jupyter_text>Poll the SQS queueTo verify that the messages has been sent to SQS in the slave region poll the queue for messages. You should get messages from the queue. Feel free to execute polling multiple times.<jupyter_code># Long poll for message on provided SQS queue
response = c_sqs.receive_message(
    QueueUrl=queue_url,
    AttributeNames=[
        'All'
    ],
    MaxNumberOfMessages=10,
    MessageAttributeNames=[
        'All'
    ],
    WaitTimeSeconds=20
)

print("queue_url: {}\n".format(queue_url))

for message in response['Messages']:
    body = message['Body']
    message_id = message['MessageId']
    #print(message)
    print("message_id: {}\nbody: {}\n".format(message_id, body))<jupyter_output><empty_output><jupyter_text>SNS/LambdaTo verify that message have been sent also to the slave region with SNS and Lambda watch at CloudWatch in the slave region for the logs of your Lambda function.Logs can be found in `CloudWatch -> Logs -> /aws/lambda/<LAMBDA_FUNCTION_NAME`.  DisconnetDisconnect the device from AWS IoT Core.<jupyter_code>myAWSIoTMQTTClient.disconnect()<jupyter_output><empty_output><jupyter_text>Clean UpClean up your environment:* Remove permissions from the Lambda function* Unsubscribe the Lambda from the SNS topic* Delete SNS topic* Delete IoT topic rules<jupyter_code>response = c_lambda.remove_permission(
    FunctionName=lambda_name,
    StatementId=statement_id
)
print("response: {}".format(json.dumps(response, indent=4, default=str)))
response = c_sns.unsubscribe(SubscriptionArn=subscription_arn)
print("response: {}".format(json.dumps(response, indent=4, default=str)))
response = c_sns.delete_topic(TopicArn=topic_arn)
print("response: {}".format(json.dumps(response, indent=4, default=str)))
response = c_iot.delete_topic_rule(ruleName=topic_rule_name_sqs)
print("response: {}".format(json.dumps(response, indent=4, default=str)))
response = c_iot.delete_topic_rule(ruleName=topic_rule_name_sns)
print("response: {}".format(json.dumps(response, indent=4, default=str)))
response = c_sns.delete_topic(TopicArn=topic_arn)
print("response: {}".format(json.dumps(response, indent=4, default=str)))<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.10609385550303849, 'fraction_numerical': 0.005823767724510466, 'mean_word_length': 5.701923076923077, 'pattern_counts': {'":': 3, '<': 94, '<?xml version=': 0, '>': 95, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 3, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '23', 'dir': 'jupyter-structured-clean-dedup', 'id': '488756', 'license': 'MIT-0', 'n_tokens_mistral': 3811, 'n_tokens_neox': 3693, 'n_words': 925, 'path': 'jupyter/10_Multi_Region_Data.ipynb', 'repo_name': 'aws-samples/reinvent2019-iot335-code', 'size': '11848'}