# file: listener.aus

## class: KafkaListener

[41:14] (extern: com.lehman.aussom.KafkaListener) **extends: object** 

The Aussom Server ClientListener connector for Kafka consumer
groups. Configure it, then hand it to the server with
app.registerListener(listener): the runtime gives it a dedicated
thread, drives its lifecycle across reloads and shutdown, and
applies the auto-restart policy from applications.yaml (keyed by
the listener name).
lst = new KafkaListener("orders.created");
lst.setBootstrap("broker1:9092");
lst.setTopics([ "orders.created" ]);
lst.setOnMessage(::onOrder);
app.registerListener(lst);
The onMessage handler runs on the listener thread with one
KafkaRecord argument. While a batch's handlers run no new records
are pulled - one poll batch must finish inside
max.poll.interval.ms (default 5 minutes) or the group evicts the
listener; slow handlers should lower setMaxPollRecords or raise
setMaxPollIntervalMs.

#### Methods

- **KafkaListener** (`string Name`)

	> Creates a listener. The name is the stable identifier used in logs and per-listener overrides in applications.yaml, and the default consumer group id.

	- **@p** `Name` is a string with the listener name.


- **newListener** (`string Name`)


- **setBootstrap** (`string Servers`)

	> Sets the bootstrap address.

	- **@p** `Servers` is a string with host:port[,host:port...].
	- **@r** `this` object


- **setGroupId** (`string Id`)

	> Overrides the consumer group id (the listener name by default).

	- **@p** `Id` is a string with the group id.
	- **@r** `this` object


- **setTopics** (`list Topics`)

	> Subscribes to topics. Set exactly one of setTopics / setTopicPattern.

	- **@p** `Topics` is a list of topic name strings.
	- **@r** `this` object


- **setTopicPattern** (`string Regex`)

	> Subscribes to every topic matching a regex pattern.

	- **@p** `Regex` is a string with the topic pattern.
	- **@r** `this` object


- **setAutoOffsetReset** (`string Mode`)

	> Sets where to start when the group has no committed offset.

	- **@p** `Mode` is a string: earliest, latest (default), or none.
	- **@r** `this` object


- **setPollTimeoutMs** (`int Ms`)

	> Sets the poll wait, which is also the stop/pause reaction bound. Default 1000.

	- **@p** `Ms` is an int with the poll timeout in milliseconds.
	- **@r** `this` object


- **setMaxPollRecords** (`int N`)

	> Caps how many records one poll returns - and with it the work one batch's handlers must finish inside the poll interval.

	- **@p** `N` is an int with the record cap.
	- **@r** `this` object


- **setMaxPollIntervalMs** (`int Ms`)

	> Sets the processing deadline between polls before the group evicts this listener. Default 300000 (5 minutes).

	- **@p** `Ms` is an int with the interval in milliseconds.
	- **@r** `this` object


- **setKeyFormat** (`string Fmt`)

	> Sets the key format.

	- **@p** `Fmt` is a string: string (default) or bytes (Buffer).
	- **@r** `this` object


- **setValueFormat** (`string Fmt`)

	> Sets the value format.

	- **@p** `Fmt` is a string: string (default) or bytes (Buffer).
	- **@r** `this` object


- **setConfig** (`string Key, Value`)

	> Sets any consumer config key directly (security included; also applied to the dead-letter producer). The ssl.*.location keys are rejected here - use the dedicated setters.

	- **@p** `Key` is a string with the config key.
	- **@p** `Value` is the config value.
	- **@r** `this` object


- **setConfigs** (`map Cfg`)

	> Applies a whole config map at once.

	- **@p** `Cfg` is a map of config key to value.
	- **@r** `this` object


- **setSslKeystoreFile** (`string Path`)

	> Sets the TLS keystore file. Under Aussom Server the path is resolved through the app_data sandbox.

	- **@p** `Path` is a string with the app-relative keystore path.
	- **@r** `this` object


