Package app.l2nx.gs.adapter.core.events

Class EventsPublisher

java.lang.Object

app.l2nx.gs.adapter.core.events.EventsPublisher

public final class EventsPublisher
extends Object

Bounded-queue + single-daemon-thread fan-out for outbound events. Caller threads (NxEventsImpl.publishX(...)) call enqueue(EventEnvelope) which is O(1) and never blocks on Kafka latency. The daemon nx-events-publisher drains the queue, builds a ProducerRecord, stamps the Nx-Message-Type header, and hands off to the supplied EventsPublisher.Sender (production: NxKafka.sendBytesKeyRecord).

Drop policy on full queue:

• EventsPublisher.DropPolicy.NEWEST (default) — drops the incoming envelope atomically via queue.offer() returning false; no eviction race under multi-threaded producers.
• EventsPublisher.DropPolicy.OLDEST — evicts the head and admits the new envelope so recent facts displace stale snapshots. Head-poll and newcomer-offer are not atomic, so concurrent producers on a full queue may over-count droppedTotal (an enqueue can both evict a head and lose its slot to another caller).

Counters expose the publisher's health via currentStatus() for the heartbeat events module slot.

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static enum	EventsPublisher.DropPolicy
static interface	EventsPublisher.ProducerFlusher	Bridge to a synchronous producer flush.
static interface	EventsPublisher.Sender	Bridge to the actual Kafka send.

Constructor Summary

Constructors

Constructor	Description
EventsPublisher(@Nullable Map<String,String> familyTopics, EventsPublisher.Sender sender, EventsConfig config, app.l2nx.gs.adapter.core.events.EventTypeRegistry registry)
EventsPublisher(@Nullable Map<String,String> familyTopics, EventsPublisher.Sender sender, EventsPublisher.ProducerFlusher producerFlusher, EventsConfig config, app.l2nx.gs.adapter.core.events.EventTypeRegistry registry)

Method Summary

Modifier and Type	Method	Description
app.l2nx.gs.adapter.api.kafka.ops.ModuleStatus	currentStatus()	Build a heartbeat slot snapshot.
void	enqueue(@Nullable app.l2nx.gs.adapter.core.events.EventEnvelope envelope)	Append an envelope to the publish queue.
boolean	flush(long timeoutMs)	Synchronously drain the queue into the sender and block on a producer flush until in-flight records reach the broker, or timeoutMs elapses.
boolean	isFamilyEnabled(String familyKey)	True when the family key has a non-empty topic configured.
void	stop()	Stop the publisher.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Details

EventsPublisher

public EventsPublisher(@Nullable Map<String,String> familyTopics,
 EventsPublisher.Sender sender,
 EventsConfig config,
 app.l2nx.gs.adapter.core.events.EventTypeRegistry registry)

EventsPublisher

public EventsPublisher(@Nullable Map<String,String> familyTopics,
 EventsPublisher.Sender sender,
 EventsPublisher.ProducerFlusher producerFlusher,
 EventsConfig config,
 app.l2nx.gs.adapter.core.events.EventTypeRegistry registry)

Method Details

stop

public void stop()

Stop the publisher. Signals the daemon, waits up to shutdownDrainMs for in-flight envelopes to drain, then cancels. Idempotent.

enqueue

public void enqueue(@Nullable app.l2nx.gs.adapter.core.events.EventEnvelope envelope)

Append an envelope to the publish queue. On full queue, applies the configured EventsPublisher.DropPolicy; never blocks longer than the ArrayBlockingQueue.offer() fast path.

Drop counter accounts for every lost envelope, including races. With EventsPublisher.DropPolicy.OLDEST and concurrent callers, an enqueue may evict the head AND lose its own envelope to another caller filling the freed slot first — both losses are counted (droppedTotal += 2).

flush

public boolean flush(long timeoutMs)

Synchronously drain the queue into the sender and block on a producer flush until in-flight records reach the broker, or timeoutMs elapses. Does NOT stop the daemon — this is a flush, not a shutdown; the daemon may drain concurrently (both feed the same thread-safe sender).

Never throws (game-exit safety) — a flusher failure is logged and reported as false.

Returns:

true if the queue emptied and the producer flush returned within the budget; false on timeout (records may still be in flight)

currentStatus

public app.l2nx.gs.adapter.api.kafka.ops.ModuleStatus currentStatus()

Build a heartbeat slot snapshot.

State semantics: ACTIVE when the daemon is running, DISABLED when it has not been started or has been stopped. A future enhancement may surface DEGRADED via a rolling failure-ratio window; operators derive degradation from the raw counters today.

isFamilyEnabled

public boolean isFamilyEnabled(String familyKey)

True when the family key has a non-empty topic configured.