BLE 센서가 짧은 상태값은 잘 보내는데 긴 문자열에서는 끝부분을 잃는다면, 무선 감도와 함께 애플리케이션 메시지의 크기를 확인해야 합니다. GATT 특성 하나에 담고 싶은 데이터와 한 번의 알림으로 전달할 수 있는 데이터는 크기가 다를 수 있습니다.
이 글은 일반적인 단일 GATT Notification을 이용하는 자체 프로토콜의 설계 예입니다. 특정 센서의 공식 데이터 형식이나 실제 무선 성능 시험 결과는 아닙니다.
MTU에서 실제 데이터 공간 계산하기
일반적인 ATT Handle Value Notification에는 opcode 1바이트와 handle 2바이트가 들어가므로 특성 값의 공간은 ATT_MTU - 3입니다. 긴 특성 값 전체가 자동으로 여러 Notification으로 나뉘어 전송된다고 가정하면 안 됩니다. Bluetooth SIG의 ATT 명세에 이 구조가 정의되어 있습니다.
LE의 기본 Unenhanced ATT에서는 기본 MTU가 23바이트이므로 특성 값은 20바이트입니다. 이 기본값은 EATT까지 포괄하는 설명이 아닙니다. 연결 후에는 해당 ATT bearer에 실제로 설정된 MTU와 장치 API의 제한을 사용해야 합니다. GATT 명세의 LE MTU 설명
예를 들어 실제 MTU가 23이고 자체 조각 헤더가 4바이트라면, 한 알림에 실을 본문은 16바이트입니다. 문자 수로 잘라서는 안 됩니다. UTF-8 한글은 여러 바이트이므로 문자열을 먼저 바이트로 인코딩하고, 수신 측에서 전체를 합친 다음 디코딩합니다.
Notification과 업무 완료 응답 구분하기
Notification에는 ATT 수준의 응답 PDU가 없고, Indication에는 Confirmation이 있습니다. 이는 Bluetooth SIG의 ATT 입문 설명에서도 구분합니다. Notification에 응답이 없다는 사실이 모든 하위 계층에 재전송이 없다는 뜻은 아닙니다.
앱이 명령을 받아 데이터베이스에 저장하거나 모터의 동작을 끝냈다는 사실은 별도의 상태입니다. 필요한 경우 자체 메시지에 명령 번호를 넣고, accepted, completed, failed와 같은 업무 상태를 정의합니다. Indication 확인만으로 실제 동작 완료를 표시하면 장애 원인을 놓치기 쉽습니다.
연속 센서값은 최신 상태가 중요할 수 있고, 설정 변경은 결과 확인이 중요할 수 있습니다. 데이터 종류별로 누락 허용 여부와 재시도 규칙을 정하는 것이 출발점입니다.
조각 헤더와 수신 버퍼 규칙 정하기
다음은 설명을 위한 4바이트 헤더입니다. 각 필드는 부호 없는 1바이트이며, 기존 표준 서비스와 호환되는 형식은 아닙니다.
| 위치 | 필드 | 역할 |
|---|---|---|
| 0 | version | 해석할 형식의 버전 |
| 1 | message_id | 현재 연결에서 메시지 구분 |
| 2 | index | 0부터 시작하는 조각 번호 |
| 3 | count | 전체 조각 수 |
수신기는 연결과 메시지 번호를 묶어 버퍼를 관리합니다. 연결이 끊어지면 미완성 버퍼를 버리고, 번호를 재사용하기 전에 이전 메시지가 정리되도록 정합니다. 1바이트 번호는 금방 순환하므로 장기 보관용 이벤트 ID로 쓰지 않습니다.
같은 번호의 조각이 다시 오면 내용이 같은지 검사합니다. 내용이 다르면 해당 메시지는 오류 처리합니다. 전체 조각 수, 최대 조립 크기와 미완성 버퍼 개수에 상한을 두고, 시간 초과된 버퍼를 지워야 합니다. 조각 순서가 뒤섞이는 경우를 시험하면 무선 계층 외에 앱의 큐 처리 오류도 찾을 수 있습니다.
Python으로 분할과 재조립 확인하기
아래 예제는 Python 표준 라이브러리만 사용합니다. 완성된 조각 목록을 한 번에 검증하는 예제로, 실제 BLE 연결이나 비동기 수신 버퍼 구현은 포함하지 않습니다. att_mtu에는 요청한 값이 아닌 연결에서 확인한 값을 전달합니다. 예제는 MTU 23~247, 메시지당 최대 255조각으로 범위를 제한합니다.
import struct
HEADER = struct.Struct("BBBB")
def split_message(text, message_id, att_mtu=23):
if att_mtu not in range(23, 248):
raise ValueError("unsupported MTU for this example")
if message_id not in range(256):
raise ValueError("message ID must fit one byte")
data = text.encode("utf-8")
capacity = att_mtu - 3 - HEADER.size
count = max(1, (len(data) + capacity - 1) // capacity)
if count not in range(1, 256):
raise ValueError("message too large")
return [
HEADER.pack(1, message_id, i, count)
+ data[i * capacity:(i + 1) * capacity]
for i in range(count)
]
def join_message(frames, att_mtu=23):
if att_mtu not in range(23, 248):
raise ValueError("unsupported MTU for this example")
key = None
chunks = {}
for frame in frames:
if len(frame) not in range(HEADER.size, att_mtu - 2):
raise ValueError("invalid frame size")
version, mid, index, count = HEADER.unpack_from(frame)
if version != 1 or count == 0 or index not in range(count):
raise ValueError("invalid header")
current = (version, mid, count)
if key is not None and current != key:
raise ValueError("mixed messages")
key = current
chunk = frame[HEADER.size:]
if index in chunks and chunks[index] != chunk:
raise ValueError("conflicting duplicate")
chunks[index] = chunk
if key is None or len(chunks) != key[2]:
raise ValueError("missing fragments")
return b"".join(chunks[i] for i in range(key[2])).decode("utf-8")
message = "온도=25.1;습도=48.0;상태=정상"
frames = split_message(message, message_id=7)
received = list(reversed(frames)) + [frames[0]]
print(len(frames), max(map(len, frames)))
print(join_message(received) == message)
결과는 첫 줄 3 20, 둘째 줄 True입니다. UTF-8 문자열을 세 조각으로 나누고, 순서를 바꾸고 같은 조각을 한 번 더 넣어도 원문으로 복원합니다. 한 조각이 없거나 같은 헤더에 다른 내용이 들어오면 오류를 냅니다. 입력은 이 예제의 정수·문자열 타입 계약을 따르는 것으로 가정합니다.
실제 장치에서 확인할 항목
MTU를 바꾸기 전후로 메시지 길이와 수신 결과를 기록하고, 연결 중단 후 이전 조각이 새 메시지에 섞이지 않는지 확인합니다. 수신 앱이 느릴 때 송신 큐의 여유 공간과 전송 API의 반환값도 관찰합니다. MTU 증가만으로 처리량이 일정 비율로 늘어난다고 가정하지 않습니다.
최초 연결에서는 형식 버전을 확인하고, 지원하지 않는 버전은 명확한 오류로 처리합니다. 기존 버전의 필드를 다른 의미로 재사용하기보다 새 버전 또는 새 특성을 도입하는 편이 진단하기 쉽습니다. 테스트 도구 사용법은 자료실의 nRF Connect 안내를 참고할 수 있습니다.