- **setSslTruststoreFile** (`string Path`)

	> Sets the TLS truststore file. Under Aussom Server the path is resolved through the app_data sandbox.

	- **@p** `Path` is a string with the app-relative truststore path.
	- **@r** `this` object


- **setCommitMode** (`string Mode`)

	> Sets the commit mode.

	- **@p** `Mode` is a string: batch (default; one commit per cleanly dispatched poll batch), record (commit after every record), or auto (client auto-commit).
	- **@r** `this` object


- **setErrorMode** (`string Mode`)

	> Sets what happens when a handler fails.

	- **@p** `Mode` is a string: continue (default; skip the record and keep going), seek (rewind so it redelivers - beware poison records), or stop (exit; the runtime's auto-restart policy takes over).
	- **@r** `this` object


- **setDeadLetterTopic** (`string Topic`)

	> Copies failed records to a dead-letter topic (same key, value, and headers, plus dlt.original.topic/partition/offset and dlt.error headers) before skipping them. If the copy itself fails the loop stops so the record is never lost. Per-record dispatch only.

	- **@p** `Topic` is a string with the dead-letter topic name.
	- **@r** `this` object


- **setOnMessage** (`callback Cb`)

	> Sets the per-record handler, called on the listener thread with one KafkaRecord argument. Set exactly one of setOnMessage / setOnBatch.

	- **@p** `Cb` is a callback taking (object Rec).
	- **@r** `this` object


- **setOnBatch** (`callback Cb`)

	> Sets the batch handler, called once per poll with the whole batch. Error modes then apply at batch granularity.

	- **@p** `Cb` is a callback taking (list Recs).
	- **@r** `this` object


- **setOnError** (`callback Cb`)

	> Sets an optional error handler, called with the error text when a handler fails.

	- **@p** `Cb` is a callback taking (string ErrText).
	- **@r** `this` object


- **setOnAssigned** (`callback Cb`)

	> Sets an optional hook called with the newly assigned partitions (a list of topic-partition maps) when the group rebalances.

	- **@p** `Cb` is a callback taking (list Partitions).
	- **@r** `this` object


- **setOnRevoked** (`callback Cb`)

	> Sets an optional hook called with the revoked partitions when the group rebalances. Processed offsets are committed before it fires.

	- **@p** `Cb` is a callback taking (list Partitions).
	- **@r** `this` object


- **getListenerName** ()

	> Gets the listener name.

	- **@r** `A` string with the listener name.


- **getIsRunning** ()

	> Reports whether the poll loop is running.

	- **@r** `A` bool with true while running.


- **getIsPaused** ()

	> Reports whether back-pressure pause is requested.

	- **@r** `A` bool with true while paused.


- **startListener** ()

	> Runs the poll loop on the calling thread, blocking until stopListener() is called. Under Aussom Server the runtime calls this itself after app.registerListener() - apps only call it directly for plain-CLI use (typically on a Thread).

	- **@r** `this` object


- **stopListener** ()

	> Signals the poll loop to exit. Safe to call from any thread, any number of times.

	- **@r** `this` object


- **pauseListener** ()

	> Pauses record dispatch without leaving the group (the loop keeps polling empty batches, so it keeps heartbeating).

	- **@r** `this` object


- **resumeListener** ()

	> Resumes record dispatch after pauseListener().

	- **@r** `this` object




## class: KafkaShareListener

[296:14] (extern: com.lehman.aussom.KafkaShareListener) **extends: object** 

The share-group ClientListener connector: queue semantics
(broker 4.2+). Same lifecycle as KafkaListener, but
acknowledgement is per record - accept on handler success,
release or reject per the error mode on failure. The broker's
delivery-count limit archives records that keep failing.
lst = new KafkaShareListener("orders.workers");
lst.setBootstrap("broker1:9092");
lst.setTopics([ "orders.created" ]);
lst.setOnMessage(::onOrder);
app.registerListener(lst);

#### Methods

- **KafkaShareListener** (`string Name`)

	> Creates a share listener. The name is the stable identifier used in logs and applications.yaml overrides, and the default share group id.

	- **@p** `Name` is a string with the listener name.


- **newListener** (`string Name`)


- **setBootstrap** (`string Servers`)

	> Sets the bootstrap address.

	- **@p** `Servers` is a string with host:port[,host:port...].
	- **@r** `this` object


- **setGroupId** (`string Id`)

	> Overrides the share group id (the listener name by default).

	- **@p** `Id` is a string with the group id.
	- **@r** `this` object


- **setTopics** (`list Topics`)

	> Subscribes to topics. Share groups have no pattern subscribe.

	- **@p** `Topics` is a list of topic name strings.
	- **@r** `this` object


- **setPollTimeoutMs** (`int Ms`)

	> Sets the poll wait, which is also the stop/pause reaction bound. Default 1000.

	- **@p** `Ms` is an int with the poll timeout in milliseconds.
	- **@r** `this` object


- **setKeyFormat** (`string Fmt`)

	> Sets the key format.

	- **@p** `Fmt` is a string: string (default) or bytes (Buffer).
	- **@r** `this` object


- **setValueFormat** (`string Fmt`)

	> Sets the value format.

	- **@p** `Fmt` is a string: string (default) or bytes (Buffer).
	- **@r** `this` object


- **setConfig** (`string Key, Value`)

	> Sets any share consumer config key directly. The ssl.*.location keys are rejected here - use the dedicated setters.

	- **@p** `Key` is a string with the config key.
	- **@p** `Value` is the config value.
	- **@r** `this` object


- **setConfigs** (`map Cfg`)

	> Applies a whole config map at once.

	- **@p** `Cfg` is a map of config key to value.
	- **@r** `this` object


- **setSslKeystoreFile** (`string Path`)

	> Sets the TLS keystore file. Under Aussom Server the path is resolved through the app_data sandbox.

	- **@p** `Path` is a string with the app-relative keystore path.
	- **@r** `this` object


- **setSslTruststoreFile** (`string Path`)

	> Sets the TLS truststore file. Under Aussom Server the path is resolved through the app_data sandbox.

	- **@p** `Path` is a string with the app-relative truststore path.
	- **@r** `this` object


- **setErrorMode** (`string Mode`)

	> Sets what happens when the handler fails.

	- **@p** `Mode` is a string: release (default; the record redelivers, bounded by the broker's delivery-count limit), reject (drop as poison), or stop (exit; auto-restart takes over).
	- **@r** `this` object


- **setOnMessage** (`callback Cb`)

	> Sets the per-record handler, called on the listener thread with one KafkaRecord argument.

	- **@p** `Cb` is a callback taking (object Rec).
	- **@r** `this` object


- **setOnError** (`callback Cb`)

	> Sets an optional error handler, called with the error text when the message handler fails.

	- **@p** `Cb` is a callback taking (string ErrText).
	- **@r** `this` object


- **getListenerName** ()

	> Gets the listener name.

	- **@r** `A` string with the listener name.


- **getIsRunning** ()

	> Reports whether the poll loop is running.

	- **@r** `A` bool with true while running.


- **getIsPaused** ()

	> Reports whether back-pressure pause is requested.

	- **@r** `A` bool with true while paused.


- **startListener** ()

	> Runs the poll loop on the calling thread, blocking until stopListener() is called.

	- **@r** `this` object


- **stopListener** ()

	> Signals the poll loop to exit. Safe to call from any thread, any number of times.

	- **@r** `this` object


- **pauseListener** ()

	> Pauses record dispatch (the loop stops polling; share-group records are only locked while acquired, so nothing is held back from other members).

	- **@r** `this` object


- **resumeListener** ()

	> Resumes record dispatch after pauseListener().

	- **@r** `this` object




