MQTT 브로커 · 클라이언트 구현 가이드

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
BL192 연동 시험용 최소 MQTT 3.1.1 브로커 (Python 표준 라이브러리만 사용)

  · 외부 패키지가 필요 없습니다. python3 만 있으면 동작합니다.
  · QoS 0 / 1 의 CONNECT · SUBSCRIBE · PUBLISH · PINGREQ · DISCONNECT 를 처리합니다.
  · BL192 가 올려 보내는 JSON 페이로드를 사람이 읽기 쉬운 형태로 출력합니다.

실행
    python3 mqtt_mini_broker.py            # 0.0.0.0:1883 에서 대기
    python3 mqtt_mini_broker.py 1884       # 포트 지정

BL192 쪽 설정
    Cloud platform → Custom Cloud → Host IP or Domain = 이 PC 의 IP
    Port = 1883, Publish Topic = BL192-PUB, Subscribe Topic = BL192-SUB

주의: 시험·검증용입니다. 인증·TLS·세션 유지·재전송을 구현하지 않았으므로
      운영 환경에는 Mosquitto 나 EMQX 같은 정식 브로커를 쓰십시오.

어큐시스 주식회사 (ACUSYS Co., Ltd.) — BLIIoT 공식 한국 총판
"""
import json
import selectors
import socket
import sys
import time

# ─────────────────────────────────────────────────────────
# MQTT 3.1.1 패킷 타입
CONNECT, CONNACK, PUBLISH, PUBACK = 1, 2, 3, 4
SUBSCRIBE, SUBACK, UNSUBSCRIBE, UNSUBACK = 8, 9, 10, 11
PINGREQ, PINGRESP, DISCONNECT = 12, 13, 14


# ── 가변 길이 정수 (Remaining Length) ─────────────────────
def encode_varint(value: int) -> bytes:
    out = bytearray()
    while True:
        b = value % 128
        value //= 128
        if value > 0:
            b |= 0x80
        out.append(b)
        if value == 0:
            return bytes(out)


def decode_varint(buf: bytes, start: int):
    """(값, 사용한 바이트 수) 반환. 데이터가 모자라면 (None, 0)."""
    value, multiplier, used = 0, 1, 0
    for _ in range(4):
        if start + used >= len(buf):
            return None, 0
        b = buf[start + used]
        used += 1
        value += (b & 0x7F) * multiplier
        if not (b & 0x80):
            return value, used
        multiplier *= 128
    raise ValueError("Remaining Length 필드가 손상되었습니다")


# ── 2바이트 길이 프리픽스 UTF-8 문자열 ────────────────────
def encode_string(s: str) -> bytes:
    raw = s.encode("utf-8")
    return len(raw).to_bytes(2, "big") + raw


def decode_string(buf: bytes, pos: int):
    if pos + 2 > len(buf):
        return None, pos
    n = int.from_bytes(buf[pos:pos + 2], "big")
    pos += 2
    if pos + n > len(buf):
        return None, pos
    return buf[pos:pos + n].decode("utf-8", "replace"), pos + n


# ── 패킷 빌더 ─────────────────────────────────────────────
def build_connack(rc: int = 0) -> bytes:
    return bytes([CONNACK << 4, 2, 0, rc])


def build_suback(packet_id: int, rc: int = 0) -> bytes:
    return bytes([SUBACK << 4, 3, packet_id >> 8, packet_id & 0xFF, rc])


def build_puback(packet_id: int) -> bytes:
    return bytes([PUBACK << 4, 2, packet_id >> 8, packet_id & 0xFF])


def build_pingresp() -> bytes:
    return bytes([PINGRESP << 4, 0])


def build_publish(topic: str, payload: bytes) -> bytes:
    var = encode_string(topic)
    body = var + payload
    return bytes([PUBLISH << 4]) + encode_varint(len(body)) + body


# ── 토픽 필터 매칭 (+ 와 # 지원) ──────────────────────────
def topic_matches(filt: str, topic: str) -> bool:
    f, t = filt.split("/"), topic.split("/")
    for i, part in enumerate(f):
        if part == "#":
            return True
        if i >= len(t):
            return False
        if part != "+" and part != t[i]:
            return False
    return len(f) == len(t)


# ── BL192 페이로드 보기 좋게 출력 ─────────────────────────
def format_payload(raw: bytes) -> str:
    """BL192 기본 데이터 형식(평면 JSON)을 사람이 읽기 쉽게 정리합니다."""
    try:
        obj = json.loads(raw.decode("utf-8"))
    except Exception:
        return raw.decode("utf-8", "replace")
    if not isinstance(obj, dict):
        return raw.decode("utf-8", "replace")

    labels = {"ipv4-address": "IP 주소", "ip": "IP 주소",
              "time": "시간", "timestamp": "시간", "seq": "시퀀스"}
    groups, lines = {}, []
    for k, v in obj.items():
        if k.startswith("REG") and k[3:].isdigit():
            num = int(k[3:])
            groups.setdefault(num // 1000, []).append((num % 1000, v))
        elif k in ("time", "timestamp") and isinstance(v, (int, float)):
            stamp = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(v))
            lines.append(f"    {labels[k]} : {stamp} (epoch {v})")
        else:
            lines.append(f"    {labels.get(k, k)} : {v}")
    for g in sorted(groups):
        cells = " ".join(f"{s:03d}={v}" for s, v in sorted(groups[g]))
        lines.append(f"    REG{g}xxx : {cells}")
    return "\n".join(lines)


def log(msg: str) -> None:
    print(time.strftime("[%H:%M:%S]"), msg, flush=True)


# ── 클라이언트 상태 ───────────────────────────────────────
class Client:
    def __init__(self, sock, addr):
        self.sock, self.addr = sock, addr
        self.buf = bytearray()
        self.client_id = ""
        self.subs = []
        self.connected = False


class Broker:
    def __init__(self, port=1883):
        self.port = port
        self.sel = selectors.DefaultSelector()
        self.clients = {}          # socket -> Client
        self.total_messages = 0

    # 구독자에게 전달
    def broadcast(self, topic: str, payload: bytes):
        pkt = build_publish(topic, payload)
        for c in list(self.clients.values()):
            if c.connected and any(topic_matches(f, topic) for f in c.subs):
                try:
                    c.sock.sendall(pkt)
                except OSError:
                    pass

    def handle_packet(self, c: Client, ptype: int, flags: int, body: bytes) -> bool:
        """반환값 False 면 연결을 끊습니다."""
        if ptype == CONNECT:
            pos = 0
            _proto, pos = decode_string(body, pos)
            pos += 4                       # level(1) + flags(1) + keepalive(2)
            cid, pos = decode_string(body, pos)
            c.client_id = cid or f"anon-{c.addr}"
            c.connected = True
            log(f"CONNECT: {c.client_id} ({c.addr})")
            c.sock.sendall(build_connack(0))

        elif ptype == SUBSCRIBE:
            packet_id = int.from_bytes(body[0:2], "big")
            pos = 2
            while pos < len(body):
                topic, pos = decode_string(body, pos)
                if topic is None or pos >= len(body):
                    break
                pos += 1                   # 요청 QoS 1바이트
                c.subs.append(topic)
                log(f'SUBSCRIBE: {c.client_id} -> "{topic}"')
            c.sock.sendall(build_suback(packet_id, 0))

        elif ptype == PUBLISH:
            qos = (flags >> 1) & 0x03
            pos = 0
            topic, pos = decode_string(body, pos)
            packet_id = 0
            if qos > 0:
                packet_id = int.from_bytes(body[pos:pos + 2], "big")
                pos += 2
            payload = bytes(body[pos:])
            self.total_messages += 1
            log(f"PUBLISH [{topic}] {c.client_id} (QoS {qos}, {len(payload)}바이트)")
            print(format_payload(payload), flush=True)
            self.broadcast(topic, payload)
            if qos == 1:
                c.sock.sendall(build_puback(packet_id))

        elif ptype == PINGREQ:
            c.sock.sendall(build_pingresp())

        elif ptype == DISCONNECT:
            return False

        return True

    def drain(self, c: Client) -> bool:
        """버퍼에 들어 있는 완전한 패킷을 모두 처리합니다."""
        while True:
            if len(c.buf) < 2:
                return True
            remaining, used = decode_varint(c.buf, 1)
            if remaining is None:
                return True                # 길이 필드가 아직 덜 도착
            total = 1 + used + remaining
            if len(c.buf) < total:
                return True                # 본문이 아직 덜 도착
            ptype = (c.buf[0] >> 4) & 0x0F
            flags = c.buf[0] & 0x0F
            body = bytes(c.buf[1 + used:total])
            del c.buf[:total]
            if not self.handle_packet(c, ptype, flags, body):
                return False

    def close(self, sock):
        c = self.clients.pop(sock, None)
        if c:
            log(f"연결 종료: {c.client_id or c.addr}")
        try:
            self.sel.unregister(sock)
        except Exception:
            pass
        sock.close()

    def run(self):
        srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        srv.bind(("0.0.0.0", self.port))
        srv.listen(16)
        srv.setblocking(False)
        self.sel.register(srv, selectors.EVENT_READ, "server")
        log(f"MQTT 브로커 시작 - 0.0.0.0:{self.port} 에서 대기 중... (Ctrl+C 로 종료)")

        try:
            while True:
                for key, _ in self.sel.select(timeout=1.0):
                    if key.data == "server":
                        sock, addr = srv.accept()
                        sock.setblocking(False)
                        a = f"{addr[0]}:{addr[1]}"
                        self.clients[sock] = Client(sock, a)
                        self.sel.register(sock, selectors.EVENT_READ, "client")
                        log(f"새 TCP 연결: {a}")
                        continue

                    sock = key.fileobj
                    c = self.clients.get(sock)
                    if c is None:
                        continue
                    try:
                        data = sock.recv(4096)
                    except OSError:
                        self.close(sock)
                        continue
                    if not data:
                        self.close(sock)
                        continue
                    c.buf.extend(data)
                    try:
                        alive = self.drain(c)
                    except Exception as e:
                        log(f"패킷 처리 오류({c.addr}): {e}")
                        alive = False
                    if not alive:
                        self.close(sock)
        except KeyboardInterrupt:
            log(f"브로커를 종료합니다. 총 수신 메시지: {self.total_messages}")
        finally:
            for sock in list(self.clients):
                self.close(sock)
            srv.close()


if __name__ == "__main__":
    port = int(sys.argv[1]) if len(sys.argv) > 1 else 1883
    Broker(port).run()

파일 내려받기 (신청 후 저장) →

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
BL192 데이터 수신 · 출력 제어 예제 (Python 표준 라이브러리만 사용)

  · BL192 가 올리는 JSON(REG####)을 받아 채널 값으로 풀어 출력합니다.
  · 구독 토픽으로 명령을 보내 디지털 출력(DO)을 제어하는 예도 들어 있습니다.

실행
    python3 bl192_client.py --host 192.168.3.1 --sub BL192-PUB
    python3 bl192_client.py --host 192.168.3.1 --sub BL192-PUB \
                            --pub BL192-SUB --set REG1000=1

레지스터 번호는 BL192 웹 UI 의 I/O Module → IO status 에 나오는
Modbus Address 와 같습니다. 예) DO 1슬롯 1000~1003, DI 2000~2007
→ MQTT 식별자는 그 앞에 REG 를 붙인 REG1000, REG2000 … 입니다.

어큐시스 주식회사 (ACUSYS Co., Ltd.) — BLIIoT 공식 한국 총판
"""
import argparse
import json
import socket
import sys
import time

