Edge Fonksiyonlarından MQTT (Cloudflare Workers, Vercel)

Çoğu edge runtime'ında ham TCP socket'leri ve Node.js ağ yığını yoktur; bu yüzden alışılmış MQTT istemci kütüphaneleri orada çalışmaz. Ama WebSocket API'si vardır ve MQTT, WebSocket üzerinden gayet iyi çalışır. Bir Cloudflare Worker wss://broker.connectmq.com:8084/mqtt adresini açabilir, bir CONNECT gönderebilir, publish edip kapatabilir — hepsi tek bir istek içinde. Vercel'de Node.js runtime'ı normal bir MQTT istemcisini çalıştırır; yani edge yöntemi yalnızca salt edge runtime'ları için gereklidir.

Bu eğitim Cloudflare Workers için bağımlılıksız bir MQTT publisher'ı oluşturuyor, ctx.waitUntil'i açıklıyor, önemli platform limitlerini listeliyor ve her mesajı tarayıcınızda nasıl doğrulayacağınızı gösteriyor.

Edge runtime'ları neyi yapabilir, neyi yapamaz

Pratik kural: edge fonksiyonları publisher'dır, subscriber değil. Bir HTTP isteğine tepki verir, bir veya birkaç MQTT mesajı yayar ve biter. Sürekli dinlemesi gereken her şey uzun ömürlü bir servise aittir.

Vercel kullanıyorsanız en basit çözüm, MQTT ile konuşan route için Node.js runtime'ını kullanmak ve onu diğer serverless fonksiyonlar gibi ele almaktır — AWS Lambda'dan publish yazısındaki bağlantı yeniden kullanım kalıpları doğrudan geçerlidir. Vercel de bugün iş yüklerinin çoğunu Node.js runtime'ına yönlendiriyor.

WebSocket üzerinde minimal bir MQTT istemcisi

QoS 0 ile publish için yalnızca üç paket gerekir: CONNECT, PUBLISH ve DISCONNECT; buna CONNACK'i okumak eklenir. Bayt düzeni basittir — sabit bir başlık, değişken uzunluklu bir "remaining length" alanı ve uzunluk önekli UTF-8 string'ler (ayrıntılar için MQTT paket yapısı yazısına bakın).

const enc = new TextEncoder();

// 2 baytlık big-endian uzunluk önekli UTF-8 string
const str = (s) => { const b = enc.encode(s); return [b.length >> 8, b.length & 0xff, ...b]; };

// MQTT'nin değişken uzunluklu "remaining length" alanı
function remLen(n) {
  const out = [];
  do {
    let digit = n % 128;
    n = Math.floor(n / 128);
    if (n > 0) digit |= 0x80;
    out.push(digit);
  } while (n > 0);
  return out;
}

const packet = (header, body) => new Uint8Array([header, ...remLen(body.length), ...body]);

function connectPacket(clientId, username, password, keepAlive = 30) {
  let flags = 0x02;                       // clean session
  if (username) flags |= 0x80;
  if (password) flags |= 0x40;
  const body = [...str('MQTT'), 4, flags, keepAlive >> 8, keepAlive & 0xff, ...str(clientId)];
  if (username) body.push(...str(username));
  if (password) body.push(...str(password));
  return packet(0x10, body);              // MQTT 3.1.1 CONNECT
}

const publishPacket = (topic, payload) => packet(0x30, [...str(topic), ...enc.encode(payload)]);
const DISCONNECT = new Uint8Array([0xe0, 0x00]);

export async function publishMqtt(env, topic, payload) {
  const resp = await fetch(env.MQTT_WS_URL, {
    headers: { Upgrade: 'websocket', 'Sec-WebSocket-Protocol': 'mqtt' },
  });
  const ws = resp.webSocket;
  if (!ws) throw new Error(`WebSocket upgrade failed: HTTP ${resp.status}`);
  ws.accept();

  const connack = new Promise((resolve, reject) => {
    const timer = setTimeout(() => reject(new Error('CONNACK timeout')), 5000);
    ws.addEventListener('message', (ev) => {
      const b = new Uint8Array(ev.data);
      if (b[0] !== 0x20) return;
      clearTimeout(timer);
      b[3] === 0 ? resolve() : reject(new Error(`CONNACK return code ${b[3]}`));
    });
    ws.addEventListener('close', () => reject(new Error('socket closed before CONNACK')));
  });

  ws.send(connectPacket(`edge-${crypto.randomUUID().slice(0, 8)}`, env.MQTT_USER, env.MQTT_PASS));
  await connack;
  ws.send(publishPacket(topic, payload));
  ws.send(DISCONNECT);
  ws.close(1000, 'done');
}

Bazı ayrıntılarda hata yapmak kolaydır:

• Subprotocol başlığı. Çoğu broker, upgrade'i yalnızca istemci Sec-WebSocket-Protocol: mqtt sunarsa kabul eder. Bu başlık olmadan genellikle HTTP 400 alırsınız ya da socket hemen kapanır.
• Yol. ConnectMQ'da WebSocket listener'ı /mqtt yolundadır; yanlış yol HTTP 404 döndürür (WebSocket yolu 404).
• Rastgele client ID'ler. Yüzlerce edge lokasyonu Worker'ınızı aynı anda çalıştırabilir; sabit bir ID onların birbirini koparmasına yol açar.
• Binary frame'ler. WebSocket üzerinden MQTT binary mesajlar kullanır; kodun Uint8Array göndermesinin nedeni budur.

