Werkspur Docs
Zur Website

AMQP-Listener

Zielsystem und Listener sind zwei Artefakte

Für eingehende AMQP-Nachrichten brauchen Sie ein AMQP-Zielsystem und einen AMQP-Listener. Beide haben verschiedene Aufgaben:

Der Listener enthält keine Broker-Adresse und kein Passwort. Er referenziert das Zielsystem nur über destination. Das Zielsystem enthält umgekehrt keine Queue und keinen eingehenden Ereignistyp.

Die Oberfläche bietet unter „Zielsysteme“ einen Editor für die Verbindung. Für AMQP-Listener gibt es dort derzeit keinen Editor. Sie pflegen Listener über die Admin-API /api/amqp-listeners. Zusätzlich schaltet eine nicht leere Umgebungsvariable FOUNDRY_AMQP_URL die eingehende Bridge ein.

Dieser Umgebungswert ist nur ein Schalter. WERKSPUR prüft beim Start das Schema amqp:// oder amqps://, verwendet den Wert danach aber nicht für eine Verbindung. Die tatsächliche Broker-Adresse und Anmeldung stammen bei jedem Listener aus dessen referenziertem AMQP-Zielsystem. Ein leerer Wert lässt Definitionen gespeichert, konsumiert ihre Queues aber nicht.

Ausgehendes AMQP ist davon unabhängig. Der Aktionsschritt amqp.publish verbindet sich direkt über sein Zielsystem und braucht den Schalter FOUNDRY_AMQP_URL nicht. Die eingehende Bridge veröffentlicht auch keine Antwort auf eine ausgehende Nachricht, solange Sie dafür nicht ausdrücklich eine Queue, einen Listener und einen Ereignis-Trigger konfigurieren.

Platzhalter für API-Beispiele

Verwenden Sie in Dokumentation und Tests keine echten Zugangsdaten. Die folgenden Variablen und JSON-Werte sind Platzhalter und dürfen nicht unverändert in ein Produktivsystem übernommen werden:

export WERKSPUR='https://werkspur.internal.invalid'
export WERKSPUR_SESSION='<SESSION_TOKEN>'

Das Secret selbst legen Sie vor dem Zielsystem unter „Secrets“ an. Im Zielsystem steht nur sein Name. Details dazu finden Sie unter Secrets.

AMQP-Zielsystem anlegen

Admins können das Zielsystem in der Oberfläche oder über POST /api/destinations anlegen:

curl -sS -X POST "$WERKSPUR/api/destinations" \
  -H "Authorization: Bearer $WERKSPUR_SESSION" \
  -H 'Content-Type: application/json' \
  -d '{
    "name": "inbound-broker",
    "kind": "amqp",
    "base_url": "amqps://broker.internal.invalid:5671/",
    "username": "<BROKER_USER>",
    "auth": {"secret": "<SECRET_NAME>"}
  }'

Für kind: amqp gelten folgende Grenzen:

Das Zielsystem speichert kein Queue- oder Listener-Feld. Derselbe Verbindungsdatensatz kann von mehreren Listenern verwendet werden.

Listener definieren

Danach legen Sie den Listener über POST /api/amqp-listeners an:

curl -sS -X POST "$WERKSPUR/api/amqp-listeners" \
  -H "Authorization: Bearer $WERKSPUR_SESSION" \
  -H 'Content-Type: application/json' \
  -d '{
    "name": "incoming-events",
    "destination": "inbound-broker",
    "queue": "werkspur.events.in",
    "event_type": "integration.event.received"
  }'

Die Listener-Felder sind:

Eine erfolgreiche Definition antwortet mit 200 und {"name":"incoming-events"}. Derselbe Name ersetzt die bestehende Definition. Die Bridge beendet dann den alten Konsum und startet nach dem nächsten Abgleich mit den neuen Feldern. GET /api/amqp-listeners listet alle Listener. DELETE /api/amqp-listeners/{name} entfernt eine Definition. Es gibt weder enabled noch Listener-spezifische Zugangsdaten, Filter oder Transformationsfelder.

Laufzeit, Bestätigung und Wiederholungen

Die Bridge gleicht Listener alle 30 Sekunden ab. Für jeden Listener löst sie Zielsystem und Secret bei jedem Verbindungsversuch neu auf. Fehlt ein Zielsystem, ist es nicht vom Typ amqp oder lässt sich das Secret nicht auflösen, überspringt die Bridge den Versuch und prüft nach 30 Sekunden erneut.

