Basically I want to publish a bunch of events to the ring, let the consumers do their thing and get a notification per event back to the publisher of that particular event. Looks like the Axon code does something similar using the executor of the disruptor and in the EventPublisher consumer if a callback is defined then it is submitted. A submit checks for no exceptions then “delegates” onSuccess back to the original command callback.
Haven’t toyed with Axon before but apparently the EventPublisher is supposed to publish back to EventStore or bus. Is this done in the CommandCallback onSuccess?
Anyway what I am wondering about is how you’ll tie the publisher back to the processed event data (last consumer) if it is needed in the response from the publisher. Synchronous calls.
the link you sent doesn’t work. It’s probably a link to an email in your inbox, which I cannot read.
in Axon, each command is represented by a single Disruptor Event (i.e. task). I use a Callback mechanism for the provider of the command to be able to do some processing based on the outcome. The callback is invoked in a different thread than the one that created it. If you want to wait for the processing to complete, you must use something like a CountDownLatch to make the producer wait until the result is available. That’s also what Axon can do.
Axon uses a UnitOfWork to keep track of the status of a command. It keeps track of the aggregates that need to be stored and the events that need to be published.
I think the mechanism I have implemented in the FutureCallback might be the thing you’re looking for. You could call that from one of the disruptor threads, or from a separate thread pool. In Axon, I do the latter, because callbacks can potentially send other commands or do processing that takes too long for a disruptor thread to be efficient.
@Allard, it was the wright-up describing the Axon integration with Disruptor.
Just want to double check so I understand. The UnitOfWork is state carrier that is saved in the event store. The Disruptor variation houses extra stuff for aggregations. But it is not the glue for client synchronous commands (in other words the client gets a blocking response).
First glance I thought the callback with the onSuccess was only for further business logic. Saw the FutureCallback with the latch that ties onSuccess to the get. Interesting. Will play around with that concept. Thanks! Have you had in gotch-ya with it?
The UnitOfWork is more a state carrier for the entire “transaction” for the handling of a command. One of the parts is to keep track of events to store and publiish. You’re right in that it has nothing to do with the synchronous commands.
The count down latch approach has not let me down until so far (and it’s used a lot). Just make sure that the latch.countDown is always invoked, no matter what happens. Forgetting to do so might result in threads waiting indefinitely. In Axon, you have that guarantee (even in distributed situations). The biggest down side of that callback is the Future interface. The methods declared on it are abominable, mainly due to the declared exception.
Allard, appreciate the time in explaining Axon. The little proof of concept I’m doing is along the lines of you’ll have done in Axon. Using a great NoSQL database called Neo4j but I want to separate out the “commands” (or insert/updates/deletes) from the reads. The Disruptor gives me an ordering for “commands” whereby I can lower the isolation levels on Neo4j and do other stuff with uniqueness that otherwise wouldn’t be possible. Not a whole lot different than what Datomic does. CQRS inspired.