Reactive Streams

Reactive Streams enable TCP-based, backpressured communication between thatDot products. Use them to connect Quine Enterprise and Novelty for bidirectional data processing pipelines.

Role	Description	Use Case
Publisher (Server)	Binds to a port and broadcasts data to connected subscribers	Standing query outputs, observation results
Subscriber (Client)	Connects to a publisher to receive streamed data	Ingest from another product's output

---

ReactiveStream as Ingest Source

The ReactiveStream ingest source connects as a client to an existing reactive stream publisher. Use this to ingest data from another product's standing query or observation output.

Example

In this example, we configure Quine Enterprise to ingest data from a reactive stream server running on port 9002.

Registering via the API

To register a ReactiveStream ingest via Create Ingest Stream: POST /api/v2/graph/{graphName}/ingests:

{
  "name": "from-reactive-stream",
  "source": {
    "type": "ReactiveStream",
    "url": "localhost",
    "port": 9002,
    "format": {
      "type": "Json"
    }
  },
  "query": "CREATE (n:Event $that)",
  "parameter": "that",
  "parallelism": 1
}

Using a Recipe

The same ingest stream defined in a Recipe:

ingestStreams:
  - name: reactive-stream-ingest
    source:
      type: ReactiveStream
      url: localhost
      port: 9002
    query: |-
      MATCH (n)
      WHERE id(n) = idFrom($that.id)
      SET n = $that

Ingest Source Configuration Reference

Parameter	Type	Required	Description
type	string	Yes	Must be "ReactiveStream"
url	string	Yes	Hostname of the reactive stream server to connect to
port	integer	Yes	Port of the reactive stream server
format	object	Yes	Record format (e.g., Json, CypherJson). See Record Formats for the full list and the difference between Json and Raw.

---

ReactiveStream as Output Destination

The ReactiveStream output destination creates a server that broadcasts results to connected subscribers. Use this to publish standing query results for downstream processing by other products.

Example

Configure a standing query to output results to a reactive stream on port 9001:

{
  "name": "events-to-downstream",
  "pattern": {
    "type": "Cypher",
    "query": "MATCH (n:Event) RETURN DISTINCT id(n) AS id",
    "mode": "DISTINCT_ID"
  },
  "outputs": [
    {
      "name": "to-reactive-stream",
      "destinations": [
        {
          "type": "ReactiveStream",
          "address": "0.0.0.0",
          "port": 9001,
          "format": {
            "type": "JSON"
          }
        }
      ]
    }
  ]
}

Output Destination Configuration Reference

Parameter	Type	Required	Default	Description
type	string	Yes	—	Must be "ReactiveStream"
address	string	No	localhost	Address to bind the reactive stream server
port	integer	Yes	—	Port to bind the reactive stream server
format	object	Yes	—	Output format (JSON or Protobuf)

Cluster Limitation

Reactive Stream outputs do not function correctly when running in a cluster. Use Kafka or Kinesis for clustered deployments.

---

Connecting Quine Enterprise and Novelty

Reactive Streams enable powerful bidirectional data pipelines between Quine Enterprise and Novelty. This pattern allows you to:

1. Process graph data in Quine Enterprise
2. Stream results to Novelty for anomaly detection
3. Feed Novelty's observations back into Quine Enterprise for further analysis

Architecture

┌─────────────────────┐                    ┌─────────────────────┐
│   Quine Enterprise  │                    │       Novelty       │
│                     │                    │                     │
│  ┌───────────────┐  │   ReactiveStream   │  ┌───────────────┐  │
│  │ Standing Query├──┼──── port 9001 ────►│──┤    Ingest     │  │
│  └───────────────┘  │                    │  └───────┬───────┘  │
│                     │                    │          │          │
│                     │                    │          ▼          │
│                     │                    │  ┌───────────────┐  │
│  ┌───────────────┐  │   ReactiveStream   │  │  Observations │  │
│  │    Ingest     │◄─┼──── port 9002 ─────┤──┤    Output     │  │
│  └───────────────┘  │                    │  └───────────────┘  │
│                     │                    │                     │
└─────────────────────┘                    └─────────────────────┘

