Integrationen bauen

LowCode bietet vielfältige Möglichkeiten, um Drittsysteme miteinander zu verbinden. Ob REST-APIs, Message Queues, Datenbanken oder Cloud-Services — mit den richtigen Nodes und Patterns erstellen Sie robuste Integrationen in kürzester Zeit.

Integrations-Patterns

Die folgenden Patterns bilden die Grundlage für alle Integrationen:

Pattern	Beschreibung	Typische Nodes
Request-Response	Synchroner Aufruf eines externen Services	`http request`
Publish-Subscribe	Asynchrone Kommunikation über Message Queues	`amqp out`, `amqp in`
Database CRUD	Direkte Datenbank-Operationen	`postgresql`, `mongodb`
Event-Driven	Reaktion auf externe Events	`mqtt in`, `amqp in`
Batch Processing	Verarbeitung großer Datenmengen	`split`, `join`, `delay`

HTTP-Request Node

Der http request-Node ist das Universalwerkzeug für die Anbindung externer REST-APIs.

Grundkonfiguration


// Function Node: HTTP-Request vorbereiten
msg.url = "https://api.example.com/data";
msg.method = "GET";
msg.headers = {
  "Authorization": "Bearer " + flow.get("apiToken"),
  "Accept": "application/json"
};
return msg;

POST-Request mit Body


// Function Node: POST-Request vorbereiten
msg.url = "https://api.example.com/orders";
msg.method = "POST";
msg.headers = {
  "Content-Type": "application/json",
  "Authorization": "Bearer " + flow.get("apiToken")
};
msg.payload = {
  customerId: msg.customerId,
  items: msg.orderItems,
  total: msg.orderTotal
};
return msg;

Beispiel 1: REST-API-Integration mit Datentransformation

Dieses Beispiel zeigt, wie Daten aus einer externen API abgerufen, transformiert und in einer Datenbank gespeichert werden.

Szenario

Kundendaten aus einem CRM-System abrufen und in eine lokale PostgreSQL-Datenbank synchronisieren.

Flow-Aufbau


[inject: Alle 15 Min]
  → [fn: prepareRequest]
  → [http request: GET CRM-API]
  → [fn: transformData]
  → [split: Einzelne Kunden]
  → [fn: buildUpsertQuery]
  → [postgresql: Upsert]
  → [join: Ergebnisse sammeln]
  → [fn: logResult]
  → [debug: Zusammenfassung]

Request vorbereiten


// Function Node: prepareRequest
msg.url = `${env.get("CRM_API_URL")}/customers`;
msg.method = "GET";
msg.headers = {
  "Authorization": `Bearer ${flow.get("crmToken")}`,
  "Accept": "application/json"
};
 
// Nur Änderungen seit letztem Sync abrufen
const lastSync = flow.get("lastSyncTimestamp") || "2020-01-01T00:00:00Z";
msg.url += `?updatedSince=${lastSync}`;
 
return msg;

Daten transformieren


// Function Node: transformData
const crmCustomers = msg.payload.data || [];
 
// CRM-Format in internes Format transformieren
msg.payload = crmCustomers.map(c => ({
  externalId: c.id,
  firstName: c.first_name,
  lastName: c.last_name,
  email: c.contact_email,
  phone: c.contact_phone || null,
  company: c.organization_name || null,
  updatedAt: new Date().toISOString()
}));
 
// Sync-Zeitstempel aktualisieren
flow.set("lastSyncTimestamp", new Date().toISOString());
 
node.status({
  fill: "green",
  shape: "dot",
  text: `${msg.payload.length} Kunden transformiert`
});
 
return msg;

Datenbank-Upsert


// Function Node: buildUpsertQuery
const customer = msg.payload;
 
msg.params = [
  customer.externalId,
  customer.firstName,
  customer.lastName,
  customer.email,
  customer.phone,
  customer.company,
  customer.updatedAt
];
 
return msg;

Der PostgreSQL-Node führt folgende Query aus:


INSERT INTO customers (external_id, first_name, last_name, email, phone, company, updated_at)
VALUES ($1, $2, $3, $4, $5, $6, $7)
ON CONFLICT (external_id)
DO UPDATE SET
  first_name = EXCLUDED.first_name,
  last_name = EXCLUDED.last_name,
  email = EXCLUDED.email,
  phone = EXCLUDED.phone,
  company = EXCLUDED.company,
  updated_at = EXCLUDED.updated_at;

Beispiel 2: AMQP Message Queue Integration

Verwenden Sie AMQP (z.B. RabbitMQ) für die asynchrone Kommunikation zwischen Systemen.

Szenario

Bestellungen empfangen, verarbeiten und Bestätigungen in eine Antwort-Queue schreiben.

Flow-Aufbau


[amqp in: orders.incoming]
  → [fn: validateOrder]
  → [switch: valid?]
    ├─ Ja  → [fn: processOrder]
    │        → [postgresql: Bestellung speichern]
    │        → [fn: buildConfirmation]
    │        → [amqp out: orders.confirmed]
    └─ Nein → [fn: buildRejection]
             → [amqp out: orders.rejected]

Nachricht empfangen und validieren


// Function Node: validateOrder
const order = msg.payload;
 
// Pflichtfelder prüfen
const requiredFields = ["customerId", "items", "total"];
const missingFields = requiredFields.filter(f => !order[f]);
 
if (missingFields.length > 0) {
  msg.valid = false;
  msg.validationError = `Fehlende Felder: ${missingFields.join(", ")}`;
  return msg;
}
 
// Geschäftsregeln prüfen
if (order.total <= 0) {
  msg.valid = false;
  msg.validationError = "Bestellsumme muss positiv sein";
  return msg;
}
 
msg.valid = true;
return msg;

Bestätigung senden


