public interface MatsFactory extends MatsConfig.StartStoppable
initiations
like sending a message,
perform a request and publish a message. Getting hold of a MatsFactory is an implementation specific feature: The
JmsMatsFactory
is the standard, providing static factory methods, and can be backed by an ActiveMQ- or
Artemis-specific JMS ConnectionFactory.
An alternative to Java-based programmatic creation of Mats Endpoints is using Mats SpringConfig integration where
you use annotations like @EnableMats
, @MatsMapping
and @MatsClassMapping
.
Employing Mats SpringConfig, you'll need to get an instance of MatsFactory into the Spring context. The module is
called "mats-spring".
It is worth realizing that all of the methods staged(..., config)
,
single(...)
+
w/configs
;
terminator(...)
+
w/configs
are just convenience
methods to the main staged(...)
. These specializations could just as well have
resided in a utility class. They are included in the API since these relatively few methods seem to cover most
scenarios.
(Exception to this are the subscriptionTerminator(...)
+
w/configs
, as they
have different semantics, read the JavaDoc).
Regarding order of the Reply Message, State and Incoming Message, which can be a bit annoying to remember when
creating endpoints, and when writing process lambdas
: They are always ordered like this: R,
S, I, i.e. Reply, State, Incoming. This is to resemble a method signature having the implicit this
(or self
) reference as the first argument: ReturnType methodName(this, arguments)
. Thus, if you
remember that Mats is created to enable you to write messaging oriented endpoints that look like they are
methods, then it might stick! The process lambda thus has args (context, state, incomingMsg), unless it lacks state.
Even if it lacks state, the context is always first, and the incoming message is always last.staged(..)
-method of MatsFactory is [EndpointId, Reply
Class, State Class] - and each of the added stages
specifies
their own Incoming class and stage process lambda. The stage lambda
params will then be:
[Context, State, Incoming]. If you employ the lastStage(..)
, its process return lambda
will have the reply type as its return
type.single(..)
-method of MatsFactory is thus
[EndpointId, Reply Class, Incoming Class, process lambda] - you specify the process lambda directly.
The process single lambda
params are: [Context, Incoming] - the reply type is the
the return type of the lambda.terminator(..)
-method of MatsFactory is thus [EndpointId, State Class, Incoming Class, process lambda]
- also here specifying the process lambda directly. The lambda params of the process
terminator lambda
are: [Context, State, Incoming]MatsFactory
as keys in a HashMap
, i.e.
their equals and hashCode should remain stable throughout the life of the MatsFactory. Depending on the
implementation, instance equality may be sufficient. Note that the MatsFactory.MatsFactoryWrapper
is implemented to
forward equals and hashCode to wrappee.Modifier and Type | Interface and Description |
---|---|
static class |
MatsFactory.ContextLocal
Provides ThreadLocal access to attributes from the
MatsInitiator.MatsInitiate initiate context and MatsStage
process context - currently MatsFactory.ContextLocal.getAttribute(Class, String...) , which can provide you with the
transactionally demarcated SQL Connection if the Mats implementation provides such. |
static interface |
MatsFactory.FactoryConfig
Provides for a way to configure factory-wide elements and defaults.
|
static class |
MatsFactory.MatsFactoryWrapper
A base Wrapper for
MatsFactory , which simply implements MatsFactory, takes a MatsFactory instance and
forwards all calls to that. |
static interface |
MatsFactory.MatsWrapper<T>
Base Wrapper interface which Mats-specific Wrappers implements, defining four "wrappee" methods.
|
Modifier and Type | Field and Description |
---|---|
static java.util.function.Consumer<MatsConfig> |
NO_CONFIG
Simple Consumer<MatsConfig>-implementation that does nothing, for use where you e.g.
|
Modifier and Type | Method and Description |
---|---|
default void |
close()
Convenience method, particularly for Spring's default destroy mechanism: Default implemented to invoke
stop(30 seconds) . |
MatsInitiator |
getDefaultInitiator()
Gets or creates the default Initiator (whose name is 'default') from which to initiate new Mats processes, i.e.
|
java.util.Optional<MatsEndpoint<?,?>> |
getEndpoint(java.lang.String endpointId) |
java.util.List<MatsEndpoint<?,?>> |
getEndpoints() |
MatsFactory.FactoryConfig |
getFactoryConfig() |
java.util.List<MatsInitiator> |
getInitiators() |
MatsInitiator |
getOrCreateInitiator(java.lang.String name)
Gets or creates a new Initiator from which to initiate new Mats processes, i.e.
|
void |
holdEndpointsUntilFactoryIsStarted()
If this method is invoked before any endpoint is created, the endpoints will not start even though
MatsEndpoint.finishSetup() is invoked on them, but will wait till start() is invoked on the
factory. This feature should be employed in most setups where the MATS endpoints might use other services or
components whose order of creation and initialization are difficult to fully control, e.g. |
<R,I> MatsEndpoint<R,java.lang.Void> |
single(java.lang.String endpointId,
java.lang.Class<R> replyClass,
java.lang.Class<I> incomingClass,
java.util.function.Consumer<? super MatsEndpoint.EndpointConfig<R,java.lang.Void>> endpointConfigLambda,
java.util.function.Consumer<? super MatsStage.StageConfig<R,java.lang.Void,I>> stageConfigLambda,
MatsEndpoint.ProcessSingleLambda<R,I> processor)
Variation of
#single(String, Class, Class, ProcessSingleLambda) that can be configured "on the fly". |
<R,I> MatsEndpoint<R,java.lang.Void> |
single(java.lang.String endpointId,
java.lang.Class<R> replyClass,
java.lang.Class<I> incomingClass,
MatsEndpoint.ProcessSingleLambda<R,I> processor)
Sets up a
MatsEndpoint that just contains one stage, useful for simple "request the full person data for
this/these personId(s)" scenarios. |
<R,S> MatsEndpoint<R,S> |
staged(java.lang.String endpointId,
java.lang.Class<R> replyClass,
java.lang.Class<S> stateClass)
Sets up a
MatsEndpoint on which you will add stages. |
<R,S> MatsEndpoint<R,S> |
staged(java.lang.String endpointId,
java.lang.Class<R> replyClass,
java.lang.Class<S> stateClass,
java.util.function.Consumer<? super MatsEndpoint.EndpointConfig<R,S>> endpointConfigLambda)
Variation of
staged(String, Class, Class) that can be configured "on the fly". |
void |
start()
Starts all endpoints that has been created by this factory, by invoking
MatsEndpoint.start() on them. |
boolean |
stop(int gracefulShutdownMillis)
Stops all endpoints and initiators, by invoking
MatsEndpoint.stop(int) on all the endpoints, and
MatsInitiator.close() on all initiators that has been created by this factory. |
<S,I> MatsEndpoint<java.lang.Void,S> |
subscriptionTerminator(java.lang.String endpointId,
java.lang.Class<S> stateClass,
java.lang.Class<I> incomingClass,
java.util.function.Consumer<? super MatsEndpoint.EndpointConfig<java.lang.Void,S>> endpointConfigLambda,
java.util.function.Consumer<? super MatsStage.StageConfig<java.lang.Void,S,I>> stageConfigLambda,
MatsEndpoint.ProcessTerminatorLambda<S,I> processor)
Variation of
#subscriptionTerminator(String, Class, Class, ProcessTerminatorLambda) that can be
configured "on the fly", but notice that the concurrency of a SubscriptionTerminator is always 1. |
<S,I> MatsEndpoint<java.lang.Void,S> |
subscriptionTerminator(java.lang.String endpointId,
java.lang.Class<S> stateClass,
java.lang.Class<I> incomingClass,
MatsEndpoint.ProcessTerminatorLambda<S,I> processor)
Special kind of terminator that, in JMS-style terms, subscribes to a topic instead of listening to a queue (i.e.
|
<S,I> MatsEndpoint<java.lang.Void,S> |
terminator(java.lang.String endpointId,
java.lang.Class<S> stateClass,
java.lang.Class<I> incomingClass,
java.util.function.Consumer<? super MatsEndpoint.EndpointConfig<java.lang.Void,S>> endpointConfigLambda,
java.util.function.Consumer<? super MatsStage.StageConfig<java.lang.Void,S,I>> stageConfigLambda,
MatsEndpoint.ProcessTerminatorLambda<S,I> processor)
Variation of
#terminator(String, Class, Class, ProcessTerminatorLambda) that can be configured "on the
fly". |
<S,I> MatsEndpoint<java.lang.Void,S> |
terminator(java.lang.String endpointId,
java.lang.Class<S> stateClass,
java.lang.Class<I> incomingClass,
MatsEndpoint.ProcessTerminatorLambda<S,I> processor)
Sets up a
MatsEndpoint that contains a single stage that typically will be the reply-to endpointId for a
request initiation , or that can be used to directly send a
"fire-and-forget" style invocation to. |
default MatsFactory |
unwrapFully()
In a situation where you might be given a
MatsFactory.MatsFactoryWrapper , but need the actual implementation, this
method allows you to just always call matsFactory.unwrapFully() instead of first checking whether it
is a proxy and only then cast and unwrap. |
default <I> I |
unwrapTo(java.lang.Class<I> iface)
In a situation where you might be given a
MatsFactory.MatsFactoryWrapper , but need to find a wrappee that implements
a specific interface, this method allows you to just always call matsFactory.unwrapTo(iface) instead
of first checking whether it is a proxy and only then cast and unwrap. |
boolean |
waitForReceiving(int timeoutMillis)
Waits until all endpoints have fully entered the receive-loops, i.e.
|
static final java.util.function.Consumer<MatsConfig> NO_CONFIG
#terminator(String, Class, Class, Consumer, Consumer, ProcessTerminatorLambda)
.MatsFactory.FactoryConfig getFactoryConfig()
MatsFactory.FactoryConfig
on which to configure the factory, e.g. defaults for concurrency.<R,S> MatsEndpoint<R,S> staged(java.lang.String endpointId, java.lang.Class<R> replyClass, java.lang.Class<S> stateClass)
MatsEndpoint
on which you will add stages. The first stage is the one that will receive the
incoming (typically request) DTO, while any subsequent stage is invoked when the service that the previous stage
sent a request to, replies.
Unless the state object was sent along with the request
or
send
, the first stage will get a newly constructed empty state
instance, while the subsequent stages will get the state instance in the form it was left in the previous stage.endpointId
- the identification of this MatsEndpoint
, which are the strings that should be provided to the
MatsInitiator.MatsInitiate.to(String)
or MatsInitiator.MatsInitiate.replyTo(String, Object)
methods for this
endpoint to get the message. Typical structure is "OrderService.placeOrder"
for public
endpoints, or "OrderService.private.validateOrder"
for private (app-internal) endpoints.replyClass
- the class that this endpoint shall return.stateClass
- the class of the State DTO that will be sent along the stages.MatsEndpoint
on which to add stages.<R,S> MatsEndpoint<R,S> staged(java.lang.String endpointId, java.lang.Class<R> replyClass, java.lang.Class<S> stateClass, java.util.function.Consumer<? super MatsEndpoint.EndpointConfig<R,S>> endpointConfigLambda)
staged(String, Class, Class)
that can be configured "on the fly".<R,I> MatsEndpoint<R,java.lang.Void> single(java.lang.String endpointId, java.lang.Class<R> replyClass, java.lang.Class<I> incomingClass, MatsEndpoint.ProcessSingleLambda<R,I> processor)
MatsEndpoint
that just contains one stage, useful for simple "request the full person data for
this/these personId(s)" scenarios. This sole stage is supplied directly, using a specialization of the processor
lambda which does not have state (as there is only one stage, there is no other stage to pass state to), but
which can return the reply by simply returning it on exit from the lambda.
Do note that this is just a convenience for the often-used scenario where for example a request will just be
looked up in the backing data store, and replied directly, using only one stage, not needing any multi-stage
processing.endpointId
- the identification of this MatsEndpoint
, which are the strings that should be provided to the
MatsInitiator.MatsInitiate.to(String)
or MatsInitiator.MatsInitiate.replyTo(String, Object)
methods for this
endpoint to get the message. Typical structure is "OrderService.placeOrder"
for public
endpoints, or "OrderService.private.validateOrder"
for private (app-internal) endpoints.replyClass
- the class that this endpoint shall return.incomingClass
- the class of the incoming (typically request) DTO.processor
- the stage
that will be invoked to process the incoming message.MatsEndpoint
, but you should not add any stages to it, as the sole stage is already added.<R,I> MatsEndpoint<R,java.lang.Void> single(java.lang.String endpointId, java.lang.Class<R> replyClass, java.lang.Class<I> incomingClass, java.util.function.Consumer<? super MatsEndpoint.EndpointConfig<R,java.lang.Void>> endpointConfigLambda, java.util.function.Consumer<? super MatsStage.StageConfig<R,java.lang.Void,I>> stageConfigLambda, MatsEndpoint.ProcessSingleLambda<R,I> processor)
#single(String, Class, Class, ProcessSingleLambda)
that can be configured "on the fly".<S,I> MatsEndpoint<java.lang.Void,S> terminator(java.lang.String endpointId, java.lang.Class<S> stateClass, java.lang.Class<I> incomingClass, MatsEndpoint.ProcessTerminatorLambda<S,I> processor)
MatsEndpoint
that contains a single stage that typically will be the reply-to endpointId for a
request initiation
, or that can be used to directly send a
"fire-and-forget" style invocation
to. The sole stage is supplied directly.
This type of endpoint cannot reply, as it has no-one to reply to (hence "terminator").
Do note that this is just a convenience for the often-used scenario where an initiation requests out to some
service, and then the reply needs to be handled - and with that the process is finished. That last endpoint which
handles the reply is what is referred to as a terminator, in that it has nowhere to reply to. Note that there is
nothing hindering you in setting the replyTo endpointId in a request initiation to point to a single-stage or
multi-stage endpoint - however, any replies from those endpoints will just go void.
It is possible to initiate
from within a terminator, and one
interesting scenario here is to do a publish
to a
subscriptionTerminator
. The idea
is then that you do the actual processing via a request, and upon the reply processing in the terminator, you
update the app database with the updated information (e.g. "order is processed"), and then you publish an "update
caches" message to all the nodes of the app, so that they all have the new state of the order in their caches
(or, in a push-based GUI logic, you might want to update all users' view of that order). Note that you (as in the
processing node) will also get that published message on your instance of the SubscriptionTerminator.
It is technically possible reply
from within a terminator - but it hard to
envision many wise usage scenarios for this, as the stack at a terminator would probably be empty.endpointId
- the identification of this MatsEndpoint
, which are the strings that should be provided to the
MatsInitiator.MatsInitiate.to(String)
or MatsInitiator.MatsInitiate.replyTo(String, Object)
methods for this
endpoint to get the message. Typical structure is "OrderService.placeOrder"
for public
endpoints (which then is of a "fire-and-forget" style, since a terminator is not meant to reply), or
"OrderService.terminator.validateOrder"
for private (app-internal) terminators that is
targeted by the replyTo(endpointId,..)
invocation of an
initiation.stateClass
- the class of the State DTO that will may be provided by the
request initiation
(or that was sent along with the
invocation
).incomingClass
- the class of the incoming (typically reply) DTO.processor
- the stage
that will be invoked to process the incoming message.MatsEndpoint
, but you should not add any stages to it, as the sole stage is already added.<S,I> MatsEndpoint<java.lang.Void,S> terminator(java.lang.String endpointId, java.lang.Class<S> stateClass, java.lang.Class<I> incomingClass, java.util.function.Consumer<? super MatsEndpoint.EndpointConfig<java.lang.Void,S>> endpointConfigLambda, java.util.function.Consumer<? super MatsStage.StageConfig<java.lang.Void,S,I>> stageConfigLambda, MatsEndpoint.ProcessTerminatorLambda<S,I> processor)
#terminator(String, Class, Class, ProcessTerminatorLambda)
that can be configured "on the
fly".<S,I> MatsEndpoint<java.lang.Void,S> subscriptionTerminator(java.lang.String endpointId, java.lang.Class<S> stateClass, java.lang.Class<I> incomingClass, MatsEndpoint.ProcessTerminatorLambda<S,I> processor)
MatsInitiator.MatsInitiate.publish(Object)
or MatsInitiator.MatsInitiate.replyToSubscription(String, Object)
methods.
Notice that the concurrency of a SubscriptionTerminator is always 1, as it makes no sense to have multiple
processors for a subscription - all of the processors would just get an identical copy of each message. If
you do need to handle massive amounts of messages, or your work handling is slow, you should instead of handling
the work in the processor itself, rather accept the message as fast as possible and send the work out to be
processed by some kind of thread pool. (The tool MatsFuturizer
does this for its Future-completion
handling).endpointId
- the identification of this MatsEndpoint
, which are the strings that should be provided to the
MatsInitiator.MatsInitiate.to(String)
or MatsInitiator.MatsInitiate.replyTo(String, Object)
methods for this
endpoint to get the message.stateClass
- the class of the State DTO that will may be provided by the
request initiation
(or that was sent along with the
invocation
).incomingClass
- the class of the incoming (typically reply) DTO.processor
- the stage
that will be invoked to process the incoming message.MatsEndpoint
, but you should not add any stages to it, as the sole stage is already added.<S,I> MatsEndpoint<java.lang.Void,S> subscriptionTerminator(java.lang.String endpointId, java.lang.Class<S> stateClass, java.lang.Class<I> incomingClass, java.util.function.Consumer<? super MatsEndpoint.EndpointConfig<java.lang.Void,S>> endpointConfigLambda, java.util.function.Consumer<? super MatsStage.StageConfig<java.lang.Void,S,I>> stageConfigLambda, MatsEndpoint.ProcessTerminatorLambda<S,I> processor)
#subscriptionTerminator(String, Class, Class, ProcessTerminatorLambda)
that can be
configured "on the fly", but notice that the concurrency of a SubscriptionTerminator is always 1.java.util.List<MatsEndpoint<?,?>> getEndpoints()
MatsEndpoint
s created on this MatsFactory
.java.util.Optional<MatsEndpoint<?,?>> getEndpoint(java.lang.String endpointId)
endpointId
- which MatsEndpoint
to return, if present.MatsEndpoint
if present, Optional.empty()
if not.MatsInitiator getDefaultInitiator()
ProcessContext.initiate(..)
. The idea here is that you thus can
create methods that both can be used from the outside of a Mats Stage (thus resulting in an ordinary initiation),
but if the same method is invoked within a Mats Stage, the initiation will partake in the same transactional
demarcation as the rest of what happens within that Mats Stage. Note that you get a bit strange semantics wrt.
the exceptions that MatsInitiator.initiate(..)
and
MatsInitiator.initiateUnchecked(..)
raises: Outside of a
Mats Stage, they can throw in the given situations those exceptions describe. However, within a Mats Stage, they
will never throw those exceptions, since the actual initiation is not performed until the Mats Stage exits. But,
if you want to make such a dual mode method that can be employed both outside and within a Mats Stage, you should
thus code for the "Outside" mode, handling those Exceptions as you would in an ordinary initiation.
If you would really NOT want this - i.e. you for some reason want the initiation performed within the stage to
execute even though the Mats Stage fails - you may use getOrCreateInitiator(String)
, and you can even
request the same underlying default initiator by just supplying that method with the argument "default". Please,
however, make sure you understand the quite heavy consequence of this: If the Mats Stage throws, the
retry-mechanism will kick in, running the Mats Stage one more time, and you will thus potentially send that
message many times - one time per retry - since such an initiation with a NON-default MatsInitiator is
specifically then not part of the Stage's transactional demarcation.
Just to ensure that this point comes across: The returned MatsInitiator is Thread Safe, and meant for reuse:
You are not supposed to create one instance of MatsInitiator
per message you need to send, and
then close it afterwards - rather either create one for the entire application, or e.g. for each component:
The MatsInitiator
can have underlying backend resources attached to it - which also means that it needs
to be closed
for a clean application shutdown (Note that all MatsInitiators are
closed when MatsFactory.stop()
is invoked).MatsInitiator
, whose name is 'default', on which messages can be
initiated
.MatsFactory.ContextLocal.getAttribute(Class, String...)
MatsInitiator getOrCreateInitiator(java.lang.String name)
MatsInitiator
could be that each initiator might have its own
connection to the underlying message broker. You also might want to name the initiators based on what part of the
application uses it; The name of the initiator shows up in monitors and tooling.
IMPORTANT NOTICE!! Please read the JavaDoc of getDefaultInitiator()
for important information
wrt. transactional demarcation when employing a NON-default MatsInitiator within a Mats Stage.
Just to ensure that this point comes across: The returned MatsInitiator is Thread Safe, and meant for reuse:
You are not supposed to create one instance of MatsInitiator
per message you need to send, and
then close it afterwards - rather either create one for the entire application, or e.g. for each component:
The MatsInitiator
can have underlying backend resources attached to it - which also means that it needs
to be closed
for a clean application shutdown (Note that all MatsInitiators are
closed when MatsFactory.stop()
is invoked).MatsInitiator
, on which messages can be initiated
.java.util.List<MatsInitiator> getInitiators()
MatsInitiator
s created on this MatsFactory
.void start()
MatsEndpoint.start()
on them.
Subsequently clears the holdEndpointsUntilFactoryIsStarted()
-flag.start
in interface MatsConfig.StartStoppable
void holdEndpointsUntilFactoryIsStarted()
MatsEndpoint.finishSetup()
is invoked on them, but will wait till start()
is invoked on the
factory. This feature should be employed in most setups where the MATS endpoints might use other services or
components whose order of creation and initialization are difficult to fully control, e.g. typically in an IoC
container like Spring. Set up the "internal machinery" of the system, including internal services, and only after
all this is running, then fire up the endpoints. If this is not done, the endpoints might start consuming
messages off of the MQ (there might already be messages waiting when the service boots), and thus invoke
services/components that are not yet fully started.
Note: To implement delayed start for a specific endpoint, simply hold off on invoking
MatsEndpoint.finishSetup()
until you are OK with it being started and hence starts consuming messages
(e.g. when the needed cache service is finished populated); This semantics works both when
holdEndpointsUntilFactoryIsStarted() has been invoked or not.MatsEndpoint.finishSetup()
,
MatsEndpoint.start()
boolean waitForReceiving(int timeoutMillis)
MatsEndpoint.waitForReceiving(int)
on all the endpoints started from this factory.
Note: If there are no Endpoints registered, this will immediately return true
!waitForReceiving
in interface MatsConfig.StartStoppable
timeoutMillis
- number of milliseconds before giving up the wait, returning false
. 0 is indefinite
wait, negative values are not allowed.true
if the entity started within the timeout, false
if it did not start.boolean stop(int gracefulShutdownMillis)
MatsEndpoint.stop(int)
on all the endpoints, and
MatsInitiator.close()
on all initiators that has been created by this factory. They can be started again
individually, or all at once by invoking start()
Should be invoked at application shutdown.stop
in interface MatsConfig.StartStoppable
gracefulShutdownMillis
- number of milliseconds to let the stage processors wait after having asked for them to shut down,
and interrupting them if they have not shut down yet.true
if all Endpoints (incl. e.g. threads) and resources (e.g. JMS Connections) closed
successfully.default void close()
stop(30 seconds)
.default <I> I unwrapTo(java.lang.Class<I> iface)
MatsFactory.MatsFactoryWrapper
, but need to find a wrappee that implements
a specific interface, this method allows you to just always call matsFactory.unwrapTo(iface)
instead
of first checking whether it is a proxy and only then cast and unwrap.this
if iface.isAssignableFrom(thisClass)
, otherwise throws
IllegalArgumentException
(overridden by wrappers).default MatsFactory unwrapFully()
MatsFactory.MatsFactoryWrapper
, but need the actual implementation, this
method allows you to just always call matsFactory.unwrapFully()
instead of first checking whether it
is a proxy and only then cast and unwrap.this
for implementations (overridden by wrappers).