package forklift.replay;

import forklift.decorators.Message;
import forklift.decorators.MultiThreaded;
import forklift.decorators.OnMessage;
import forklift.decorators.OnValidate;
import forklift.decorators.Order;
import forklift.source.decorators.Topic;

import javax.inject.Inject;

@Topic("forklift.replay.es")
@MultiThreaded(10)
public class ReplayConsumer {
    @Inject private ReplayESWriter writer;
    @Message private ReplayESWriterMsg msg;

    @OnValidate
    public boolean onValidate() {
        return this.writer != null && this.msg != null;
    }

    @OnMessage
    public void onMessage() {
        this.writer.poll(msg);
    }
}

package forklift.replay;

import forklift.decorators.Message;
import forklift.decorators.MultiThreaded;
import forklift.decorators.OnMessage;
import forklift.decorators.OnValidate;
import forklift.decorators.Order;
import forklift.source.decorators.Queue;

import javax.inject.Inject;

@Queue("forklift.replay.es")
@MultiThreaded(10)
public class ReplayConsumer {
    @Inject private ReplayESWriter writer;
    @Message private ReplayESWriterMsg msg;

    @OnValidate
    public boolean onValidate() {
        return this.writer != null && this.msg != null;
    }

    @OnMessage
    public void onMessage() {
        this.writer.poll(msg);
    }
}

Move ReplayConsumer to a queue to avoid duplicate messages