CEPクライアント

CEPClient は リアルタイムルールエンジン に接続するJavaライブラリです。EQLルールを配布し、 イベントをプッシュし、ルールが発動したときに 自分のプロセス内で コールバックを受け取ります。RESTを ポーリングすることはありません。

項目	値
座標	plantpulse-platform:plantpulse-cep-client
サーバー	plantpulse-cep — デフォルトポート 7401 (TLS)
要件	Java 21
認証	X-API-Key ヘッダー
スレッド	スレッドセーフ · AutoCloseable

2つの経路で通信します

ルール配布・照会は HTTPS で、イベント送受信はエンジンと共有する インメモリバス (Valkey/Redis) を経由します。そのため InMemoryClient を一緒に付けないとマッチングコールバックが届きません。

はじめに

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();
}

shutdown() を忘れるとプロセスが終了しません

インメモリバスが 非デーモンスレッド を起動します。CEPClient を閉じてもそのスレッドは残り、 JVMが終了しません。バッチジョブで特に顕著です。

できること

メソッド	処理内容
deploy(EQL, listener)	ルール配布 + マッチングコールバック登録
validate(EQL)	サーバーでコンパイルのみ実行 — 配布しない
undeploy(name)	ルール解除 + リスナー削除
isDeployed(name) · getEQL(name) · getText(name)	ステータス照会
getEQLMap() · getEQLMapCount()	配布済みルール全体
sendEvent(type, json)	イベント送信 (単発 · バッチ)
setApiKey(key)	ランタイムでのキー交換

キーが null の場合、または空の場合はヘッダーを付けません — 認証を無効にしたサーバーとも互換性があります。

一括送信チューニング

sendEvent は内部的にまとめて送信します。毎秒数万件をプッシュする際に調整します。

パラメータ	デフォルト値	意味
shardCount	4	flushワーカースレッド数
queueSize	600,000	全体キューサイズ
flushTriggerSize	8,000	このサイズに達するとflush
maxInflightBatches	64	同時に進行可能なflush数
batchIntervalMs	50	時間ベースのflush間隔
skipResult	true	結果を待たない(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();

キューが満杯になったらどうなるか

queueSize を超えると新しいイベントが拒否されます。増やす前に なぜ排出されないのか を先に 確認してください — たいていはエンジンではなくルールが重いのです。

失敗の扱い方

例外	発生時
EPCompileException	EQLコンパイル失敗 (validate · deploy)
CEPClientClosedException	close() の後に使用
IllegalArgumentException	不正な引数 — 空のルール名、空のEQL
RuntimeException	サーバー4xx/5xx(認証失敗401を含む)、ネットワークエラー

関連情報

Data Gatewayクライアント · CEPコンポーネント · EQLの使い方