CEP Client

CEPClient is a Java library that attaches to the real-time rule engine. It deploys EQL rules, pushes in events, and receives callbacks inside your own process when rules trigger. It does not poll REST.

Item	Value
Coordinates	plantpulse-platform:plantpulse-cep-client
Server	plantpulse-cep — default port 7401 (TLS)
Requirements	Java 21
Authentication	X-API-Key header
Threading	Thread-safe · AutoCloseable

It talks in two directions

Rule deployment and lookup go over HTTPS, while event send/receive goes over the in-memory bus (Valkey/Redis) shared with the engine. That is why you must also attach InMemoryClient for matching callbacks to arrive.

Getting Started

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

If you omit shutdown(), the process will not exit

The in-memory bus starts a non-daemon thread. Even if you close CEPClient, that thread remains, preventing the JVM from terminating. This is especially noticeable in batch jobs.

What You Can Do

Method	What It Does
deploy(EQL, listener)	Deploy rules + register matching callbacks
validate(EQL)	Compile only on the server — does not deploy
undeploy(name)	Undeploy rules + remove listeners
isDeployed(name) · getEQL(name) · getText(name)	Query status
getEQLMap() · getEQLMapCount()	All deployed rules
sendEvent(type, json)	Send events (single · batch)
setApiKey(key)	Rotate keys at runtime

If the key is null or empty, no header is attached — compatible with servers that have authentication disabled.

Tuning Bulk Sends

sendEvent batches internally before sending. Tune it when pushing tens of thousands of events per second.

Parameter	Default	Meaning
shardCount	4	Number of flush workers
queueSize	600,000	Overall queue size
flushTriggerSize	8,000	Flush when this size is reached
maxInflightBatches	64	Max concurrent flushes
batchIntervalMs	50	Time-based flush interval
skipResult	true	Does not wait for results (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();

What happens when the queue fills up

If you exceed queueSize, new events are rejected. Before increasing it, first look at why things are not draining — usually it is not the engine but a heavy rule.

Handling Failures

Exception	When
EPCompileException	EQL compilation failure (validate · deploy)
CEPClientClosedException	Used after close()
IllegalArgumentException	Invalid arguments — empty rule name, empty EQL
RuntimeException	Server 4xx/5xx (including 401 auth failure), network errors

See Also

Data Gateway Client · CEP Component · EQL Usage