March 7 - June 27, 2024 - Nstream is hitting the road with Confluent on the #DataInMotionTour! / Learn More

NATS Ingress

Overview

NATS is a lightweight, high performance messaging system that makes it easy to communicate between distributed systems. By integrating NATS with Nstream, users can efficiently stream data into Nstream for real-time processing and analytics.

Agenda of this guide:

  1. Provides instructions on setting up NATS as an ingress data source for Nstream.
  2. Provides a NATS Adapter library that greatly facilitates ingestion from NATS Streams and corresponding Subjects.
  3. Demonstrates how to consume from NATS Subjects and process responses in Web Agents using minimal boilerplate.

Prerequisites

Dependencies

Gradle

implementation 'io.nstream:nstream-adapter-nats:4.9.16'

Maven

<dependency>
  <groupId>io.nstream</groupId>
  <artifactId>nstream-adapter-nats</artifactId>
  <version>4.9.16</version>
  <type>module</type>
</dependency>

Glossary

Ideal NatsIngestingPatch Conditions

If:

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

# server.recon
"provisions": {
  @provision("natsConnectionProvision") {
    class: "nstream.adapter.nats.NatsConnectionProvision",
    def: {
      "servers": "nats-adapter-broker:4222",
    }
  }
}

"nats-adapter": @fabric {
  @plane(class: "nstream.adapter.runtime.AppPlane")
  # NatsIngestingAgent replacement
  @node {
    uri: "/ingesting"
    @agent(class: "nstream.adapter.nats.NatsIngestingPatch") {
      natsIngressConf: @natsIngressSettings {
        firstFetchDelayMillis: 5000,
        fetchIntervalMillis: 20000,
        natsIngressStreamName: "NatsStream",
        natsIngressSubjectName: "NatsSubject"
        contentTypeOverride: json
        relaySchema: @command {
          nodeUri: {
            "/dynamic/",
            $val + 1
          },
          laneUri: "unused"
          value:
        }
      }
    }
  }
}

# Configure desired web settings (e.g. port)
# ...

Configuration

Connection and Consumer

As a first step, a singleton NatsConnectionProvision 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.

NatsIngressSettings

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 NatsIngestingPatch can be easily overridden. Create a new class that extends the NatsIngestingPatch and override the ingest method.

// CustomIngestingAgent.java
public class CustomIngestingAgent extends NatsIngestingPatch {
  @Override
  protected void ingest(Message 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 NatsIngestingPatch.


Nstream is licensed under the Redis Source Available License 2.0 (RSALv2).