Datenflüsse

Die konzeptuelle Beschreibung steht unter Wie es funktioniert. Hier folgen die Abläufe auf Klassenebene.

Erkennung (Timer)

sequenceDiagram
    autonumber
    participant Timer as GetChanges (Timer)
    participant Init as InitializationService
    participant PJ as ProcessJobsForRetrievalInOrderService
    participant Job as JobService
    participant WM as WatermarkService
    participant Conn as Source-Connector
    participant SBS as ServiceBusService
    participant SB as Service Bus

    Timer->>Init: EnsureInitializedAsync()
    Note over Init: einmalig: Storage → Connections → Mappings → Jobs
    Timer->>PJ: Execute()
    PJ->>Job: GetJobsToRun()
    Note over Job,WM: GetAllWatermarks() (Bulk) für LastCronRun
    loop je Job nach Order
        PJ->>WM: GetWatermarkByJobName(job)
        PJ->>Conn: RetrieveChangesAsync(mapping, job, watermark, filter)
        loop seitenweise
            Conn-->>PJ: QueueMessage (serialisiert)
            PJ->>SBS: AddMessageOrSendAsync(msg, priority)
            alt Batch voll (200) oder Größenlimit
                SBS->>SB: SendMessagesAsync(batch)
            end
            PJ->>PJ: lastChangeTracker = msg.ChangeTracker
        end
        PJ->>SBS: FlushBatch()
        PJ->>WM: UpdateWatermarkByJobName(newWatermark, lastCronRun)
    end

• Die Serialisierung (Key-Aufbau → Feld-Mappings → Source-Operationen / Resolver) passiert im Connector (ConnectorBase.Serialize).
• Die Watermark wird erst am Ende eines Jobs geschrieben. Bricht der Lauf vorher ab, wird beim nächsten Lauf erneut ab dem alten Stand gelesen; SkipOutdatedMessages fängt Doppel-Nachrichten im Ziel ab.
• Bei reinem Zeitstempel-Vorrücken ohne fachliche Änderung wird JobActionType.TimeStamp verwendet.
• MetricsService zählt ChangesDetected je Job.

Erkennung (Durable)

sequenceDiagram
    autonumber
    participant HTTP as StartGetChangesOrchestration
    participant Orch as GetChangesOrchestrator
    participant State as JobState (Entity)
    participant Act as GetChangesActivity
    participant Conn as Source-Connector
    participant SBS as ServiceBusService

    HTTP-->>HTTP: 202 + InstanceId
    loop je Job aus GetJobsToRun
        Orch->>State: InProgress(jobName)?
        alt in Bearbeitung
            Orch->>Orch: überspringen
        else
            Orch->>State: Add(jobName)
            loop do … while HasMoreRecord
                Orch->>Act: CallActivity(JobActivityParam)
                Act->>Conn: RetrieveChangesAsync(…, MaxBatchSize=3)
                Act->>SBS: SendMessage(record, priority)
                Act-->>Orch: Page, Cookie, Watermark, HasMoreRecord
            end
            Orch->>State: Remove(jobName)
        end
    end

Der Durable-Pfad persistiert die Watermark nicht und ist im Timer-Betrieb nicht aktiv.

Verarbeitung (Service Bus → Zielsystem)

sequenceDiagram
    autonumber
    participant SB as Service Bus Queue
    participant PQI as Process*QueueItems
    participant Init as InitializationService
    participant PQM as ProcessQueueMessageService
    participant Conn as Target-Connector
    participant SBS as ServiceBusService

    SB->>PQI: Nachricht
    PQI->>PQI: Lock-Renewal-Task starten (30 s)
    PQI->>Init: EnsureInitializedAsync()
    PQI->>PQM: Execute(busMessage)
    PQM->>PQM: QueueMessage deserialisieren, Job + Mapping laden
    alt DeliveryCount > 1
        PQM->>PQM: Backoff DeliveryCount² · 1000 ms
    end
    alt SkipOutdatedMessages und Quelle älter als Ziel
        PQM->>SBS: SendMessage → sbqignored
    else NewUpdate
        PQM->>Conn: Upsert(message, mapping[, responseMapping, job])
        opt Response vorhanden
            PQM->>SBS: SendMessage(response, priority)
        end
    else Delete
        PQM->>Conn: Delete(message, mapping)
    else InvokeAction
        PQM->>Conn: InvokeAction(message, job)
    end
    alt Erfolg
        PQI->>SB: CompleteMessageAsync()
    else TransitiveException
        PQM->>SBS: SendMessage → sbqtransitive*
        PQI->>SB: CompleteMessageAsync()
    else andere Exception
        PQI->>SB: AbandonMessageAsync(+ Error-Properties)
        Note over SB: nach maxDeliveryCount ⇒ Dead-Letter
    end
    PQI->>PQI: Lock-Renewal stoppen

