n8n-nodes-themq

Native n8n community nodes for TheMQ — a lightweight,
Redis-backed message broker built around queues and consumers.

Install

In n8n: Settings → Community Nodes → Install → n8n-nodes-themq.
Or npm install n8n-nodes-themq in ~/.n8n, then restart n8n.

Add a TheMQ API credential: the TheMQ Base URL
(e.g. http://themq:3000) and the API Token — the API_TOKEN value
pinned in docker-compose.yml (overridable via environment).

TheMQ Trigger

Listens to a TheMQ queue over a persistent consumer connection —
every message starts an execution immediately. There is no polling
interval; deactivating the workflow closes the connection and pending
messages are requeued for redelivery.

Credentials
Queue

Options
  + Acknowledge
  + Max Concurrent Executions

Acknowledge

Every delivered message is leased, not given away: it must be
acknowledged, or it comes back. There is no deadline — like a
RabbitMQ manual-ack consumer, the broker holds the message however
long the workflow takes. When the acknowledgement happens depends
on this mode:

• Immediately — acknowledged as soon as it arrives, with no
  concurrency limit (RabbitMQ auto-ack). The workflow still runs, but a later failure can no
  longer return the message. Use it when losing a message on failure is
  acceptable.
• Execution Finishes — acknowledged when the execution finishes,
  however long that takes, whether it succeeded or not.
• Execution Finishes Successfully — acknowledged only on success, however
  long that takes. On failure the message goes back to the queue and is
  delivered again (with redelivered: true and a higher deliveryCount).
• Specified Later in Workflow — the trigger does not settle. Add a
  TheMQ → Acknowledge node where the message is truly done; its
  fields pick up the trigger item automatically. The message stays held
  without a deadline until then. If the run ends without
  acknowledgement, success acknowledges and failure returns the message.

Whichever mode you pick, a dropped connection returns its pending
messages immediately for redelivery. If the connection drops while the
workflow is active (TheMQ restart, deploy, network blip), the trigger
reconnects on its own with backoff instead of failing — only
misconfiguration (bad credentials, deleted queue, protocol errors)
deactivates it with an error. Delivery is at-least-once:
keep your workflow idempotent if duplicates would hurt.

Max Concurrent Executions

At most this many messages being processed at the same time (default
1). The broker leases no more than this many unacknowledged messages
to the trigger, so further messages simply wait in the queue until one
is acknowledged. The option is hidden while Acknowledge is
Immediately — with nothing ever left unacknowledged there is nothing
to cap, so that mode flows without a limit. Raise the limit when your
workflow can safely run in parallel.

A message that keeps failing will keep coming back. To stop the loop,
check deliveryCount with an IF node and route poison messages to an
TheMQ → Acknowledge node — acknowledging removes them for good.
(The Delete Message operation is for messages still waiting in the
queue, not for ones already delivered.)

Output item

{
  "queue": "orders",
  "messageId": "order-42",
  "consumerId": "cons_abc",
  "data": { "message": "hello" },
  "deliveryCount": 1,
  "redelivered": false
}

Troubleshooting

"Timed out waiting for a TheMQ hello reply" on activation.
The trigger opened the connection but the server never answered. Check,
in order:

1. TheMQ is 3.0.0+ and reachable from n8n:
   docker compose pull && docker compose up -d, then
   curl http://<host>:3000/health/ready.
2. Redis is healthy (/health/ready returns 200, not 503).
3. The TheMQ logs around activation (docker compose logs themq):
   Consumer connected means the hello succeeded; subscribe-failed /
   frame-failed entries explain the refusal.

Node shows "TheMQ is not reachable". The Base URL in your credential
is wrong or TheMQ is down. Verify with
curl http://<host>:3000/health/live from the n8n host.

"UNAUTHENTICATED". The API Token in your credential is wrong or stale.
Fetch the current one from Redis or pin API_TOKEN.

TheMQ node

Exactly four operations — nothing else:

Publish · Acknowledge · Requeue · Delete Message

• Publish — Queue, Message ID (required; the upsert key), Message
  Data, Upsert flag (update the ID in place instead of conflicting),
  On Conflict (Error, or Skip to return the existing message's state with
  skipped: true instead of failing — works for queued and leased
  messages), Delay (wait before the message becomes available).
• Acknowledge — marks a trigger-delivered message as successfully
  processed. Queue and Message ID default to the trigger item, so no
  wiring is needed.
• Requeue — returns a leased message (picked up but unacked) to the
  queue. Queue and Message ID default to the trigger item; the optional
  Options (+) take a replacement Message Data and a Delay before it
  becomes available again. Empty options requeue with the original
  details. Waiting messages (never picked up) are edited with Upsert
  publish instead.
• Delete Message — removes a waiting message so it is never
  processed. Queue and Message ID default to the trigger item.

TheMQ Trigger (Specified Later in Workflow)
      ↓
   Process
      ↓
TheMQ → Acknowledge

Or automatic:

TheMQ Trigger (Execution Finishes Successfully)
      ↓
   Process

Examples

Publish from any workflow (Message ID is required — it is the upsert key;
tick Upsert to update an existing ID instead of conflicting):

Schedule Trigger → TheMQ (Publish: queue "orders", id "order-42", data {...})

Competing workers — two activated workflows with triggers on the same
queue each get different messages, FIFO.

Drop a queued message before it is ever delivered:

TheMQ → Delete Message

Send a delivered message back with new data, a delay, or both:

TheMQ → Requeue

Development

npm install
npm run typecheck
npm run lint
npm run format
npm run build        # tsc -> dist/ (+ icons)

License

ISC — see LICENSE.