Scheitert die Broker-Verbindung oder schließt der Lieferkanal, versucht der Listener nach zwei Sekunden erneut zu verbinden. WERKSPUR deklariert die Eingangsqueue dauerhaft, verarbeitet mit Prefetch 1 und bestätigt jede Nachricht manuell.

Für eine gültige Nachricht gilt diese Reihenfolge:

  1. Die Bridge prüft Größe und JSON-Syntax.
  2. Sie hängt die JSON-Payload ohne Flattening oder Bridge-spezifische Transformation als Domain-Ereignisdaten an den Stream amqp:<listener-name> an. Der Ereignistyp entspricht event_type. Die JSONB-Speicherung darf dabei die Reihenfolge von Objektschlüsseln ändern.
  3. Der Ereignisspeicher legt im selben Commit die Weitergabe an den internen Ereignisstream vor.
  4. Erst nach dem erfolgreichen Anhängen bestätigt die Bridge die AMQP-Nachricht.

Bei einem konkurrierenden Anhängen liest die Bridge die aktuelle Stream-Position einmal neu und wiederholt den Vorgang genau einmal. Bleibt er erfolglos, weist sie die Nachricht mit Requeue zurück. Der Broker darf sie dann erneut liefern.

Auch nach einem erfolgreichen dauerhaften Anhängen bleibt die Zustellung mindestens einmal. Schlägt die anschließende Bestätigung fehl, protokolliert die Bridge den Fehler. Der Broker kann dieselbe Nachricht erneut liefern. Die Bridge wertet keine Nachrichten-ID aus und führt keine Deduplizierung der Payload durch. Eine erneute Lieferung kann deshalb ein zweites Domain-Ereignis und einen zweiten Triggerlauf erzeugen. Nachgelagerte Aktionen und externe Empfänger müssen Wiederholungen anhand eines stabilen fachlichen Schlüssels erkennen oder dieselbe Wirkung gefahrlos erneut anwenden können.

Größenlimit und Dead-Letter-Queue

Die Payload darf höchstens 256 KiB, also 256 * 1024 Byte, groß sein und muss syntaktisch gültiges JSON enthalten. Ob das JSON ein Objekt, ein Array oder ein einzelner JSON-Wert ist, prüft die Bridge nicht weiter. Für die spätere Übergabe von ${event.data.<feld>} brauchen Sie jedoch ein JSON-Objekt mit den erwarteten Feldern.

Eine größere oder syntaktisch ungültige Nachricht gilt als nicht wiederholbar fehlerhaft. Die Bridge erzeugt kein Domain-Ereignis und sendet eine negative Bestätigung ohne Requeue. Die deklarierte Eingangsqueue verweist solche Nachrichten an eine dauerhafte Fanout-Exchange namens <queue>.dlx. Daran ist eine ebenfalls dauerhafte Dead-Letter-Queue gebunden, die denselben Namen <queue>.dlx trägt.

Prüfen Sie die Dead-Letter-Queue, bevor Sie eine Nachricht erneut veröffentlichen. Eine unveränderte, ungültige Payload würde erneut dort landen. Die API liefert für diesen asynchronen Fehler keinen Statuscode an den ursprünglichen Absender.

Übergabe an einen Trigger

Ein erfolgreiches Anhängen durch den Listener wird über den internen Ereignisstrom für event-Trigger sichtbar. Der Trigger muss exakt auf denselben event_type reagieren. Felder des ursprünglichen JSON-Objekts stehen beim Abgleich unter data und in Eingaben als ${event.data.<feld>} bereit.

Ein passender Triggerausschnitt sieht beispielsweise so aus:

{
  "type": "event",
  "match": {"event_type": "integration.event.received"},
  "action": "process-incoming-event",
  "inputs": {"reference": "${event.data.reference}"},
  "run_as": "operator",
  "enabled": true
}

Der Listener löst die Aktion nicht direkt aus. Zwischen beiden liegen das dauerhafte Anhängen des Ereignisses, die interne Weitergabe und der Ereignisabgleich. Für diesen Übergabepfad müssen PostgreSQL und NATS verfügbar sein. Ein fehlendes Feld in ${event.data.<feld>} lässt den Abgleich den Triggerlauf auslassen; der bereits gespeicherte AMQP-Eingang bleibt trotzdem bestätigt.

Nicht unterstützte Annahmen

Die ausgehende Verwendung eines AMQP-Zielsystems beschreibt Zielsysteme. Die nachgelagerte Ereignisdefinition erläutert Trigger. Eine Diagnosefolge ohne echte Zugangsdaten finden Sie unter Integration: Fehlersuche.

ZurückMQTT & Telemetrie WeiterAPI & Authentifizierung