IgnisMQ¶

IgnisMQ is a Java library that provides reliable asynchronous message processing built on top of Magazine and Aerospike, with first-class Dropwizard integration for microservice environments.

Key Features¶

How It Works¶

sequenceDiagram
    participant App as Your Application
    participant MQ as IgnisMQManager
    participant Q as MagazineQueue
    participant M as Magazine (Aerospike)
    participant S as Sideline Magazine

    App->>MQ: getQueue("order-queue")
    MQ-->>App: IQueue
    App->>Q: publish(order)
    Q->>M: magazine.load(json)

    Note over Q: A consumer task polls on the worker pool

    Q->>M: magazine.fire()
    M-->>Q: MagazineData
    Q->>App: handler.handle(order)

    alt Handler returns false
        Q->>S: sidelineMagazine.load(json)
        Note over S: Message saved for retry
        alt Sideline accepted it
            Q->>M: magazine.delete(data)
        else Sideline refused it
            Note over M: Left in place for the sweeper<br/>- deleting would lose the only copy
        end
    else Handler returns true
        Q->>M: magazine.deleteAll(batch)
    end

    Note over S: A shovel task runs periodically
    S-->>M: Move messages back to main queue

Quick Start¶

// 1. Extend the bundle
public class MyIgnisMQBundle extends IgnisMQBundle<MyConfig> {
    @Override
    protected IgnisMQContext context(MyConfig config) {
        return IgnisMQContext.builder()
                .clientId(config.getClientId())
                .farmId(config.getFarmId())
                .storage(new AerospikeStorage(config.getAerospikeConfig(), "my-ns"))
                .curatorFramework(curator)
                .build();
    }
}

// 2. Register in your Application
bootstrap.addBundle(ignisMQBundle);

// 3. Initialize handlers and create a queue
manager.initialiseMessageHandlers(Map.of(
    "order-handler", Map.entry(Order.class, new OrderHandler())
));

manager.createQueue(CreateQueueRequest.builder()
    .name("order-queue")
    .concurrency(8)
    .messageHandlerType("order-handler")
    .build());

// 4. Publish messages
manager.getQueue("order-queue").publish(new Order("ORD-123", 99.99));

// 1. Create storage and manager
BaseStorage storage = new AerospikeStorage(aerospikeConfig, "my-ns");
IgnisMQManager manager = new IgnisMQManager(
    "my-client", storage, new ObjectMapper(),
    new SimpleMeterRegistry(), curatorFramework, "farm-1",
    IgnisMQSettings.defaults()
);

// 2. Register handlers
manager.initialiseMessageHandlers(Map.of(
    "order-handler", Map.entry(Order.class, new OrderHandler())
));

// 3. Create a queue and publish
manager.createQueue(CreateQueueRequest.builder()
    .name("order-queue")
    .concurrency(4)
    .messageHandlerType("order-handler")
    .build());

manager.getQueue("order-queue").publish(new Order("ORD-123", 99.99));

Message Lifecycle at a Glance¶

stateDiagram-v2
    [*] --> Published: publish()
    Published --> Fired: consumer fires message
    Fired --> Consumed: handler returns true
    Fired --> Sidelined: handler returns false / exception
    Consumed --> [*]: message deleted
    Sidelined --> Shoveled: shovel task runs
    Shoveled --> Published: message re-queued
    Fired --> Swept: message stuck (no ack)
    Swept --> Sidelined: sweeper recovers

Architecture Overview¶

flowchart TD
    subgraph Application
        App["Your App"] --> Bundle["IgnisMQBundle"]
        Bundle --> Manager["IgnisMQManager"]
    end

    subgraph Queue Management
        Manager --> Q1["MagazineQueue #1"]
        Manager --> Q2["MagazineQueue #2"]
        Manager --> QN["MagazineQueue #N"]
    end

    subgraph Per-Queue Components
        Q1 --> C["Consumer tasks<br/>(1s fixed delay)"]
        Q1 --> SH["Shovel tasks<br/>(configurable interval)"]
    end

    subgraph Cluster Coordination
        Manager --> TI["TaskInitializer"]
        TI --> LE["LeaderElector"]
        LE --> ZK["ZooKeeper"]
        TI --> SW["Sweeper<br/>(one assigned instance)"]
    end

    subgraph Storage
        C --> Mag["Magazine<br/>(Main)"]
        C --> Side["Magazine<br/>(Sideline)"]
        SH --> Mag
        SH --> Side
        SW --> AQS["AerospikeQueueService"]
        Mag --> AS["Aerospike"]
        Side --> AS
        AQS --> AS
    end

    Manager -.->|refreshQueues<br/>every 5 min| AS

What to Read Next¶