Pulsar Ingress

Nstream provides a Pulsar Adapter library that greatly facilitates ingestion from Pulsar topics. This guide demonstrates how to consume from Pulsar topics and process responses in Web Agents using minimal boilerplate.




implementation 'io.nstream:nstream-adapter-pulsar:'




Ideal PulsarIngestingPatch Conditions


Then you simply need to use the PulsarIngestingPatch class, often without overriding any methods. Let’s demonstrate this by configuring an agent to receive messages from a local Pulsar instance, wholly in server.recon:

# server.recon
"provisions": {
  @provision("pulsarClient") {
    class: "nstream.adapter.pulsar.PulsarClientProvision",
    def: {
      "serviceUrl": "pulsar://localhost:6650",
      "connectionsPerBroker": 1
  @provision("consumerConf") {
    class: "nstream.adapter.pulsar.PulsarConsumerConfProvision"
    def: {
      "topicNames": "myTopic",
      "subscriptionName": "mySubscription"

vehicle: @fabric {
  @plane(class: "nstream.adapter.runtime.AppPlane")
  # PulsarIngestingAgent replacement
  @node {
    uri: "/pulsar"
    @agent(class: "nstream.adapter.pulsar.PulsarIngestingPatch") {
      pulsarIngressConf: @pulsarIngressSettings {
        clientProvisionName: "pulsarClient"
        consumerConfProvisionName: "consumerConf"
        contentTypeOverride: "json"
        relaySchema: @command {
          nodeUri: {
          laneUri: "addMessage"
          value: $value

# Configure desired web settings (e.g. port, websocket compression)
# ...


Client and Consumer

As a first step, a singleton PulsarClientProvision must be configured so a consumer may be instantiated against it. The above snippet is minimal, and an exhaustive set of config options may be found here.

The consumer configuration is handled the same way.


The previous configurations can be seen as “connection-oriented”. This part is “ingestion-oriented”, responsible for both:

Common Variations

Custom Ingest Logic

If relaying messages to another agent is not sufficient, or you would like to add custom logic on ingestion of messages, the PulsarIngestingPatch can be easily overridden. Create a new class that extends the PulsarIngestingPatch and override the ingest method.

// CustomIngestingAgent.java
public class CustomIngestingAgent<V> extends PulsarIngestingPatch<V> {

  protected void ingest(Message<V> unstructured) throws DeferrableException {
    // Custom ingestion of an individual message


Remember to update your server.recon file with the new agent you just created, instead of the PulsarIngestingPatch.

Manual Acknowledgements

By default, messages are acknowledged as soon as they are consumed, without waiting upon the results of any processing by downstream Web Agents. Future nstream-toolkit releases will introduce some convenient abstractions over the most common patterns that require finer control, so stay tuned! In the meantime, please feel free to share any of your ideas.