CONNECT, CONNACK, PUBLISH, PUBACK = 1, 2, 3, 4
SUBSCRIBE, SUBACK, PINGREQ, PINGRESP, DISCONNECT = 8, 9, 12, 13, 14


def encode_varint(value: int) -> bytes:
    out = bytearray()
    while True:
        b = value % 128
        value //= 128
        if value > 0:
            b |= 0x80
        out.append(b)
        if value == 0:
            return bytes(out)


def decode_varint(buf: bytes, start: int):
    value, multiplier, used = 0, 1, 0
    for _ in range(4):
        if start + used >= len(buf):
            return None, 0
        b = buf[start + used]
        used += 1
        value += (b & 0x7F) * multiplier
        if not (b & 0x80):
            return value, used
        multiplier *= 128
    raise ValueError("Remaining Length 필드가 손상되었습니다")


def encode_string(s: str) -> bytes:
    raw = s.encode("utf-8")
    return len(raw).to_bytes(2, "big") + raw


def decode_string(buf: bytes, pos: int):
    n = int.from_bytes(buf[pos:pos + 2], "big")
    pos += 2
    return buf[pos:pos + n].decode("utf-8", "replace"), pos + n


def build_connect(client_id: str, keepalive: int = 60) -> bytes:
    var = encode_string("MQTT") + bytes([4, 0x02]) + keepalive.to_bytes(2, "big")
    body = var + encode_string(client_id)
    return bytes([CONNECT << 4]) + encode_varint(len(body)) + body


