MQTT ile Dosya ve Görsel Gönderme: Parçalama, Base64 ve Birleştirme

Evet, MQTT ile dosya ve görsel gönderebilirsiniz: payload yalnızca baytlardan oluşur, dolayısıyla bir JPEG ya da log dosyası diskte nasıl duruyorsa öyle publish edilebilir. İşin zor kısmı ayrıntılarda: broker'ın boyut limitinin altında kalmak, büyük dosyaları parçalara (chunk) bölmek, onları doğru sırayla yeniden birleştirmek ve sonucun orijinalle birebir aynı olduğunu kanıtlamak.

Bu rehber, ne zaman ham binary göndermeniz ve ne zaman base64 kullanmanız gerektiğini, manifest içeren basit bir parçalama protokolünü ve birkaç dakikada test edebileceğiniz çalışan bir Python gönderici ile alıcıyı anlatıyor.

Binary mi, base64 mü?

MQTT payload'un içinde ne olduğuyla ilgilenmez. Bir görseli publish etmeden önce kodlamanız gerekmez; yaygın tüm istemci kütüphaneleri ham baytları kabul eder (Python'da bytes, MQTT.js'te Buffer / Uint8Array, Java ve .NET'te byte[]).

Base64'ü yalnızca baytların bir metin belgesinin içinde yaşaması gerektiğinde tercih edersiniz:

600 KB'lık bir fotoğraf base64 ile kodlandığında yaklaşık 800 KB olur ve alıcı onu kullanabilmek için önce JSON'u parse edip decode etmek zorundadır. İki ucu da siz kontrol ediyorsanız binary gönderin.

MQTT 5 ile binary payload'u sarmak yerine ona metadata ekleyebilirsiniz: Content Type property'sini image/jpeg yapın ve dosya adını bir user property içine koyun. Payload Format Indicator değerini 0'da (belirtilmemiş baytlar) bırakın; 1 yapmak broker'a ve alıcıya payload'un UTF-8 metin olduğunu söyler, oysa bir JPEG metin değildir.

Boyut limitlerinizi bilin

MQTT spesifikasyonu 268.435.455 bayta, yani paket başına yaklaşık 256 MB'a kadar remaining length'e izin verir. Pratikte kimse buna yaklaşmaz, çünkü broker'lar ve servisler kendi maksimumlarını uygular:

Limitler sürümden sürüme ve plandan plana değişir; tabloyu bir başlangıç noktası olarak görün ve kendi broker'ınızda doğrulayın. MQTT 5'te sunucu limitini CONNACK içindeki Maximum Packet Size property'siyle bildirir; bunu aşan bir publish genellikle 0x95 (Packet too large) reason code'lu bir disconnect ile sonuçlanır. MQTT 3.1.1 broker'ları ise çoğunlukla bağlantıyı sessizce kapatır; bu da rastgele bir kopma gibi görünür. Packet too large rehberi bu belirtileri adım adım ele alır, paket boyutu hesaplayıcı ise topic ve property'lerin payload boyutuna ne kadar eklediğini gösterir.

