marvin-j97/corinth

📬 Portable message queue written in Rust

Corinth

Corinth is a simple and portable message queue server written in Rust.

Features

• FIFO (strict message order, exactly-once delivery)
• file system persistence (AOF .jsonl format)
• explicit message acknowledgment (ack)
• message deduplication
• Dead-letter queues (DLQ)
• Optional queue maximum size
• JSON REST API

Run

Grab a pre-compiled binary (https://github.com/marvin-j97/corinth/releases).

If you're using an OS other than Windows, Linux, Mac you'll have to build a binary yourself. Run:

(Unix)
./corinth-[version]

(Windows)
corinth-[version].exe
.\corinth-[version].exe

That's it. By default the server runs on port 44444.

By using environment variables, you can change some settings:

Name	Description	Default
CORINTH_PORT	Port the server runs on	44444
CORINTH_BASE_FOLDER	Folder where persistent data is stored	./.corinth
CORINTH_COMPACT_INTERVAL	Compaction interval (in seconds)	86400

API documentation

See https://marvin-j97.github.io/corinth/api/.

Getting started

Create a queue named 'my-queue'. Queues are persistent by default.

curl -X PUT http://localhost:44444/queue/my-queue

Enqueue an item to queue ('item' can be any JSON object)

curl -X POST http://localhost:44444/queue/my-queue/enqueue -H "Content-Type: application/json" --data "{ \"messages\": [{ \"item\": { \"name\": \"My item\" }, \"deduplication\": null }] }"

Dequeue item from queue

curl -X POST http://localhost:44444/queue/my-queue/dequeue

By default, messages need to be acknowledged, otherwise they will be requeued after a specific timeout. If you don't care about acknowledging, you can acknowledge on dequeuing instead:

curl -X POST http://localhost:44444/queue/my-queue/dequeue?ack=true

Acknowledge message reception

curl -X POST http://localhost:44444/queue/my-queue/[message id]/ack

Build from source

cargo build --release

Docker

https://hub.docker.com/r/dotvirus/corinth

docker pull dotvirus/corinth

Build the image

docker build -t corinth .

Where corinth is the name of the image you are about to create.

Running a new container using the image

docker run -d -it -p 127.0.0.1:8080:44444 --rm --name corinth-queue corinth

To verify if everything worked, open up a browser and enter localhost:8080

Change application port

Use the CORINTH_PORT variable to change the port Corinth will bind to.

docker run -d -it -p 127.0.0.1:8080:22222 --env CORINTH_PORT=22222 --rm --name corinth-queue corinth

Docker persistence

To create a persistent queue within Docker you need to mount a volume and make sure to correctly set the CORINTH_BASE_FOLDER environment variable.

docker run -d -it --env CORINTH_BASE_FOLDER=/corinth/.corinth -p 127.0.0.1:8080:44444 --mount source=corinthvol,target=/corinth --rm --name corinth-queue corinth

Pay attention to the --mount command. This will create a volume and mount it to your Docker container as '/corinth'. In this example, Corinth will store its persistence in the .corinth folder (/corinth/.corinth).

You can inspect the volume details using

docker volume inspect corinthvol

For more details on Docker volumes, check out the official docs here.

Node.js client

https://github.com/marvin-j97/corinth.js

corinth's Issues

Redeliver attempts counter in Message

DELETE /queue/<id>

Deletes a queue

Benchmarks

Update README, documentation, code (comments)

GET /queues

List names & number of queues

DELETE /queue/<id>/dequeue

Removes and returns the queue head.

?ack=true (optional): Automatically confirm reception, increments done counter
otherwise: add message to ack map, if reception is not confirmed, it will be added back to the queue

Build release scripts/workflows

dequeue DELETE -> POST

Compact route

GET /queue/:name: return ack & dedup times

Management console

• Create app folder with a Vue 3 Vite app.
• Serve index.html, js & css files over server.
• Create button component
• Create input component
• Queue details page
• -> Link to DLQ if it exists
• optionally poll for data changes
• Purge queue button
• Delete queue button
• Edit queue (description, deduplication time & requeue time)
• Create queue with DLQ
• Create queue with description
• Create queue with custom requeue, deduplication time, max size
• In queue listing, optionally poll for data changes or refresh button
• Show queue's disk & memory size in listing
• Server info page (index page)
• Enqueue message(s)
• Dequeue message(s)
• Peek queue head
• Finishing touches/refactor

Dependabot couldn't find a package.json for this project

Dependabot couldn't find a package.json for this project.

Dependabot requires a package.json to evaluate your project's current JavaScript dependencies. It had expected to find one at the path: /test/package.json.

If this isn't a JavaScript project, or if it is a library, you may wish to disable updates for it from within Dependabot.

View the update logs.

Add assets folder to workflows & Dockerfile

dockerfile

Deduplication on enqueue

DELETE /queue/<id>/purge

Removes all items from a queue
Clear deduplication set & ack map
Rewrite files

?reset_stats=true

POST /queue/<id>/enqueue

Enqueue a new item to a queue

?deduplication_id (optional)

message gets internal ID no matter what

Cleanup dedup map

Benchmark Vec vs VecDeque vs LinkedList

Message updatedAt timestamp

Error logging

Use eprintln everywhere

Add optional queue description property

Auto-compact interval

• Start interval for new queues
• Start interval for queues loaded from disk
• CORINTH_COMPACT_INTERVAL environment variable
• Unit tests

Optimize docker image size

Use separate build context

Webhook events

API docs

Optional queue max length

GET /queue/<id>/peek

Return the queue head. Don't dequeue queue head though.

Add queue disk size & lastCompactedAt properties

Use Path::join

Use ActixWeb?

Dead letter queue rerouting when msg requeue threshold reached

MessageState -> Failed

dead_letter_target
dead_letter_threshold

queues designated as dead letter queues cannot be deleted

Basic access management

Test: Metadata file updating after PATCH

Don't allow max length to change after creation

Diagnostics (read/write times, size over time), show in dashboard as graphs

POST /queue/<id>/<msg>/ack

Acknowledge reception of a message

Fix tests

PATCH /queue/<id> to change times (and possibly other future settings)

Support batch /dequeue

Priority queue

Support batch /enqueue

Unit test DELETE & PURGE

Remove on-the-fly queue creation on enqueue

• Remove logic
• Update API
• Adjust unit tests

PUT /queue/<id>

Create a queue

Queue persistence (optional)

• Store items
• Load items
• Store queue metadata
• Apply stored queue metadata
• Auto detect existing queues on startup
• Unit tests