// Function Node: buildConfirmation
msg.payload = {
  orderId: msg.orderId,
  status: "confirmed",
  timestamp: new Date().toISOString(),
  estimatedDelivery: new Date(Date.now() + 3 * 86400000).toISOString()
};
return msg;

Beispiel 3: Elasticsearch-Integration

Integrieren Sie Elasticsearch für Volltextsuche und Log-Analyse.

Szenario

Prozess-Events in Elasticsearch indexieren und eine Such-API bereitstellen.

Flow: Events indexieren


[engine event: processFinished]
  → [fn: buildElasticDocument]
  → [http request: POST Elasticsearch]
  → [fn: checkResponse]
  → [debug: Indexierungsstatus]


// Function Node: buildElasticDocument
const event = msg.payload;
const indexName = `process-events-${new Date().toISOString().slice(0, 7)}`;
 
msg.url = `${env.get("ELASTICSEARCH_URL")}/${indexName}/_doc`;
msg.method = "POST";
msg.headers = { "Content-Type": "application/json" };
msg.payload = {
  processModelId: event.processModelId,
  processInstanceId: event.processInstanceId,
  eventType: event.eventType,
  duration: event.duration || null,
  timestamp: new Date().toISOString(),
  metadata: event.metadata || {}
};
 
return msg;

Flow: Such-API


[http in: GET /api/search]
  → [fn: buildSearchQuery]
  → [http request: POST Elasticsearch]
  → [fn: formatResults]
  → [http response]


// Function Node: buildSearchQuery
const query = msg.req.query.q || "*";
const from = parseInt(msg.req.query.offset) || 0;
const size = parseInt(msg.req.query.limit) || 20;
 
msg.url = `${env.get("ELASTICSEARCH_URL")}/process-events-*/_search`;
msg.method = "POST";
msg.headers = { "Content-Type": "application/json" };
msg.payload = {
  query: {
    bool: {
      should: [
        { match: { processModelId: query } },
        { match: { "metadata.description": query } }
      ]
    }
  },
  from: from,
  size: size,
  sort: [{ timestamp: "desc" }]
};
 
return msg;


// Function Node: formatResults
const esResponse = msg.payload;
const hits = esResponse.hits || { total: { value: 0 }, hits: [] };
 
msg.payload = {
  success: true,
  total: hits.total.value,
  results: hits.hits.map(h => ({
    id: h._id,
    score: h._score,
    ...h._source
  }))
};
msg.statusCode = 200;
return msg;

Error Handling mit Catch Nodes

Robustes Error Handling ist der Schlüssel zu zuverlässigen Integrationen.

Catch-Node Pattern


[catch: Scope = aktuelle Flow-Gruppe]
  → [fn: classifyError]
  → [switch: Fehlertyp]
    ├─ Netzwerk    → [fn: retry]     → [delay] → [Ursprungs-Flow]
    ├─ Validierung → [fn: logError]  → [debug]
    └─ Unbekannt   → [fn: alertOps]  → [email: Ops-Team]


// Function Node: classifyError
const error = msg.error;
const source = error.source ? error.source.type : "unknown";
 
if (source === "http request" && error.message.includes("ECONNREFUSED")) {
  msg.errorType = "network";
} else if (error.message.includes("validation")) {
  msg.errorType = "validation";
} else {
  msg.errorType = "unknown";
}
 
msg.errorLog = {
  type: msg.errorType,
  message: error.message,
  source: error.source ? error.source.name : "unknown",
  timestamp: new Date().toISOString()
};
 
return msg;

Retry-Pattern mit Exponential Backoff


// Function Node: retry
let retryCount = msg.retryCount || 0;
const maxRetries = 3;
 
if (retryCount >= maxRetries) {
  // Maximale Versuche erreicht
  node.error(`Max Retries erreicht für: ${msg.url}`, msg);
  msg.payload = { error: "Service nicht erreichbar nach 3 Versuchen" };
  return [null, msg]; // Zweiter Ausgang: Fehlgeschlagen
}
 
retryCount++;
msg.retryCount = retryCount;
 
// Exponentielles Backoff: 2s, 4s, 8s
msg.delay = Math.pow(2, retryCount) * 1000;
 
node.status({
  fill: "yellow",
  shape: "ring",
  text: `Retry ${retryCount}/${maxRetries} in ${msg.delay / 1000}s`
});
 
return [msg, null]; // Erster Ausgang: Erneut versuchen

Best Practices

1. Environment Variables nutzen

Speichern Sie URLs und Credentials niemals im Flow:


// Korrekt: Environment Variables
const apiUrl = env.get("CRM_API_URL");
const apiKey = env.get("CRM_API_KEY");

2. Timeouts konfigurieren

Setzen Sie immer Timeouts für HTTP-Requests (im http request-Node konfigurierbar oder via msg):


msg.requestTimeout = 10000; // 10 Sekunden Timeout

3. Idempotenz sicherstellen

Stellen Sie sicher, dass Ihre Integrationen wiederholbar sind, ohne ungewollte Seiteneffekte:


// Idempotenten Key mitgeben
msg.headers["Idempotency-Key"] = `${msg.orderId}-${msg.timestamp}`;

4. Logging und Monitoring

Protokollieren Sie alle wichtigen Integrationsschritte:


// Status im Node anzeigen
node.status({
  fill: "green",
  shape: "dot",
  text: `Letzte Sync: ${new Date().toLocaleTimeString()}`
});

Nächste Schritte

REST-APIs entwickeln — Eigene APIs bereitstellen
Workflow-Integration — Integrationen mit BPMN-Prozessen verbinden
AMQP-Nodes — Erweiterte AMQP-Features im Enterprise Image
ProcessCube® Konzepte — Architektur-Patterns vertiefen