domo42 / saga-lib Goto Github PK
View Code? Open in Web Editor NEWJava saga library to organize domain events. The goal is to organize state and activities triggered by any sort of messages.
License: Apache License 2.0
Java saga library to organize domain events. The goal is to organize state and activities triggered by any sort of messages.
License: Apache License 2.0
The default implementation InMemoryTimeoutManager of the TimeoutManager contains an executor which should be shutdown if the MessageStream is no longer in use but there is no way to do this currently.
Its also akward to get to the TimeoutManager instance itself, you almost always have to provide your own TimeoutManager to control the Scheduler (which brings me to the next issue).
It would be better to have some way to shutdown the MessageStream either via an explicit close, shutdown method or at least provide such a method on the TimeoutManager interface.
Consider you have more than one saga handling a specific message and in the first saga an exception occurs in the event handler. This exception will stop the whole SagaExecutionChain and the subsequent event handler of other sagas will never be executed.
In the SagaModules you can react on exceptions though, but these handlers are called outside the saga execution chain. Which means in case of an exception in a saga event handler you only get the execution context of this failed saga and no information about the subsequent sagas.
If the SagaInterceptor interface provides also an onError method like SagaModule has, the exception can be handled individually for the current saga (e.g. closing resources like DB sessions; automatically rollback active transactions, ... ) and still decide with the ExecutionContext wether to continue or stop the current execution chain for this message.
I have extended the following things in a patch
TimeoutManager
Updated to Java 8 and org.reflections 0.9.10
Where can I upload the patch?
A) at
https://github.com/Domo42/saga-lib/wiki/Sagas
B) at
https://github.com/Domo42/saga-lib/wiki/HandlerDescription
C) at
https://github.com/Domo42/saga-lib/wiki/Timeouts
D) at
https://github.com/Domo42/saga-lib
Change from:
Provider<? extends Saga> createProvider(Class sagaClass);
to:
<T extends Saga> Provider<T> createProvider(Class<T> sagaClass);
We have code like this in one of our saga base classes:
@EventHandler
public final void handleTimeout(final Timeout timeout) {
if (state().removePendingTimeoutId(timeout.getId())) {
onTimeout(timeout.getId(), timeout.getName(), timeout.getData());
} else {
// logging
}
}
It would be nice to remove this from the base class and add to the SagaInterceptor because this is exactly the use case for an Interceptor.
Either by explicitly canceling an invocation.
@Override
public void onHandlerExecuting(final Saga<?> saga, final ExecutionContext context, final SagaInvocation invocation, final Object message) {
if (message instanceof Timeout && !saga.state().removePendingTimeoutId(((Timeout) message).getId())) {
invocation.cancel();
}
}
Or even better would be to give the Interceptor the responsiblity for the invocation. This allows for maximum flexibility (other frameworks like Spring or Guice do the same).
See https://github.com/google/guice/wiki/AOP
@Override
public void onHandlerExecuting(final Saga<?> saga, final ExecutionContext context, final HandlerInvoker invoker, final Object message) {
if (message instanceof Timeout && saga.state().removePendingTimeoutId(((Timeout) message).getId())) {
invoker.invoke(saga, message);
}
}
A saga may choose to send out multiple requests in parallel.
If a saga can have a list of instance keys, it will be able to handle multiple responses regardless of message order
In some scenarios I would like to have better control over the state lookup and exactly when a handler invocation takes place in the sagaLib. I know there are already the Interceptors but they dont allow me to fine tune the state lookup.
When I know based on the type or execution context what lookup I need to do I can save some time and warning logs be explicitly specifying them.
This can also be helpful when I do want to use my own Timeout class and not the one already provided by the sagaLib but I would like to have the same saga instance lookup mechanism as the sagaLib already uses for its internal Timeouts.
Also for other use cases I might already know how to find a certain saga.
Therefor it would be nice to specify a "custom saga instance matcher strategy" when configuring the MessageStream (and maybe also a method to overwrite them later on).
This Strategy (or maybe Strategies) should provide me a way to decide which saga(s) I want to invoke based on:
The reason for this is to have more control over which saga handler(s) are invoked and when a (maybe expensiv e) call to the StateStorage is made.
I think its prevered to have a Map<Object, Object> to be able to use anything as a key. Especially existing value types or enums are good keys.
In my case I would like to replace those strings with enums.
This would also be a non-breaking change (i guess)
I can provide a pull-request later.
Add an additional annotation for a method on a saga that is called before any of the handlers.
It would be nice to have better support for nested executions.
A nested execution is when you call into the MessageStream twice (or more) in the same call stack.
Such a call stack would look like this:
myproject.anothersaga.handle(AnotherMessage)
com.sagaLib.xxx
com.sagaLib.xxx
com.sagaLib.MessageStream.handle
myproject.mysaga.handle(MyMessage)
com.sagaLib.xxx
com.sagaLib.xxx
com.sagaLib.MessageStream.handle`
This is helpfull to synchronously handle events in the same transaction.
The current MessageStream API could be exectend like this:
handle(Object message, Map<String, Object> headers)
handle(ExecutionContext parentContext, Object message, Map<String, Object> headers)
This combined with my second feature request "Custom Instance Matcher Strategies" would be very powerfull.
It would probably be a good idea to give the possibility to specify the "Custom Instance Matcher Strategy" seperatly for nested executions. This means that you can specify the Strategy based on the message type beeing handled and on the current execution context (which can be a nested context or not).
There is one problem with the new way of handling onError
and onFinished
of the SagaModules.
The ModulesInvoker#finish() method does swallow all exceptions without rethrowing any exception thrown by SagaModule#onFinished after all onFinished methods where invoked.
This has quite some implications in my case because as a caller of the MessageStream I now don't get any information back if a message handling did succeed or not.
I really need a way to handle this kind of errors.
if SagaModules onFinished or onError method throws an InterruptedException the ModuleInvoker catches it and does not restore the interrupted state.
public void finish() {
// swallows all exceptions and does not rethrow any of them!
Lists.reverse(finishers).forEach(Runnable::run);
}
private static void tryExecute(final Runnable runnable, final SagaModule module) {
try {
runnable.run();
} catch (Exception ex) {
// swallows InterruptedException!
LOG.error("Error executing function on module {}", module, ex);
}
}
The TimeoutId type could be an interface so that a TimeoutManager implementation can choose a suitable implementation.
Implementations could be UUTimeoutId, StringTimeoutId and NumericTimeoutId.
When storing the timeout state in a (distributed) database you could use the db key as the TimeoutId.
Right now as a workaround the builder instance's build() method has to be invoked once (so that it is not null any longer) before the ExecutionContext provider can be customized.
The default InMemoryTimeoutManager uses a Scheduler with a thread pool size of 50 which is really just waaaay to big.
Timeout handlers/callbacks should be fast and non-blocking and therefore a single thread is sufficient. Multiple threads would only be necessary if the next scheduled timeout callback already needs to be executed but the scheduler thread is still busy executing the previous timeout callback. If you run in this situation than something is wrong with your timeout callbacks.
The issues itself is about the fact that one must subclass the InMemoryTimeoutManager and cannot just plugin/inject a different implementation which would be nicer from a configuration standpoint.
Maybe you can provide a way to explicitly define an SchedulerService instance in the ModuleBuilder which is used for timeout scheduling.
As for all of my issues I would like to provide a pull request but first I am waiting for your opinion and also I have very little time for doing this now unfortunately.
Cache load method returns null when no key reader is found, resulting in an exception in the guava cache. This results in an ugly log entry that is actually not necessary
Implementation should be changed allowing caching of "not available" information as well without any exception.
A declarative, efficient, and flexible JavaScript library for building user interfaces.
๐ Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
The Web framework for perfectionists with deadlines.
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. ๐๐๐
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
We are working to build community through open source technology. NB: members must have two-factor auth.
Open source projects and samples from Microsoft.
Google โค๏ธ Open Source for everyone.
Alibaba Open Source for everyone
Data-Driven Documents codes.
China tencent open source team.