Step 1: Quine Enterprise Standing Query Output

Configure Quine Enterprise to output standing query results to a reactive stream:

{
  "name": "events-to-novelty",
  "pattern": {
    "type": "Cypher",
    "query": "MATCH (n:Event) RETURN DISTINCT id(n) AS id",
    "mode": "DISTINCT_ID"
  },
  "outputs": [
    {
      "name": "to-novelty",
      "resultEnrichment": {
        "query": "MATCH (n) WHERE id(n) = $that.data.id RETURN {eventId: id(n), type: n.type, timestamp: n.timestamp} AS result",
        "parameter": "that",
        "parallelism": 1
      },
      "destinations": [
        {
          "type": "ReactiveStream",
          "address": "0.0.0.0",
          "port": 9001,
          "format": {
            "type": "JSON"
          }
        }
      ]
    }
  ]
}

Step 2: Create a Novelty Model

Before creating an ingest in Novelty, you must create a model to store observations. Model names must be 1–16 lowercase characters, start with a letter, and contain only lowercase letters and digits (no hyphens, underscores, or special characters).

curl -X POST http://localhost:8181/api/v2/system/models \
  -H 'Content-Type: text/plain' \
  -d 'eventanalysis'

Returns 201 Created if the model is new, or 204 No Content if it already exists.

Step 3: Novelty Ingest from Quine Enterprise

Configure Novelty to ingest from Quine Enterprise's reactive stream and output observations to another reactive stream. The model is specified in the URL path (/model/{modelName}/ingests) — this model must already exist.

{
  "name": "from-quine",
  "transformation": "event-transform",
  "source": {
    "type": "ReactiveStream",
    "url": "localhost",
    "port": 9001,
    "format": {
      "type": "Json"
    }
  },
  "parallelism": 1,
  "outputWorkflow": {
    "type": "EachResult",
    "destinations": [
      {
        "type": "ReactiveStream",
        "address": "0.0.0.0",
        "port": 9002,
        "format": {
          "type": "JSON"
        }
      }
    ]
  }
}

Step 4: Quine Enterprise Ingest from Novelty

Configure Quine Enterprise to ingest Novelty's observations:

{
  "name": "from-novelty",
  "source": {
    "type": "ReactiveStream",
    "url": "localhost",
    "port": 9002,
    "format": {
      "type": "Json"
    }
  },
  "query": "CREATE (n:NoveltyObservation {score: $that.score, observation: $that.observation, sequence: $that.sequence, processedAt: datetime()})",
  "parameter": "that",
  "parallelism": 1
}

---

Troubleshooting

Connection Refused

If you see connection errors when starting a ReactiveStream ingest:

• Verify the publisher (server) is running and bound to the expected port
• Check that the url and port in your ingest configuration match the publisher's address and port
• Ensure no firewall rules are blocking the connection

No Data Received

If the connection succeeds but no data flows:

• Verify the format types match between publisher and subscriber (e.g., both using Json)
• Check that the standing query or observation output is actively producing results
• Use the standing query statistics API to verify results are being generated

Backpressure

If data processing slows or stalls:

• The subscriber may be overwhelmed; increase parallelism or optimize your ingest query
• Check downstream systems (persistor, standing queries) for bottlenecks
• Monitor Quine Enterprise metrics for queue depth and processing rates

---

Cluster-Wide Ingest

A reactive stream is a single subscription to one url:port, so it cannot be divided. Under cluster-wide ingest it plans as one slice, which the cluster may relocate between members.

Because the source has no resume state and cannot replay, relocation overlaps owners: the new owner starts before the old one stops, producing a bounded window of duplicate records rather than an unbounded and invisible gap. This is reported in notes at plan time and in warnings at create time, and it is why idempotent ingest queries matter for this source.

Server-sent events and WebSocket ingests behave identically for the same reason.

Running a reactive stream as a cluster-wide ingest buys you managed placement and automatic recovery onto another member, not parallelism. If you want the definition pinned to one host instead, use pinnedTo.

Additional Resources

• Standing Queries - Configure standing query outputs
• Ingest Streams Overview - General ingest configuration
• REST API Reference - Complete API documentation