Büyük mesajlar daha az belirgin şekillerde de zarar verir: aktarım sürerken TCP bağlantısını tıkarlar (keep-alive ping'leri ve diğer mesajlar arkalarında bekler), broker onları kuyruğunda tutan her offline oturum için bellekte saklamak zorundadır ve kayıplı bir hatta başarısız olan tek bir 5 MB'lık publish baştan sona yeniden gönderilir.

Basit bir parçalama protokolü

Rahat bir tekil mesajdan büyük her şeyi (iyi bir kural 64–256 KB) parçalara bölün. Protokolü sıkıcı tutun:

1. Manifest, testmqtt/files/<device>/<file_id>/meta üzerinde: dosya adı, toplam boyut, parça boyutu, parça sayısı ve SHA-256 hash'i içeren JSON.
2. Parçalar, testmqtt/files/<device>/<file_id>/chunk/<index> üzerinde: kodlanmamış ham baytlar.
3. Alıcı parçaları index'e göre toplar; hepsi geldiğinde birleştirir ve hash'i karşılaştırır.

Neden index topic'te? MQTT sıralamayı yalnızca tek bir publisher'dan gelen aynı topic için garanti eder; farklı topic'lerdeki parçalar herhangi bir sırayla teslim edilebilir. QoS 1 ile bir parça iki kez de gelebilir; parçaları index anahtarlı bir map'te saklamak tekrarları zararsız kılar. Index'i topic'e koymak payload'u saf dosya verisi olarak tutar; dosya başına tek bir topic tercih ederseniz her parçanın başına 4 baytlık binary bir header da ekleyebilirsiniz.

Parçalar için QoS 1 kullanın. QoS 0 kötü bir ağda er ya da geç birini kaybeder ve tek parçası eksik bir dosya işe yaramaz. QoS 2 parça başına ek bir gidiş-dönüş getirir ama pek fayda sağlamaz, çünkü alıcı tekrarları zaten ayıklar.

Gönderici (Python)

İstemciyi pip install "paho-mqtt>=2" ile kurun. Gönderici bir sonraki parçayı göndermeden önce her PUBACK'i bekler; bu doğal bir akış kontrolü sağlar ve broker'ın inflight penceresinin taşmasını önler:

import hashlib, json, math, os, sys, time, uuid
import paho.mqtt.client as mqtt

HOST, PORT = "broker.connectmq.com", 1883
CHUNK_SIZE = 32 * 1024                      # mesaj başına 32 KB
path = sys.argv[1]

with open(path, "rb") as f:
    data = f.read()

file_id = uuid.uuid4().hex[:12]
base = f"testmqtt/files/demo/{file_id}"
total = math.ceil(len(data) / CHUNK_SIZE)

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"sender-{file_id}")
# client.username_pw_set("username", "password")
client.connect(HOST, PORT)
client.loop_start()

manifest = {
    "name": os.path.basename(path),
    "size": len(data),
    "chunk_size": CHUNK_SIZE,
    "chunks": total,
    "sha256": hashlib.sha256(data).hexdigest(),
}
client.publish(f"{base}/meta", json.dumps(manifest), qos=1).wait_for_publish(10)

for i in range(total):
    chunk = data[i * CHUNK_SIZE:(i + 1) * CHUNK_SIZE]
    info = client.publish(f"{base}/chunk/{i}", chunk, qos=1)
    info.wait_for_publish(10)               # PUBACK'i bekle
    time.sleep(0.02)                        # paylaşımlı limitlerin çok altında kal

print(f"{manifest['name']} {total} parça olarak gönderildi, id {file_id}")
client.loop_stop()
client.disconnect()

Alıcı (Python)

Alıcı cihazın tüm dosyalarına subscribe olur, gelen parçaları saklar ve hash eşleştiğinde dosyayı yazar:

import hashlib, json
import paho.mqtt.client as mqtt

HOST, PORT = "broker.connectmq.com", 1883
transfers = {}                              # file_id -> {"meta": ..., "chunks": {}}

def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe("testmqtt/files/demo/+/#", qos=1)

def on_message(client, userdata, msg):
    parts = msg.topic.split("/")            # testmqtt/files/demo/<id>/meta|chunk/<n>
    file_id, kind = parts[3], parts[4]
    t = transfers.setdefault(file_id, {"meta": None, "chunks": {}})

    if kind == "meta":
        t["meta"] = json.loads(msg.payload)
    elif kind == "chunk":
        t["chunks"][int(parts[5])] = msg.payload   # tekrarlar sadece üzerine yazar

    meta = t["meta"]
    if meta and len(t["chunks"]) == meta["chunks"]:
        blob = b"".join(t["chunks"][i] for i in range(meta["chunks"]))
        if hashlib.sha256(blob).hexdigest() == meta["sha256"]:
            with open("received-" + meta["name"], "wb") as f:
                f.write(blob)
            print(f"OK  {meta['name']} ({len(blob)} bayt)")
        else:
            print(f"HATA {meta['name']}: hash eşleşmiyor")
        del transfers[file_id]

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
client.on_connect = on_connect
client.on_message = on_message
client.connect(HOST, PORT)
client.loop_forever()

