소개

AWS IoT Core의 기본 Ingest 기능이 출시되었습니다.
데이터 전송만을 원하는 유스 케이스에서 사용하면 Message broker에 연결하지 않고 규칙 엔진을 발화할 수 있는 것 같습니다.
메리트는 날린, 메시지 브로커의 비용이 걸리지 않는다고 하는 것으로, 데이터 송신만을 생각하는 유스 케이스로 편리할까 생각합니다.

사용법

topic의 시작 부분에 "$aws/rules/{실행하려는 규칙 이름}을 추가하기만 하면 됩니다.
규칙 엔진 설명은 $aws/rules/{rule 이름} 을 생략할 수 있습니다.
즉, data/{device_name}에서 데이터를 전송하는 경우,

보낸 사람의 주제

$aws/rules/{rule_name}/data/{device_name}

규칙 엔진 SQL 절을 작성하는 방법

data/#

잘, Device 측에서 두는 topic명의 update를 할 수 있으면, AWS IoT측의 설정 변경 불필요합니다.
기본 Ingest 주제의 처음 세 레벨($aws/rules/rule-name)은 주제의 8개 세그먼트 길이 한계 또는 256자 합계 한계로 계산되지 않습니다.
또 존재하지 않는 룰명을 topic에 지정하면, RuleNotFound의 metric이 증가한다.

공개 개발 문서는 여기

테스트

Lambda 설치

Basic Ingest가 오면 받은 메시지를 되돌리는 Lambda를 준비합니다. 뒤집는 topic은 "basic_ingest/got"이라고 합니다.
role은 Lambdabasic과 AWS IoT에 publish할 수 있는 role가 필요합니다.

just_publish.py

from __future__ import print_function
import json
import boto3

iot = boto3.client('iot-data')

def lambda_handler(event, context):
  try:
    iot.publish(
        topic = "basic_ingest/got",
        qos = 0,
        payload = json.dumps(event)
    )
  except Exception as e:
    print("exception")
    print(e.message)

  return

AWS IoT Rule 엔진

다음과 같이 설정했습니다.

테스트 클라이언트

클래스에서 숨겨진 Python Device SDK는 생략되었습니다. 자세한 내용은 여기을 참조하십시오.
하고 있는 일은 단순히 $aws/rules/basic_ingest(상에서 설정한 룰명)/data/인수로 지정한 디바이스명
무한 루프로 데이터를 게시합니다.
또한 Shadow와 동기화하고 있지만 이번에는별로 관련이 없으므로 무시하십시오.
github의 바시 c부 b스 b. py이면 인수의 topic에 사용하고 싶은 topic을 지정하는 것으로 거의 같을 수 있습니다.

test_client.py

import sys
import time
import logging
import argparse
import IoT_common as IoT
import dependency as dc


BASIC_INGEST = "$aws/rules/basic_ingest/data/"
#-------------------------------------------------------------------------------
def init_log():
    logger = logging.getLogger()
    handler = logging.StreamHandler(sys.stdout)
    logger.addHandler(handler)
    logger.setLevel(logging.WARN)
    logging.basicConfig()

    return logger
#-------------------------------------------------------------------------------
def device_main():
    try:
        logger = init_log()
        logger.info("start")
        parser = argparse.ArgumentParser()
        init_info = dc.arg_check(parser)
        device_name = init_info['device_name']
        iot = IoT.IoT_common(device_name,
                             init_info['endpoint'],
                             init_info['certs'])
        iot.create_shadow()
        status_str = dc.set_status(iot.WAIT_TIME)
        iot.updateShadow(status_str)
        iot.create_mqtt()
        topic = BASIC_INGEST + device_name
        while True:
            payload = dc.create_data(device_name)
            logger.debug("payload:{}".format(payload))
            iot.PublishMsg(topic, payload)
            time.sleep(iot.WAIT_TIME)
    except Exception as e:
        logger.error("Error main()")
        logger.error("error msg:\n{}".format(e.message))

if __name__ == '__main__':
    device_main()

에서 이 클라이언트를 시작합니다.

테스트 확인

AWS IoT 테스트 화면에서 #을 subscribe해 봅니다.
기본 ingest는 규칙 엔진에 직접 연결되어 있으므로 subscribe에는 표시되지 않으며 Lambda가 되돌린 메시지만 표시됩니다.


그래서, 메시지 브로커를 통과하지 않으면 테스트 클라이언트의 메시지는 subscriber에서 수신 할 수 없지만 규칙 엔진이 점화되고 있음을 알았습니다.

요약

그래서 device가 사용하는 topic의 변경만으로 이 기능을 사용할 수 있는 것을 알았습니다. 주의점으로서는 policy의 운용을 엄격하게 하고 있는 분은, Resource 관리로 이 topic을 용서하는 것을 잊지 않고, 라고 하는 근처가 됩니다.
또, 메시지 publish를 subscriber에게 메시지를 전달하고 싶은 경우는, 이 기능이 아니고 종래의 topic을 사용한 pub/sub의 메시지 브로커 경유로 사용할 필요가 있습니다.

면책

본 투고는, 개인의 의견으로, 소속하는 기업이나 단체는 관계 없습니다.
또 게재하고 있는 sample 프로그램의 동작에 관해서도 보장하지 않으므로, 참고 정도로 해 주세요.