Package app.l2nx.gs.adapter.core.commands

Class CommandsConsumer

java.lang.Object
app.l2nx.gs.adapter.core.commands.CommandsConsumer
public final class CommandsConsumer extends Object
Inbound commands consumer + dispatcher. Owns one Kafka Consumer on a single daemon thread nx-commands-consumer. Polls commandsTopic, resolves handlers via CommandTypeRegistry, deserializes via Gson, invokes CommandHandler.handle(C, app.l2nx.gs.adapter.api.spi.CommandContext), publishes the CommandResult reply to commandsRepliesTopic, awaits the per-batch reply-flush, then commits the batch offset.

Threading model — CommandHandler.handle(C, app.l2nx.gs.adapter.api.spi.CommandContext) runs synchronously on the consumer thread; game-state mutations require an explicit HostExecutor.sync(Runnable) hop bounded by l2nx.commands.host-sync-timeout-ms. Slow handlers slow this consumer.

At-least-once contract on both legs — for every batch:

  1. Dispatch all records, issuing reply sends asynchronously.
  2. Await up to l2nx.commands.reply-flush-timeout-ms for the producer's send-callbacks to fire (i.e. broker has acknowledged or definitively failed). This bounds the at-most-once reply window: a JVM crash before the flush completes leaves the batch uncommitted, so the records redeliver and the handlers re-emit the replies (handlers must be idempotent).
  3. commitSync the consumer offset.

Error boundaries (per dispatch step):

  • Unknown Nx-Message-Type → reply ErrorCode.UNSUPPORTED_COMMAND, commit.
  • Gson JsonSyntaxException → reply ErrorCode.VALIDATION_FAILED, commit.
  • Handler HostExecutorTimeoutException → reply ErrorCode.UNAVAILABLE with error.cause = "host-executor-timeout", commit (handler hop did not finish in time; web side may retry).
  • Handler RuntimeException → reply ErrorCode.INTERNAL_ERROR with class+message in errorDetails, commit.
  • Handler returns null → reply ErrorCode.INTERNAL_ERROR with error.cause = "handler-returned-null", commit.
  • Handler Throwable (OOM / Error) → log ERROR, do NOT reply, do NOT commit, rethrow (consumer thread aborts; JVM-level handler decides next).

Reply path bypasses the events publisher's queue and goes directly to the Kafka producer via the supplied CommandsConsumer.ReplySender bridge — Kafka's own record-accumulator absorbs back-pressure, and reply records have different drop semantics from outbound events (a dropped reply means web-side timeout, no semantic recovery).

  • Nested Class Summary

    Nested Classes
    Modifier and Type
    Class
    Description
    static interface 
    CommandsConsumer.ReplySender
    Bridge to the actual Kafka send.

  • Method Summary

    Modifier and Type
    Method
    Description
    app.l2nx.gs.adapter.api.kafka.ops.ModuleStatus
    currentStatus()
    Heartbeat slot snapshot for the commands module.
    void
    start()
    Spawn the consumer daemon and subscribe to inboundTopic.
    void
    stop()
    Signal the daemon to stop, wake it from any blocking poll, await join up to shutdownTimeoutMs, then close the Kafka consumer.

    Methods inherited from class java.lang.Object

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

  • Method Details

    • start

      public void start()
      Spawn the consumer daemon and subscribe to inboundTopic. Idempotent.

    • stop

      public void stop()
      Signal the daemon to stop, wake it from any blocking poll, await join up to shutdownTimeoutMs, then close the Kafka consumer. Idempotent.

    • currentStatus

      public app.l2nx.gs.adapter.api.kafka.ops.ModuleStatus currentStatus()
      Heartbeat slot snapshot for the commands module.