Teslim onayı gerekiyorsa publishPacket'i QoS 1'e genişletin (0x32 başlığı ve topic'ten sonra iki baytlık bir packet identifier) ve kapatmadan önce PUBACK'i (0x40) bekleyin. Payload'ları küçük tutun: paket boyutu hesaplayıcı ağdaki tam bayt sayısını gösterir.

Worker

import { publishMqtt } from './mqtt.js';

export default {
  async fetch(request, env, ctx) {
    const url = new URL(request.url);
    const device = url.searchParams.get('device') ?? 'demo';
    const message = JSON.stringify({ device, path: url.pathname, colo: request.cf?.colo, ts: Date.now() });

    // Hemen yanıt ver; MQTT publish'ini arka planda tamamla
    ctx.waitUntil(
      publishMqtt(env, `testmqtt/edge/${device}/hits`, message)
        .catch((err) => console.error('mqtt publish failed:', err.message)),
    );
    return new Response('queued\n');
  },
};

name = "edge-mqtt"
main = "src/index.js"
compatibility_date = "2026-10-01"

[vars]
MQTT_WS_URL = "https://broker.connectmq.com:8084/mqtt"

Kimlik bilgilerini wrangler.toml içine değil, secret olarak saklayın: npx wrangler secret put MQTT_USER ve npx wrangler secret put MQTT_PASS. URL'nin https:// kullandığına dikkat edin — Workers fetch'i Upgrade başlığını gördüğünde WebSocket upgrade'ini kendisi yapar.

ctx.waitUntil, yanıt gönderildikten sonra promise sonuçlanana kadar Worker'ı ayakta tutar; böylece ziyaretçi broker el sıkışmasını beklemez. Çağıranın publish'in başarılı olup olmadığını bilmesi gerekiyorsa, yanıt vermeden önce await publishMqtt(...) kullanın ve hata durumunda 502 döndürün.

cloudflare:sockets ile ham TCP

Cloudflare Workers, cloudflare:sockets içindeki connect() ile giden TCP bağlantıları da açabilir; secureTransport: "on" ile TLS dahil. Yukarıdaki paket fonksiyonları değişmeden çalışır — ws.send yerine socket.writable'a yazın ve CONNACK'i socket.readable'dan okuyun. Böylece WebSocket yerine 8883 portunu kullanabilirsiniz. Yine de WebSocket daha taşınabilir seçenektir, çünkü aynı kod diğer edge platformlarında ve tarayıcılarda da çalışır (WebSocket üzerinden MQTT).

Tasarımda dikkate alınacak limitler

Kesin değerler platforma ve plana göre değişir ve zamanla güncellenir; güncel dokümantasyona bakın. MQTT için önemli olan kategoriler şunlardır:

• İstek başına CPU ve gerçek süre. Bir TLS + WebSocket + MQTT el sıkışması birkaç ağ gidiş-dönüşüne mal olur; mesaj başına bir bağlantı yerine istek başına bir toplu publish yapın.
• Güvenilebilecek bir bağlantı yeniden kullanımı yok. Isolate'ler yeniden kullanılabilir, ancak her isteği yeni bir bağlantı olarak ele alın.
• Giden bağlantı ve subrequest limitleri. Her WebSocket bu limitlere dahildir.
• Giden portlar. Bazı platformlar giden isteklerin kullanabileceği portları kısıtlar. 8084'e izin verilmiyorsa 443 üzerindeki bir broker listener'ını kullanın (443 portunda MQTT).
• Broker'daki bağlantı hızı. Sitenizdeki bir trafik sıçraması, bir MQTT bağlantı sıçramasına dönüşür. Yüksek hacimli olaylar için bunları tek bir bağlantıyı açık tutan bir toplayıcı servise göndermeyi düşünün.

Sık karşılaşılan hatalar

WebSocket'e özgü diğer hatalar WebSocket connection failed yazısında ele alınıyor.

Test etmek

1. Online MQTT istemcisini açın, güvenli WebSocket ile broker.connectmq.com adresine bağlanın ve testmqtt/edge/# topic'ine abone olun.
2. Worker'ı npx wrangler dev ile yerelde çalıştırın ve çağırın: curl "http://localhost:8787/checkout?device=kiosk-1".
3. Tarayıcıda cihaz adını ve zaman damgasını içeren bir mesaj görünmelidir. Görünmüyorsa wrangler dev konsolunda publishMqtt'ten gelen hata mesajına bakın.
4. npx wrangler deploy ile deploy edin ve workers.dev URL'sine karşı tekrarlayın; colo alanı artık mesajı hangi edge lokasyonunun publish ettiğini gösterir.

Edge mesajlarının tarayıcınıza gelişini izleyin

Edge fonksiyonları ile MQTT, WebSocket'te buluşur: yaklaşık elli satırlık paket kodlaması, herhangi bir edge runtime'ını bağımlılıksız bir MQTT publisher'ına dönüştürür. Fonksiyonları kısa ömürlü publisher'lar olarak tutun, kimlik bilgilerini sıkı ACL'lerle koruyun ve uzun ömürlü abonelikleri bağlı kalabilen servislere bırakın.

Sıkça sorulan sorular