CEP-Client

CEPClient ist eine Java-Bibliothek für die Echtzeit-Regelengine. Sie verteilt EQL-Regeln, schiebt Ereignisse ein und empfängt Callbacks in Ihrem Prozess, wenn eine Regel zutrifft. Sie pollt nicht über REST.

Element	Wert
Koordinaten	plantpulse-platform:plantpulse-cep-client
Server	plantpulse-cep — Standardport 7401 (TLS)
Anforderung	Java 21
Authentifizierung	X-API-Key-Header
Threads	Thread-sicher · AutoCloseable

Zwei Wege der Kommunikation

Die Verteilung und Abfrage von Regeln erfolgt über HTTPS, der Ereignisversand und -empfang läuft über den mit der Engine geteilten In-Memory-Bus (Valkey/Redis). Daher müssen Sie InMemoryClient mit anbinden, damit Matching-Callbacks eintreffen.

Erste Schritte

import java.util.Properties;
import plantpulse.cep.client.CEPClient;
import plantpulse.cep.client.eql.EQL;
import plantpulse.cep.client.eql.EQLUpdateListener;
import plantpulse.inmemory.InMemoryClient;
import plantpulse.inmemory.redis.RedisInMemoryClient;
import plantpulse.json.JSONObject;

Properties redis = new Properties();
redis.setProperty("cache.host", "plant-server.example.com");
redis.setProperty("cache.port", "6379");
redis.setProperty("cache.password", System.getenv("PP_CACHE_PASSWORD"));

RedisInMemoryClient bus = RedisInMemoryClient.builder().fromProperties(redis).buildClient();
bus.connect();

try (CEPClient client = CEPClient.builder()
        .baseUrl("https://plant-server.example.com:7401")
        .apiKey(System.getenv("PP_CEP_API_KEY"))
        .inmemory(bus)                       // build() 가 연결까지 합니다
        .build()) {

    EQL eql = new EQL("HIGH_TEMPERATURE",
            "SELECT * FROM Point WHERE tag_id = 'TAG_001' AND value > 80");
    EQLUpdateListener listener =
            (name, matches) -> System.out.println(name + " → " + matches);

    JSONObject deployed = client.deploy(eql, listener);
    if (!deployed.getBoolean("success")) {
        System.err.println("배포 실패: " + deployed.optString("message"));
        return;
    }

    JSONObject point = new JSONObject();
    point.put("tag_id", "TAG_001");
    point.put("timestamp", System.currentTimeMillis());
    point.put("value", "97.4");
    client.sendEvent("Point", point);

    Thread.sleep(1_000L);
    client.undeploy("HIGH_TEMPERATURE");
} finally {
    // Redisson 이 논데몬 스레드를 붙잡습니다 — 반드시 명시적으로 내립니다.
    bus.shutdown();
}

Wenn Sie shutdown() vergessen, endet der Prozess nicht

Der In-Memory-Bus startet einen Nicht-Daemon-Thread. Selbst wenn Sie CEPClient schließen, bleibt dieser Thread bestehen und verhindert das Beenden der JVM. Besonders auffällig bei Batch-Jobs.

Was Sie tun können

Methode	Funktion
deploy(EQL, listener)	Regel verteilen + Matching-Callback registrieren
validate(EQL)	Nur serverseitig kompilieren — nicht verteilen
undeploy(name)	Regel aufheben + Listener entfernen
isDeployed(name) · getEQL(name) · getText(name)	Status abfragen
getEQLMap() · getEQLMapCount()	Alle verteilten Regeln
sendEvent(type, json)	Ereignis senden (einzeln · batchweise)
setApiKey(key)	Schlüssel zur Laufzeit austauschen

Wenn der Schlüssel null oder leer ist, wird kein Header angehängt — kompatibel mit Servern, bei denen die Authentifizierung deaktiviert ist.

Tuning für Massenversand

sendEvent sammelt intern und sendet gebündelt. Passen Sie dies an, wenn Sie Zehntausende Ereignisse pro Sekunde einspeisen.

Parameter	Standardwert	Bedeutung
shardCount	4	Anzahl der Flush-Worker-Threads
queueSize	600.000	Gesamtgröße der Warteschlange
flushTriggerSize	8.000	Bei dieser Größe wird geflusht
maxInflightBatches	64	Anzahl gleichzeitig laufender Flush-Vorgänge
batchIntervalMs	50	Zeitbasierter Flush-Intervall
skipResult	true	Kein Warten auf Ergebnisse (fire-and-forget)

CEPClient client = CEPClient.builder()
        .host("plant-server.example.com")
        .apiKey(System.getenv("PP_CEP_API_KEY"))
        .inmemory(bus)
        .shardCount(8)
        .flushBatchSize(16_000)
        .queueSize(1_000_000)
        .build();

Was passiert, wenn die Warteschlange voll ist

Wenn queueSize überschritten wird, werden neue Ereignisse abgelehnt. Bevor Sie die Größe erhöhen, prüfen Sie zuerst, warum nichts abfließt — meistens ist nicht die Engine das Problem, sondern die Regeln sind zu schwer.

Umgang mit Fehlern

Ausnahme	Wann
EPCompileException	EQL-Kompilierung fehlgeschlagen (validate · deploy)
CEPClientClosedException	Verwendung nach close()
IllegalArgumentException	Ungültige Argumente — leerer Regelname, leeres EQL
RuntimeException	Server 4xx/5xx (einschließlich Authentifizierungsfehler 401), Netzwerkfehler

Siehe auch

Data Gateway-Client · CEP-Komponente · EQL-Verwendung