def build_subscribe(packet_id: int, topic: str, qos: int = 0) -> bytes:
    body = packet_id.to_bytes(2, "big") + encode_string(topic) + bytes([qos])
    return bytes([(SUBSCRIBE << 4) | 0x02]) + encode_varint(len(body)) + body


def build_publish(topic: str, payload: bytes) -> bytes:
    body = encode_string(topic) + payload
    return bytes([PUBLISH << 4]) + encode_varint(len(body)) + body


def show(payload: bytes) -> None:
    """BL192 기본 데이터 형식(평면 JSON)을 채널 단위로 풀어 출력합니다."""
    try:
        obj = json.loads(payload.decode("utf-8"))
    except Exception:
        print("   (JSON 아님)", payload[:200])
        return

    meta, groups = [], {}
    for k, v in obj.items():
        if k.startswith("REG") and k[3:].isdigit():
            num = int(k[3:])
            groups.setdefault(num // 1000, []).append((num, v))
        elif k in ("time", "timestamp") and isinstance(v, (int, float)):
            meta.append("시간 " + time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(v)))
        else:
            meta.append(f"{k}={v}")

    print("  ", " · ".join(meta) if meta else "(부가 정보 없음)")
    names = {1: "DO", 2: "DI", 3: "AI", 4: "AO"}
    for g in sorted(groups):
        cells = " ".join(f"{a}={v}" for a, v in sorted(groups[g]))
        print(f"   {names.get(g, 'REG' + str(g))} : {cells}")


def main() -> int:
    ap = argparse.ArgumentParser(description="BL192 MQTT 수신 · 제어 예제")
    ap.add_argument("--host", required=True, help="브로커 IP (BL192 설정의 Host IP)")
    ap.add_argument("--port", type=int, default=1883)
    ap.add_argument("--sub", default="BL192-PUB", help="BL192 의 Publish Topic")
    ap.add_argument("--pub", default="", help="BL192 의 Subscribe Topic (제어용)")
    ap.add_argument("--set", default="", help="예: REG1000=1 (DO1 을 ON)")
    ap.add_argument("--id", default="acusys-py", help="MQTT Client ID")
    args = ap.parse_args()

    sock = socket.create_connection((args.host, args.port), timeout=10)
    sock.sendall(build_connect(args.id))

    ack = sock.recv(4)
    if len(ack) < 4 or ack[0] >> 4 != CONNACK or ack[3] != 0:
        print("CONNACK 실패 — 브로커가 실행 중인지 확인하십시오.", file=sys.stderr)
        return 1
    print(f"브로커에 연결되었습니다: {args.host}:{args.port}")

    sock.sendall(build_subscribe(1, args.sub))
    print(f'구독 시작: "{args.sub}"   (Ctrl+C 로 종료)')

    if args.pub and args.set:
        reg, _, val = args.set.partition("=")
        cmd = json.dumps({reg.strip(): int(val)}, separators=(",", ":")).encode()
        sock.sendall(build_publish(args.pub, cmd))
        print(f'제어 명령 전송: [{args.pub}] {cmd.decode()}')

    buf = bytearray()
    last_ping = time.time()
    sock.settimeout(1.0)
    try:
        while True:
            try:
                data = sock.recv(4096)
                if not data:
                    print("브로커가 연결을 끊었습니다.")
                    return 1
                buf.extend(data)
            except socket.timeout:
                pass

            # Keep Alive 유지 — 30초마다 PINGREQ
            if time.time() - last_ping > 30:
                sock.sendall(bytes([PINGREQ << 4, 0]))
                last_ping = time.time()

            while len(buf) >= 2:
                remaining, used = decode_varint(buf, 1)
                if remaining is None:
                    break
                total = 1 + used + remaining
                if len(buf) < total:
                    break
                ptype = (buf[0] >> 4) & 0x0F
                qos = (buf[0] >> 1) & 0x03
                body = bytes(buf[1 + used:total])
                del buf[:total]
                if ptype == PUBLISH:
                    topic, pos = decode_string(body, 0)
                    if qos > 0:
                        pos += 2
                    print(f"\n[{time.strftime('%H:%M:%S')}] {topic}")
                    show(body[pos:])
    except KeyboardInterrupt:
        sock.sendall(bytes([DISCONNECT << 4, 0]))
        print("\n종료합니다.")
    finally:
        sock.close()
    return 0