MetricsService zählt MessagesProcessed mit dem Ausgang (Success, Transitive, Ignored, Failed).

Response-Job

sequenceDiagram
    autonumber
    participant ProcA as Process*QueueItems (Job A: CRM → BC)
    participant BC as Target-Connector BC
    participant SB as Service Bus
    participant ProcB as Process*QueueItems (ResponseJob: BC → CRM)
    participant CRM as Target-Connector CRM

    ProcA->>BC: Upsert(message, mapping, responseMapping, job)
    BC-->>ProcA: QueueMessage mit erzeugtem Ziel-Schlüssel
    ProcA->>SB: SendMessage(responseMessage, priority)
    SB->>ProcB: Response-Nachricht
    ProcB->>CRM: Upsert (z. B. BC-Belegnummer zurückschreiben)

Fehlerpfad

flowchart TD
    MSG["Queue-Nachricht"] --> PROC["ProcessQueueMessageService.Execute"]
    PROC -->|Erfolg| DONE["Complete"]
    PROC -->|TransitiveException| TRANS["sbqtransitive*<br/>kein Trigger, liegt bis TTL"]
    PROC -->|veraltet| IGN["sbqignored<br/>kein Trigger"]
    PROC -->|andere Exception| AB["Abandon + Error-Properties"]
    AB --> RETRY{"DeliveryCount<br/>< maxDeliveryCount?"}
    RETRY -->|ja| MSG
    RETRY -->|nein| DLQ["Dead-Letter-Subqueue"]
    DLQ -.->|manuell| DLM["DeadLetterMover"]
    TRANS -.->|manuell| DLM
    DLM -.->|Resubmit ohne Error-Properties| MSG

Transitive- und Ignored-Queues werden nicht automatisch weiterverarbeitet. Sie dienen der Isolation und der Analyse. Ein Requeue erfolgt über den DeadLetterMover.

DeadLetterMover

Konsolenanwendung unter src/DeadLetterMover. Sie liest eine Queue oder deren $DeadLetterQueue, entscheidet anhand der ErrorMessage per Muster (Resubmit, None, Skip) und sendet Nachrichten ohne Error-Properties zurück in die Original-Queue, dedupliziert über SourceKey. Connection-String, Queue-Name und der Schalter UseDeadletter stehen als Konstanten im Code und werden vor dem Lauf angepasst. Der Connection-String darf nicht eingecheckt werden.

Persistenz

Watermark-Logik

Die Wertlogik liegt in WatermarkValue (Models/Common), der Service persistiert nur.

• GetWatermarkByJobName liefert bei fehlendem Eintrag eine nicht-persistierte Entity; LastCronRun == null bedeutet „Job ist fällig“.
• GetAllWatermarks ist eine Bulk-Abfrage auf PartitionKey watermark und wird vom Scheduler genutzt, damit nicht je Job eine Einzelabfrage anfällt.
• WatermarkValue.Initialize(type, stored) setzt Typ-Defaults (date → 0001-01-01, datetime → 0001-01-01T00:00:00Z). Der Typ wird getrimmt und case-insensitiv geparst; ein unbekannter Typ belässt den gespeicherten Wert und erzeugt eine Warnung.
• Advance(changeTracker) vergleicht typabhängig mit InvariantCulture und RoundtripKind.
• UpdateTableEntry schreibt ohne ETag (Last-Writer-Wins). Der HTTP-Endpunkt ExecuteJob kann parallel zum Timer laufen; ein veralteter Stand ist immer älter und führt zu Wiederholungen, nicht zu Lücken.
• Reset: Feld Watermark in der Tabelle leeren → vollständiger Neulauf. Anwendersicht: Watermarks.