Parçalar manifest'ten önce gelebilir; bu yüzden tamamlanma kontrolü yalnızca manifest'ten sonra değil, her mesajdan sonra çalışır. Production'da tamamlanmamış transferleri bir iki dakika sonra silen bir timeout ekleyin; aksi halde kaybolan bir manifest bellek sızıntısına yol açar.

Transferi test etmek

Alıcıyı bir terminalde, göndericiyi başka bir terminalde çalıştırın:

python receive_files.py
# ikinci terminal
python send_file.py photo.jpg
sha256sum photo.jpg received-photo.jpg     # macOS: shasum -a 256

İki hash de aynı olmalıdır. Sonra işi zorlaştırın: boyutu parça boyutunun tam katı olan bir dosya gönderin (klasik off-by-one hatası), boş bir dosya gönderin ve transferin ortasında ağı kesip alıcının çöp veri yazmak yerine eksik dosya bildirdiğini doğrulayın. Test sürerken online MQTT istemcisinde testmqtt/files/demo/+/meta topic'ine subscribe olup manifest'lerin gelişini izleyin.

Tek mesaja sığan küçük dosyalar için Mosquitto'nun komut satırı istemcileri yeterlidir. -f bir dosyayı payload olarak publish eder; -N ise mosquitto_sub'ın binary veriyi bozacak bir satır sonu eklemesini engeller:

mosquitto_sub -h broker.connectmq.com -t testmqtt/files/cli/snapshot -C 1 -N > out.jpg &
mosquitto_pub -h broker.connectmq.com -t testmqtt/files/cli/snapshot -q 1 -f snapshot.jpg

Daha fazla komut satırı kalıbı için mosquitto_pub ve mosquitto_sub örneklerine bakın. Broker'ınızın hangi boyuttan itibaren mesajları reddettiğini bulmak için büyük payload testini izleyin.

Tarayıcıda görsel göstermek

Bir dashboard, MQTT.js 5 ile güvenli WebSocket üzerinden gelen bir kamera görüntüsünü doğrudan gösterebilir. Payload bayt olarak gelir; onu bir Blob içine sarmanız yeterlidir:

import mqtt from "mqtt";

const client = mqtt.connect("wss://broker.connectmq.com:8084/mqtt");
const img = document.querySelector("#snapshot");

client.on("connect", () => client.subscribe("testmqtt/files/cli/snapshot"));
client.on("message", (topic, payload) => {
  URL.revokeObjectURL(img.src);
  img.src = URL.createObjectURL(new Blob([payload], { type: "image/jpeg" }));
});

Tarayıcıya giden görselleri küçük tutun (onlarca KB): subscribe olmuş her sekme her kareyi indirir.

Dosyalar için MQTT ne zaman kullanılmamalı

MQTT bir mesajlaşma protokolüdür, dosya sunucusu değildir. Firmware imajları, video klipler veya birkaç megabaytı aşan her şey için daha iyi kalıp, dosyayı bir object storage'a yükleyip yalnızca URL, boyut ve hash içeren küçük bir mesaj publish etmektir. Cihaz dosyayı kaldığı yerden devam etme ve CDN önbelleği destekleyen HTTPS ile indirir; MQTT ise en iyi yaptığı şeyi yapar: cihaza indirilecek bir şey olduğunu haber verir. MQTT ile OTA firmware güncellemeleri tam olarak bu ayrımı kullanır, payload tasarımı en iyi uygulamaları ise bildirim mesajını nasıl hafif tutacağınızı anlatır.

Dosya transferlerini online istemcide izleyin

MQTT ile dosya göndermek dört kurala indirgenir: ham bayt gönderin, broker'ın limitinin altında kalın, parçalarınızı numaralandırın ve sonda bir hash doğrulayın. Önce gidiş-dönüş testini kurun; tekrarlar, sırası karışmış parçalar ve kaybolan bir manifest gibi uç durumlar kolayca görünür ve düzeltilebilir hale gelir.

Sıkça sorulan sorular