Implementiere DefaultSchedulerControlUseCase für Scheduler-Orchestrierung
Implementiert SchedulerControlUseCase als zentralen Orchestrator: - start()-Sequenz mit STARTING → RUNNING_IDLE und vollständigem Rollback - stop()-Sequenz mit CAS-gesichertem STOPPING_BATCH_ACTIVE für laufende Batches - executeWrappedTick() (package-private) setzt RUNNING_BATCH_ACTIVE vor dem Trigger und leitet Folgezustand aus BatchRunTriggerResult-Variante ab - AtomicReference<SchedulerStatus> für threadsichere Zustandsverwaltung - Intervall wird beim Start aus SchedulerSettingsPort geladen, Minimum 30 s Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
+299
@@ -0,0 +1,299 @@
|
||||
package de.gecheckt.pdf.umbenenner.application.usecase;
|
||||
|
||||
import de.gecheckt.pdf.umbenenner.application.port.in.SchedulerControlUseCase;
|
||||
import de.gecheckt.pdf.umbenenner.application.port.in.SchedulerStartException;
|
||||
import de.gecheckt.pdf.umbenenner.application.port.in.SchedulerState;
|
||||
import de.gecheckt.pdf.umbenenner.application.port.in.SchedulerStatus;
|
||||
import de.gecheckt.pdf.umbenenner.application.port.out.BatchRunTrigger;
|
||||
import de.gecheckt.pdf.umbenenner.application.port.out.BatchRunTriggerResult;
|
||||
import de.gecheckt.pdf.umbenenner.application.port.out.ConfigurationFileLockPort;
|
||||
import de.gecheckt.pdf.umbenenner.application.port.out.RunSummary;
|
||||
import de.gecheckt.pdf.umbenenner.application.port.out.SchedulerConfig;
|
||||
import de.gecheckt.pdf.umbenenner.application.port.out.SchedulerPort;
|
||||
import de.gecheckt.pdf.umbenenner.application.port.out.SchedulerSettings;
|
||||
import de.gecheckt.pdf.umbenenner.application.port.out.SchedulerSettingsPort;
|
||||
|
||||
import org.apache.logging.log4j.LogManager;
|
||||
import org.apache.logging.log4j.Logger;
|
||||
|
||||
import java.time.Instant;
|
||||
import java.util.Objects;
|
||||
import java.util.Optional;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
|
||||
/**
|
||||
* Implementiert {@link SchedulerControlUseCase} als zentraler Orchestrator
|
||||
* des automatischen Schedulers.
|
||||
* <p>
|
||||
* Dieser Use Case:
|
||||
* <ul>
|
||||
* <li>Verwaltet den Scheduler-Lebenszyklus (Start, Stop) über einen
|
||||
* {@link SchedulerPort} und persistiert dabei den {@code enabled}-Wert.</li>
|
||||
* <li>Hält den exklusiven OS-Lock auf die Konfigurationsdatei über
|
||||
* {@link ConfigurationFileLockPort}, solange der Scheduler aktiv ist.</li>
|
||||
* <li>Liest beim Erstellen das konfigurierte Intervall via
|
||||
* {@link SchedulerSettingsPort}.</li>
|
||||
* <li>Publiziert unveränderliche Zustandssnapshots via {@link #getStatus()},
|
||||
* die threadsicher über eine {@link AtomicReference} verwaltet werden.</li>
|
||||
* </ul>
|
||||
* <p>
|
||||
* Alle Zustandsübergänge erfolgen atomar. Der {@link BatchRunTrigger} wird
|
||||
* pro Tick in einem internen Wrapper ausgeführt, der vor dem Aufruf den
|
||||
* Zustand auf {@link SchedulerState#RUNNING_BATCH_ACTIVE} setzt und nach
|
||||
* Abschluss den Folgezustand ableitet.
|
||||
* <p>
|
||||
* {@link #start()} und {@link #stop()} sind idempotent und dürfen serialisiert
|
||||
* vom GUI-Worker-Thread aufgerufen werden. {@link #getStatus()} ist jederzeit
|
||||
* von beliebigen Threads lesbar.
|
||||
*/
|
||||
public class DefaultSchedulerControlUseCase implements SchedulerControlUseCase {
|
||||
|
||||
private static final Logger logger =
|
||||
LogManager.getLogger(DefaultSchedulerControlUseCase.class);
|
||||
|
||||
private static final int MINIMUM_INTERVAL_SECONDS = 30;
|
||||
|
||||
private final SchedulerPort schedulerPort;
|
||||
private final ConfigurationFileLockPort lockPort;
|
||||
private final SchedulerSettingsPort settingsPort;
|
||||
private final BatchRunTrigger batchRunTrigger;
|
||||
private final AtomicReference<SchedulerStatus> statusRef;
|
||||
private final int intervalSeconds;
|
||||
|
||||
/**
|
||||
* Erstellt einen neuen Use Case.
|
||||
* <p>
|
||||
* Das Scheduler-Intervall wird sofort aus {@code settingsPort} gelesen;
|
||||
* Werte unter {@value MINIMUM_INTERVAL_SECONDS} Sekunden werden auf diesen
|
||||
* Mindestwert angehoben.
|
||||
*
|
||||
* @param schedulerPort technischer Scheduler-Mechanismus
|
||||
* @param lockPort OS-Lock auf die Konfigurationsdatei
|
||||
* @param settingsPort Lese-/Schreibzugriff auf Scheduler-Einstellungen
|
||||
* @param batchRunTrigger Auslöser für den eigentlichen Verarbeitungslauf
|
||||
*/
|
||||
public DefaultSchedulerControlUseCase(
|
||||
SchedulerPort schedulerPort,
|
||||
ConfigurationFileLockPort lockPort,
|
||||
SchedulerSettingsPort settingsPort,
|
||||
BatchRunTrigger batchRunTrigger) {
|
||||
this.schedulerPort = Objects.requireNonNull(schedulerPort, "schedulerPort darf nicht null sein");
|
||||
this.lockPort = Objects.requireNonNull(lockPort, "lockPort darf nicht null sein");
|
||||
this.settingsPort = Objects.requireNonNull(settingsPort, "settingsPort darf nicht null sein");
|
||||
this.batchRunTrigger = Objects.requireNonNull(batchRunTrigger, "batchRunTrigger darf nicht null sein");
|
||||
|
||||
SchedulerSettings settings = settingsPort.loadSettings();
|
||||
this.intervalSeconds = Math.max(settings.intervalSeconds(), MINIMUM_INTERVAL_SECONDS);
|
||||
this.statusRef = new AtomicReference<>(SchedulerStatus.initial());
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// SchedulerControlUseCase
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
/**
|
||||
* Startet den automatischen Scheduler.
|
||||
* <p>
|
||||
* Ist der Scheduler bereits aktiv, hat dieser Aufruf keine Wirkung (idempotent).
|
||||
* Schlägt ein Startschritt fehl, wird ein vollständiger Rollback durchgeführt
|
||||
* und der Zustand auf {@link SchedulerState#STOPPED} zurückgesetzt.
|
||||
*
|
||||
* @throws SchedulerStartException wenn der Start fehlschlägt
|
||||
*/
|
||||
@Override
|
||||
public void start() throws SchedulerStartException {
|
||||
SchedulerStatus current = statusRef.get();
|
||||
if (current.state().isActive()) {
|
||||
logger.debug("Scheduler ist bereits aktiv – Start-Aufruf wird ignoriert.");
|
||||
return;
|
||||
}
|
||||
|
||||
// Schritt 1: Zustand auf STARTING setzen
|
||||
statusRef.set(withState(current, SchedulerState.STARTING, Optional.empty()));
|
||||
|
||||
// Schritt 2: scheduler.enabled=true persistieren
|
||||
try {
|
||||
settingsPort.saveEnabled(true);
|
||||
} catch (Exception e) {
|
||||
logger.error("Scheduler konnte nicht gestartet werden: Einstellung nicht persistierbar.", e);
|
||||
statusRef.set(SchedulerStatus.initial());
|
||||
throw new SchedulerStartException(
|
||||
"Scheduler-Einstellung konnte nicht gespeichert werden.", e);
|
||||
}
|
||||
|
||||
// Schritt 3: OS-Lock erwerben
|
||||
try {
|
||||
lockPort.acquireLock();
|
||||
} catch (Exception e) {
|
||||
logger.error("Scheduler konnte nicht gestartet werden: Lock nicht erwerbbar.", e);
|
||||
tryPersistDisabled();
|
||||
statusRef.set(SchedulerStatus.initial());
|
||||
throw new SchedulerStartException(
|
||||
"Konfigurationsdatei konnte nicht gesperrt werden.", e);
|
||||
}
|
||||
|
||||
// Schritt 4: Scheduler-Adapter starten
|
||||
try {
|
||||
schedulerPort.startScheduler(new SchedulerConfig(intervalSeconds), this::executeWrappedTick);
|
||||
} catch (Exception e) {
|
||||
logger.error("Scheduler konnte nicht gestartet werden: Adapter-Start fehlgeschlagen.", e);
|
||||
lockPort.releaseLock();
|
||||
tryPersistDisabled();
|
||||
statusRef.set(SchedulerStatus.initial());
|
||||
throw new SchedulerStartException(
|
||||
"Scheduler-Adapter konnte nicht gestartet werden.", e);
|
||||
}
|
||||
|
||||
// Schritt 5: Zustand auf RUNNING_IDLE setzen
|
||||
Instant nextTick = Instant.now().plusSeconds(intervalSeconds);
|
||||
statusRef.updateAndGet(s -> withState(s, SchedulerState.RUNNING_IDLE, Optional.of(nextTick)));
|
||||
logger.info("Scheduler gestartet. Intervall: {} Sekunden.", intervalSeconds);
|
||||
}
|
||||
|
||||
/**
|
||||
* Stoppt den automatischen Scheduler.
|
||||
* <p>
|
||||
* Ist der Scheduler bereits gestoppt, hat dieser Aufruf keine Wirkung
|
||||
* (idempotent). Läuft gerade ein Batch-Tick, wechselt der Zustand zu
|
||||
* {@link SchedulerState#STOPPING_BATCH_ACTIVE}; der laufende Batch wird
|
||||
* regulär zu Ende geführt, danach wird der Zustand auf
|
||||
* {@link SchedulerState#STOPPED} gesetzt.
|
||||
*/
|
||||
@Override
|
||||
public void stop() {
|
||||
SchedulerStatus previous;
|
||||
SchedulerStatus updated;
|
||||
do {
|
||||
previous = statusRef.get();
|
||||
if (!previous.state().isActive()) {
|
||||
logger.debug("Scheduler ist bereits gestoppt – Stop-Aufruf wird ignoriert.");
|
||||
return;
|
||||
}
|
||||
SchedulerState nextState = previous.state().isBatchRunning()
|
||||
? SchedulerState.STOPPING_BATCH_ACTIVE
|
||||
: SchedulerState.STOPPED;
|
||||
updated = withState(previous, nextState, Optional.empty());
|
||||
} while (!statusRef.compareAndSet(previous, updated));
|
||||
|
||||
boolean batchWasRunning = previous.state().isBatchRunning();
|
||||
schedulerPort.stopScheduler();
|
||||
|
||||
if (!batchWasRunning) {
|
||||
tryPersistDisabled();
|
||||
lockPort.releaseLock();
|
||||
logger.info("Scheduler gestoppt.");
|
||||
} else {
|
||||
logger.info("Stop angefordert – laufender Batch wird abgewartet.");
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Gibt den aktuellen Scheduler-Zustand als unveränderlichen Snapshot zurück.
|
||||
*
|
||||
* @return aktueller Scheduler-Status; nie {@code null}
|
||||
*/
|
||||
@Override
|
||||
public SchedulerStatus getStatus() {
|
||||
return statusRef.get();
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// Tick-Wrapper (package-private für Testbarkeit)
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
/**
|
||||
* Wrapping-Implementierung des {@link BatchRunTrigger}, die der
|
||||
* {@link SchedulerPort} bei jedem Tick synchron aufruft.
|
||||
* <p>
|
||||
* Setzt vor dem Aufruf den Zustand auf
|
||||
* {@link SchedulerState#RUNNING_BATCH_ACTIVE} und leitet nach dem
|
||||
* Abschluss den Folgezustand ab:
|
||||
* <ul>
|
||||
* <li>{@link SchedulerState#RUNNING_IDLE} bei normalem Weiterlauf</li>
|
||||
* <li>{@link SchedulerState#STOPPED} wenn ein Stopp-Befehl empfangen wurde</li>
|
||||
* </ul>
|
||||
* Im Fall {@link SchedulerState#STOPPED} werden außerdem {@code enabled=false}
|
||||
* persistiert und der OS-Lock freigegeben.
|
||||
* <p>
|
||||
* Package-private, damit Unit-Tests den Tick-Wrapper direkt aufrufen können
|
||||
* ohne den Scheduler-Executor zu starten.
|
||||
*
|
||||
* @return das Ergebnis des delegierten {@link BatchRunTrigger#triggerRun()}
|
||||
*/
|
||||
BatchRunTriggerResult executeWrappedTick() {
|
||||
// Zustand auf RUNNING_BATCH_ACTIVE setzen
|
||||
statusRef.updateAndGet(s -> withState(s, SchedulerState.RUNNING_BATCH_ACTIVE, Optional.empty()));
|
||||
|
||||
// Eigentlichen Batch ausführen
|
||||
BatchRunTriggerResult result = batchRunTrigger.triggerRun();
|
||||
|
||||
// Folgezustand aus Ergebnis und aktuellem Zustand ableiten
|
||||
SchedulerStatus afterBatch = statusRef.get();
|
||||
boolean stopping = afterBatch.state() == SchedulerState.STOPPING_BATCH_ACTIVE;
|
||||
SchedulerState nextState = stopping ? SchedulerState.STOPPED : SchedulerState.RUNNING_IDLE;
|
||||
Optional<Instant> nextTickAt = stopping
|
||||
? Optional.empty()
|
||||
: Optional.of(Instant.now().plusSeconds(intervalSeconds));
|
||||
|
||||
Optional<Instant> lastRunEndedAt = afterBatch.lastRunEndedAt();
|
||||
Optional<RunSummary> lastRunSummary = afterBatch.lastRunSummary();
|
||||
Optional<String> lastError = afterBatch.lastError();
|
||||
|
||||
switch (result) {
|
||||
case BatchRunTriggerResult.Started started -> {
|
||||
lastRunEndedAt = Optional.of(started.endedAt());
|
||||
lastRunSummary = Optional.of(started.summary());
|
||||
lastError = Optional.empty();
|
||||
logger.info("Scheduler-Tick abgeschlossen: {} erfolgreich, {} fehlgeschlagen, {} übersprungen.",
|
||||
started.summary().successCount(),
|
||||
started.summary().failedCount(),
|
||||
started.summary().skippedCount());
|
||||
}
|
||||
case BatchRunTriggerResult.SkippedBusy ignored ->
|
||||
logger.debug("Scheduler-Tick übersprungen: anderer Lauf aktiv.");
|
||||
case BatchRunTriggerResult.Failed failed -> {
|
||||
lastError = Optional.of(failed.userMessage());
|
||||
logger.warn("Scheduler-Tick fehlgeschlagen: {}", failed.technicalMessage());
|
||||
}
|
||||
}
|
||||
|
||||
statusRef.set(new SchedulerStatus(
|
||||
nextState,
|
||||
lastRunEndedAt,
|
||||
lastRunSummary,
|
||||
nextTickAt,
|
||||
lastError,
|
||||
afterBatch.autostartFailed()));
|
||||
|
||||
if (stopping) {
|
||||
tryPersistDisabled();
|
||||
lockPort.releaseLock();
|
||||
logger.info("Scheduler gestoppt nach Abschluss des laufenden Batches.");
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// Hilfsmethoden
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
private void tryPersistDisabled() {
|
||||
try {
|
||||
settingsPort.saveEnabled(false);
|
||||
} catch (Exception e) {
|
||||
logger.warn("Fehler beim Zurücksetzen von scheduler.enabled auf false.", e);
|
||||
}
|
||||
}
|
||||
|
||||
private static SchedulerStatus withState(
|
||||
SchedulerStatus base, SchedulerState state, Optional<Instant> nextTickAt) {
|
||||
return new SchedulerStatus(
|
||||
state,
|
||||
base.lastRunEndedAt(),
|
||||
base.lastRunSummary(),
|
||||
nextTickAt,
|
||||
base.lastError(),
|
||||
base.autostartFailed());
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user