if __name__ == "__main__":
    sys.exit(main())

파일 내려받기 (신청 후 저장) →

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
BL192 데이터 수신 — paho-mqtt 사용 (운영 환경 권장)

    pip install paho-mqtt
    python3 bl192_paho.py

직접 구현한 소켓 버전과 달리 재연결, Keep Alive, QoS 1·2 재전송,
TLS 를 라이브러리가 처리합니다. 실제 시스템에는 이 방식을 권합니다.

어큐시스 주식회사 (ACUSYS Co., Ltd.) — BLIIoT 공식 한국 총판
"""
import json
import time

import paho.mqtt.client as mqtt

BROKER = "192.168.3.1"     # BL192 의 Host IP or Domain 에 넣은 주소
PORT = 1883
SUB_TOPIC = "BL192-PUB"    # BL192 의 Publish Topic
PUB_TOPIC = "BL192-SUB"    # BL192 의 Subscribe Topic (제어용)

# Modbus 주소 앞자리 → 채널 종류. IO status 화면의 Modbus Address 와 같습니다.
KIND = {1: "DO", 2: "DI", 3: "AI", 4: "AO"}


def on_connect(client, userdata, flags, rc, properties=None):
    if rc == 0:
        print(f"연결됨: {BROKER}:{PORT}")
        client.subscribe(SUB_TOPIC, qos=0)
        print(f'구독: "{SUB_TOPIC}"')
    else:
        print(f"연결 실패 (rc={rc})")


def on_message(client, userdata, msg):
    try:
        data = json.loads(msg.payload.decode("utf-8"))
    except Exception:
        print("JSON 아님:", msg.payload[:200])
        return

    channels, meta = {}, {}
    for key, value in data.items():
        if key.startswith("REG") and key[3:].isdigit():
            channels[int(key[3:])] = value
        else:
            meta[key] = value

    stamp = meta.get("time")
    if isinstance(stamp, (int, float)):
        meta["time"] = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(stamp))

    print(f"\n[{msg.topic}] " + " · ".join(f"{k}={v}" for k, v in meta.items()))
    for addr in sorted(channels):
        kind = KIND.get(addr // 1000, "REG")
        print(f"   {kind} {addr} = {channels[addr]}")

    # 예) DI1 이 ON 이면 DO1 을 켜는 간단한 연동
    # if channels.get(2000) == 1:
    #     client.publish(PUB_TOPIC, json.dumps({"REG1000": 1}))


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="acusys-paho")
client.on_connect = on_connect
client.on_message = on_message
client.connect(BROKER, PORT, keepalive=60)
client.loop_forever()      # 연결이 끊기면 자동으로 재연결합니다

파일 내려받기 (신청 후 저장) →

/* ============================================================
   bl192_client.c — BL192 MQTT 수신 예제 (C, 외부 라이브러리 없음)

   MQTT 3.1.1 의 CONNECT / SUBSCRIBE / PUBLISH / PINGREQ 만 직접 구현해
   BL192 가 올려 보내는 JSON 페이로드를 받아 출력합니다.

   빌드
     Windows : cl /nologo /utf-8 bl192_client.c ws2_32.lib
     Linux   : gcc -O2 -o bl192_client bl192_client.c

   실행
     bl192_client <브로커IP> [포트] [구독토픽]
     예) bl192_client 192.168.3.1 1883 BL192-PUB

   레지스터 번호는 BL192 웹 UI 의 I/O Module → IO status 에 나오는
   Modbus Address 와 같습니다. MQTT 식별자는 앞에 REG 가 붙습니다.

   주의: 시험·학습용 최소 구현입니다. QoS 1·2 의 재전송, 세션 유지,
         TLS, 인증은 구현하지 않았습니다.

   어큐시스 주식회사 (ACUSYS Co., Ltd.) — BLIIoT 공식 한국 총판
   ============================================================ */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>

#ifdef _WIN32
  #define WIN32_LEAN_AND_MEAN
  #include <winsock2.h>
  #include <ws2tcpip.h>
  #pragma comment(lib, "ws2_32.lib")
  typedef SOCKET sock_t;
  #define CLOSESOCK closesocket
#else
  #include <sys/socket.h>
  #include <netinet/in.h>
  #include <arpa/inet.h>
  #include <unistd.h>
  #include <netdb.h>
  typedef int sock_t;
  #define INVALID_SOCKET (-1)
  #define CLOSESOCK close
#endif

/* ---- MQTT 패킷 타입 ---- */
enum { CONNECT = 1, CONNACK = 2, PUBLISH = 3, PUBACK = 4,
       SUBSCRIBE = 8, SUBACK = 9, PINGREQ = 12, PINGRESP = 13, DISCONNECT = 14 };

/* ---- 가변 길이 정수 (Remaining Length) ----
   MQTT 는 본문 길이를 1~4바이트 가변 정수로 적습니다.
   각 바이트의 하위 7비트가 값, 최상위 비트가 "다음 바이트가 더 있음" 표시입니다. */
static int encode_varint(unsigned char *out, unsigned int value)
{
    int n = 0;
    do {
        unsigned char b = (unsigned char)(value % 128);
        value /= 128;
        if (value > 0) b |= 0x80;
        out[n++] = b;
    } while (value > 0);
    return n;                    /* 사용한 바이트 수 */
}

/* 반환값 0=성공, -1=데이터 부족, -2=손상 */
static int decode_varint(const unsigned char *buf, size_t avail,
                         unsigned int *value, int *used)
{
    unsigned int multiplier = 1;
    *value = 0; *used = 0;
    for (int i = 0; i < 4; ++i) {
        if ((size_t)*used >= avail) return -1;
        unsigned char b = buf[(*used)++];
        *value += (unsigned int)(b & 0x7F) * multiplier;
        if (!(b & 0x80)) return 0;
        multiplier *= 128;
    }
    return -2;
}

/* ---- 2바이트 길이 프리픽스 UTF-8 문자열 ---- */
static int encode_string(unsigned char *out, const char *s)
{
    size_t len = strlen(s);
    out[0] = (unsigned char)(len >> 8);
    out[1] = (unsigned char)(len & 0xFF);
    memcpy(out + 2, s, len);
    return (int)(len + 2);
}

/* ---- 패킷 송신 ---- */
static int send_all(sock_t s, const unsigned char *data, int len)
{
    int sent = 0;
    while (sent < len) {
        int n = send(s, (const char *)data + sent, len - sent, 0);
        if (n <= 0) return -1;
        sent += n;
    }
    return 0;
}

static int send_connect(sock_t s, const char *client_id)
{
    unsigned char var[64], payload[128], pkt[256];
    int vn = 0, pn = 0, n = 0;

    vn += encode_string(var + vn, "MQTT");
    var[vn++] = 4;              /* 프로토콜 레벨 4 = MQTT 3.1.1 */
    var[vn++] = 0x02;           /* Clean Session, 인증 없음 */
    var[vn++] = 0;              /* Keep Alive 상위 바이트 */
    var[vn++] = 60;             /* Keep Alive 60초 */
    pn += encode_string(payload + pn, client_id);

    pkt[n++] = (unsigned char)(CONNECT << 4);
    n += encode_varint(pkt + n, (unsigned int)(vn + pn));
    memcpy(pkt + n, var, vn); n += vn;
    memcpy(pkt + n, payload, pn); n += pn;
    return send_all(s, pkt, n);
}

static int send_subscribe(sock_t s, unsigned short packet_id, const char *topic)
{
    unsigned char body[300], pkt[320];
    int bn = 0, n = 0;

    body[bn++] = (unsigned char)(packet_id >> 8);
    body[bn++] = (unsigned char)(packet_id & 0xFF);
    bn += encode_string(body + bn, topic);
    body[bn++] = 0;             /* 요청 QoS 0 */

    pkt[n++] = (unsigned char)((SUBSCRIBE << 4) | 0x02);  /* 예약 비트 0x02 필수 */
    n += encode_varint(pkt + n, (unsigned int)bn);
    memcpy(pkt + n, body, bn); n += bn;
    return send_all(s, pkt, n);
}

static int send_publish(sock_t s, const char *topic, const char *payload)
{
    unsigned char pkt[1024];
    int n = 0, bn;
    unsigned char body[1000];
    bn = encode_string(body, topic);
    size_t plen = strlen(payload);
    memcpy(body + bn, payload, plen); bn += (int)plen;

    pkt[n++] = (unsigned char)(PUBLISH << 4);
    n += encode_varint(pkt + n, (unsigned int)bn);
    memcpy(pkt + n, body, bn); n += bn;
    return send_all(s, pkt, n);
}

/* ---- 페이로드 출력 ----
   평면 JSON 에서 "REGxxxx": 숫자 쌍만 뽑아 줄 단위로 정리합니다.
   (제대로 된 JSON 파서를 쓰실 수 있으면 그 편이 좋습니다) */
static void print_payload(const char *buf, int len)
{
    printf("    ");
    int shown = 0;
    for (int i = 0; i + 4 < len; ++i) {
        if (buf[i] != '"' || strncmp(buf + i + 1, "REG", 3) != 0) continue;
        int j = i + 4;
        char reg[16]; int rn = 0;
        while (j < len && buf[j] >= '0' && buf[j] <= '9' && rn < 15) reg[rn++] = buf[j++];
        reg[rn] = 0;
        while (j < len && (buf[j] == '"' || buf[j] == ':' || buf[j] == ' ')) j++;
        char val[24]; int vn = 0;
        while (j < len && buf[j] != ',' && buf[j] != '}' && vn < 23) val[vn++] = buf[j++];
        val[vn] = 0;
        if (rn == 0) continue;
        if (shown && shown % 8 == 0) printf("\n    ");
        printf("REG%s=%s ", reg, val);
        shown++;
        i = j;
    }
    if (!shown) printf("%.*s", len > 300 ? 300 : len, buf);
    printf("\n");
}

int main(int argc, char **argv)
{
    const char *host  = (argc > 1) ? argv[1] : "192.168.3.1";
    int         port  = (argc > 2) ? atoi(argv[2]) : 1883;
    const char *topic = (argc > 3) ? argv[3] : "BL192-PUB";

#ifdef _WIN32
    WSADATA wsa;
    if (WSAStartup(MAKEWORD(2, 2), &wsa) != 0) { fprintf(stderr, "WSAStartup 실패\n"); return 1; }
#endif

    sock_t s = socket(AF_INET, SOCK_STREAM, 0);
    if (s == INVALID_SOCKET) { fprintf(stderr, "소켓 생성 실패\n"); return 1; }

    struct sockaddr_in addr;
    memset(&addr, 0, sizeof(addr));
    addr.sin_family = AF_INET;
    addr.sin_port   = htons((unsigned short)port);
    if (inet_pton(AF_INET, host, &addr.sin_addr) != 1) {
        fprintf(stderr, "IP 주소 형식이 올바르지 않습니다: %s\n", host);
        return 1;
    }
    if (connect(s, (struct sockaddr *)&addr, sizeof(addr)) != 0) {
        fprintf(stderr, "브로커 연결 실패: %s:%d\n", host, port);
        return 1;
    }

    if (send_connect(s, "acusys-c") != 0) { fprintf(stderr, "CONNECT 송신 실패\n"); return 1; }

    unsigned char ack[8];
    int n = recv(s, (char *)ack, 4, 0);
    if (n < 4 || (ack[0] >> 4) != CONNACK || ack[3] != 0) {
        fprintf(stderr, "CONNACK 실패 — 브로커가 실행 중인지 확인하십시오.\n");
        CLOSESOCK(s);
        return 1;
    }
    printf("브로커에 연결되었습니다: %s:%d\n", host, port);

    send_subscribe(s, 1, topic);
    printf("구독 시작: \"%s\"   (Ctrl+C 로 종료)\n", topic);

    /* 예: 디지털 출력 1번을 ON 하려면 구독 토픽으로 아래를 보냅니다.
       send_publish(s, "BL192-SUB", "{\"REG1000\":1}"); */

    static unsigned char buf[65536];
    size_t have = 0;
    time_t last_ping = time(NULL);

    for (;;) {
        int r = recv(s, (char *)buf + have, (int)(sizeof(buf) - have), 0);
        if (r <= 0) { printf("연결이 끊어졌습니다.\n"); break; }
        have += (size_t)r;

        for (;;) {
            if (have < 2) break;
            unsigned int remaining; int used;
            int rc = decode_varint(buf + 1, have - 1, &remaining, &used);
            if (rc == -1) break;
            if (rc == -2) { have = 0; break; }
            size_t total = 1 + (size_t)used + remaining;
            if (have < total) break;

            unsigned char type = (unsigned char)((buf[0] >> 4) & 0x0F);
            unsigned char qos  = (unsigned char)((buf[0] >> 1) & 0x03);
            const unsigned char *body = buf + 1 + used;

            if (type == PUBLISH) {
                unsigned int tlen = ((unsigned int)body[0] << 8) | body[1];
                const char *tp = (const char *)body + 2;
                size_t pos = 2 + tlen;
                if (qos > 0) pos += 2;          /* Packet Identifier */
                time_t t = time(NULL);
                struct tm lt;
#ifdef _WIN32
                localtime_s(&lt, &t);
#else
                localtime_r(&t, &lt);
#endif
                printf("[%02d:%02d:%02d] %.*s\n", lt.tm_hour, lt.tm_min, lt.tm_sec,
                       (int)tlen, tp);
                print_payload((const char *)body + pos, (int)(remaining - pos));
                fflush(stdout);
            }

            memmove(buf, buf + total, have - total);
            have -= total;
        }

        /* Keep Alive 유지 — 30초마다 PINGREQ */
        if (time(NULL) - last_ping > 30) {
            unsigned char ping[2] = { (unsigned char)(PINGREQ << 4), 0 };
            send_all(s, ping, 2);
            last_ping = time(NULL);
        }
    }

    CLOSESOCK(s);
#ifdef _WIN32
    WSACleanup();
#endif
    return 0;
}

파일 내려받기 (신청 후 저장) →

// ============================================================
// Bl192Client.cs — BL192 MQTT 수신 예제 (C#, 외부 패키지 없음)
//
// MQTT 3.1.1 의 CONNECT / SUBSCRIBE / PUBLISH / PINGREQ 만 직접 구현해
// BL192 가 올려 보내는 JSON 페이로드를 받아 출력합니다.
//
// 빌드와 실행
//   .NET      : dotnet new console -o Bl192 → Program.cs 를 이 파일로 교체
//               dotnet run -- 192.168.3.1 1883 BL192-PUB
//   Windows   : csc Bl192Client.cs   (개발자 명령 프롬프트)
//               Bl192Client.exe 192.168.3.1 1883 BL192-PUB
//   Mono      : mcs Bl192Client.cs && mono Bl192Client.exe 192.168.3.1 1883 BL192-PUB
//
// 레지스터 번호는 BL192 웹 UI 의 I/O Module → IO status 에 나오는
// Modbus Address 와 같습니다. MQTT 식별자는 앞에 REG 가 붙습니다.
//
// 주의: 시험·학습용 최소 구현입니다. QoS 1·2 의 재전송, 세션 유지, TLS,
//       인증은 구현하지 않았습니다. 운영에는 MQTTnet 을 권합니다.
//
// 어큐시스 주식회사 (ACUSYS Co., Ltd.) — BLIIoT 공식 한국 총판
// ============================================================
using System;
using System.Collections.Generic;
using System.Net.Sockets;
using System.Text;

namespace Acusys.Bliiot
{
    internal static class Mqtt
    {
        public const byte Connect = 1, Connack = 2, Publish = 3, Puback = 4;
        public const byte Subscribe = 8, Suback = 9, Pingreq = 12, Pingresp = 13, Disconnect = 14;

        // ---- 가변 길이 정수 (Remaining Length) ----
        // 하위 7비트가 값, 최상위 비트가 1이면 다음 바이트가 이어집니다.
        public static void EncodeVarInt(List<byte> outBuf, int value)
        {
            do
            {
                byte b = (byte)(value % 128);
                value /= 128;
                if (value > 0) b |= 0x80;
                outBuf.Add(b);
            } while (value > 0);
        }

        // 반환값 0=성공, -1=데이터 부족, -2=손상
        public static int DecodeVarInt(byte[] buf, int start, int avail, out int value, out int used)
        {
            value = 0; used = 0;
            int multiplier = 1;
            for (int i = 0; i < 4; i++)
            {
                if (used >= avail) return -1;
                byte b = buf[start + used];
                used++;
                value += (b & 0x7F) * multiplier;
                if ((b & 0x80) == 0) return 0;
                multiplier *= 128;
            }
            return -2;
        }

        // ---- 2바이트 길이 프리픽스 UTF-8 문자열 ----
        public static void EncodeString(List<byte> outBuf, string s)
        {
            byte[] raw = Encoding.UTF8.GetBytes(s);
            outBuf.Add((byte)(raw.Length >> 8));
            outBuf.Add((byte)(raw.Length & 0xFF));
            outBuf.AddRange(raw);
        }

        // ---- 패킷 빌더 ----
        public static byte[] BuildConnect(string clientId, ushort keepAlive)
        {
            var vari = new List<byte>();
            EncodeString(vari, "MQTT");
            vari.Add(4);                      // 프로토콜 레벨 4 = MQTT 3.1.1
            vari.Add(0x02);                   // Clean Session, 인증 없음
            vari.Add((byte)(keepAlive >> 8));
            vari.Add((byte)(keepAlive & 0xFF));

            var payload = new List<byte>();
            EncodeString(payload, clientId);

            var pkt = new List<byte> { (byte)(Connect << 4) };
            EncodeVarInt(pkt, vari.Count + payload.Count);
            pkt.AddRange(vari);
            pkt.AddRange(payload);
            return pkt.ToArray();
        }

        public static byte[] BuildSubscribe(ushort packetId, string topic, byte qos)
        {
            var body = new List<byte> { (byte)(packetId >> 8), (byte)(packetId & 0xFF) };
            EncodeString(body, topic);
            body.Add(qos);

            var pkt = new List<byte> { (byte)((Subscribe << 4) | 0x02) }; // 예약 비트 0x02 필수
            EncodeVarInt(pkt, body.Count);
            pkt.AddRange(body);
            return pkt.ToArray();
        }

        public static byte[] BuildPublish(string topic, string payload)
        {
            var body = new List<byte>();
            EncodeString(body, topic);
            body.AddRange(Encoding.UTF8.GetBytes(payload));

            var pkt = new List<byte> { (byte)(Publish << 4) };
            EncodeVarInt(pkt, body.Count);
            pkt.AddRange(body);
            return pkt.ToArray();
        }

        public static byte[] BuildPingReq() { return new byte[] { Pingreq << 4, 0 }; }
        public static byte[] BuildDisconnect() { return new byte[] { Disconnect << 4, 0 }; }
    }

    internal static class Program
    {
        // Modbus 주소 앞자리 → 채널 종류
        private static readonly Dictionary<int, string> Kind = new Dictionary<int, string>
        {
            { 1, "DO" }, { 2, "DI" }, { 3, "AI" }, { 4, "AO" }
        };

        // 평면 JSON 에서 "REGxxxx": 값 쌍만 뽑아 출력합니다.
        // (System.Text.Json 이나 Newtonsoft.Json 을 쓰실 수 있으면 그 편이 안전합니다)
        private static void PrintPayload(string json)
        {
            var regs = new SortedDictionary<int, string>();
            var others = new List<string>();
            int i = 0;
            while (i < json.Length)
            {
                int q = json.IndexOf('"', i);
                if (q < 0) break;
                int q2 = json.IndexOf('"', q + 1);
                if (q2 < 0) break;
                string key = json.Substring(q + 1, q2 - q - 1);

                int colon = json.IndexOf(':', q2);
                if (colon < 0) break;
                int v = colon + 1;
                while (v < json.Length && json[v] == ' ') v++;
                int end = v;
                if (end < json.Length && json[end] == '"')
                {
                    end = json.IndexOf('"', end + 1);
                    end = (end < 0) ? json.Length : end + 1;
                }
                else
                {
                    while (end < json.Length && json[end] != ',' && json[end] != '}') end++;
                }
                string val = json.Substring(v, Math.Max(0, end - v)).Trim().Trim('"');

                int addr;
                if (key.StartsWith("REG", StringComparison.Ordinal) &&
                    int.TryParse(key.Substring(3), out addr))
                {
                    regs[addr] = val;
                }
                else if (key == "time" || key == "timestamp")
                {
                    long epoch;
                    if (long.TryParse(val, out epoch))
                    {
                        DateTime t = new DateTime(1970, 1, 1, 0, 0, 0, DateTimeKind.Utc)
                            .AddSeconds(epoch).ToLocalTime();
                        others.Add("시간=" + t.ToString("yyyy-MM-dd HH:mm:ss"));
                    }
                    else others.Add(key + "=" + val);
                }
                else
                {
                    others.Add(key + "=" + val);
                }
                i = end;
            }

            if (others.Count > 0) Console.WriteLine("   " + string.Join(" · ", others.ToArray()));

            string line = "";
            int shown = 0, lastGroup = -1;
            foreach (var kv in regs)
            {
                int group = kv.Key / 1000;
                if (group != lastGroup)
                {
                    if (line.Length > 0) Console.WriteLine(line);
                    string name;
                    line = "   " + (Kind.TryGetValue(group, out name) ? name : "REG") + " : ";
                    lastGroup = group; shown = 0;
                }
                if (shown > 0 && shown % 8 == 0)
                {
                    Console.WriteLine(line);
                    line = "        ";
                }
                line += kv.Key + "=" + kv.Value + " ";
                shown++;
            }
            if (line.Length > 0) Console.WriteLine(line);
        }

        private static int Main(string[] args)
        {
            // 콘솔에 한글이 깨지지 않도록 출력 인코딩을 UTF-8 로 맞춥니다.
            // (Windows 콘솔 기본 코드페이지에서는 한글이 ? 로 나올 수 있습니다)
            try { Console.OutputEncoding = Encoding.UTF8; } catch { }

            string host = args.Length > 0 ? args[0] : "192.168.3.1";
            int port = args.Length > 1 ? int.Parse(args[1]) : 1883;
            string topic = args.Length > 2 ? args[2] : "BL192-PUB";

            TcpClient tcp;
            try
            {
                tcp = new TcpClient();
                tcp.Connect(host, port);
            }
            catch (Exception e)
            {
                Console.Error.WriteLine("브로커 연결 실패: " + host + ":" + port + " — " + e.Message);
                return 1;
            }

            using (tcp)
            using (NetworkStream net = tcp.GetStream())
            {
                byte[] connect = Mqtt.BuildConnect("acusys-cs", 60);
                net.Write(connect, 0, connect.Length);

                var ack = new byte[4];
                int got = net.Read(ack, 0, 4);
                if (got < 4 || (ack[0] >> 4) != Mqtt.Connack || ack[3] != 0)
                {
                    Console.Error.WriteLine("CONNACK 실패 — 브로커가 실행 중인지 확인하십시오.");
                    return 1;
                }
                Console.WriteLine("브로커에 연결되었습니다: " + host + ":" + port);

                byte[] sub = Mqtt.BuildSubscribe(1, topic, 0);
                net.Write(sub, 0, sub.Length);
                Console.WriteLine("구독 시작: \"" + topic + "\"   (Ctrl+C 로 종료)");

                // 예: 디지털 출력 1번을 ON 하려면 구독 토픽으로 아래를 보냅니다.
                // byte[] cmd = Mqtt.BuildPublish("BL192-SUB", "{\"REG1000\":1}");
                // net.Write(cmd, 0, cmd.Length);

                var buf = new byte[65536];
                int have = 0;
                var chunk = new byte[4096];
                DateTime lastPing = DateTime.UtcNow;
                net.ReadTimeout = 1000;

                while (true)
                {
                    int n = 0;
                    try { n = net.Read(chunk, 0, chunk.Length); }
                    catch (System.IO.IOException) { n = -1; }   // 읽기 제한 시간 — 정상

                    if (n == 0) { Console.WriteLine("브로커가 연결을 끊었습니다."); break; }
                    if (n > 0)
                    {
                        if (have + n > buf.Length) have = 0;    // 넘치면 버리고 다시 맞춥니다
                        Buffer.BlockCopy(chunk, 0, buf, have, n);
                        have += n;
                    }

                    // TCP 는 패킷 경계를 지켜 주지 않습니다.
                    // 완전한 패킷이 만들어졌을 때만 꺼내 씁니다.
                    while (have >= 2)
                    {
                        int remaining, used;
                        int rc = Mqtt.DecodeVarInt(buf, 1, have - 1, out remaining, out used);
                        if (rc == -1) break;
                        if (rc == -2) { have = 0; break; }
                        int total = 1 + used + remaining;
                        if (have < total) break;

                        byte type = (byte)((buf[0] >> 4) & 0x0F);
                        byte qos = (byte)((buf[0] >> 1) & 0x03);
                        int bodyAt = 1 + used;

                        if (type == Mqtt.Publish)
                        {
                            int tlen = (buf[bodyAt] << 8) | buf[bodyAt + 1];
                            string tp = Encoding.UTF8.GetString(buf, bodyAt + 2, tlen);
                            int pos = 2 + tlen;
                            if (qos > 0) pos += 2;              // Packet Identifier
                            string json = Encoding.UTF8.GetString(buf, bodyAt + pos, remaining - pos);
                            Console.WriteLine();
                            Console.WriteLine("[" + DateTime.Now.ToString("HH:mm:ss") + "] " + tp);
                            PrintPayload(json);
                        }

                        Buffer.BlockCopy(buf, total, buf, 0, have - total);
                        have -= total;
                    }

                    // Keep Alive 유지 — 30초마다 PINGREQ
                    if ((DateTime.UtcNow - lastPing).TotalSeconds > 30)
                    {
                        byte[] ping = Mqtt.BuildPingReq();
                        net.Write(ping, 0, ping.Length);
                        lastPing = DateTime.UtcNow;
                    }
                }
            }
            return 0;
        }
    }
}

파일 내려받기 (신청 후 저장) →