Flux.from() is cheap and can be used to marshal the inputs everywhere
internally. With this change users ought to be able to register any
function of any Publisher type.
If an isolated function doesn't have Message in its classpath, we
will never actually need to instantiate that class. This change
makes sure we check first.
This results in a better experience for users because the consumer
that they write is only applied to a Flux that is subscribed to
by the framework once. It gives better control over the flow of
foos, e.g. if some component wants to subscribe on a thread.
User can switch off source or sink behaviour (the default is to bind
to input and output streams), and then configure the name of a
supplier (for a source) or consumer (for a sink).
Makes it possible to support other "function" types in the future.
The user is always taking a risk with the lookup that the object
returned has the generic type desired (but that hasn't changed
with this commit). FunctionCatalog is a lot simpler as a result
and also a lot more flexible.
Functions with Flux and Message (as well as POJOs and Flux of POJO
which were already supported) should now work if they are created in
an isolated class loader. Preconditions:
* The class loaders must have the reactor-core (and reactive-streams)
shared between the app and the function. Practically speaking this means
there has to be a parent class loader with just reactive types, and
sibling children for the app and the function. This is not a new
requirement (it was needed for Flux of POJO anyway).
* Message types are handled reflectively, so they don't have to be in a
shared class loader. But they do have to be on the class path on
both sides (obviously).
A registered singleton doesn't have a BeanDefinition, but it might have
a compiled type with enough generic information to pull out the
input and output types.
Stops application from blowing up if there are messages available
on startup and the broker is not there. There are still errors
(failed messages), but it doesn't stop the app from starting.
Maybe should be optional to fail fast?
It didn't really make any sense to have custom conditions that
depend on the presence or absence of beans of type Function,
Supplier, Consumer because the actual endpoints are derived
from the FunctionCatalog (which might not be based on
bean definitions). This approach is far simpler, and
reduces the amount of custom code in the stream binder.
The spring.cloud.function.stream.supplier.enabled flag
is awkward, so we should try and find a way to avoid that.
There's also no reason it should need to be set in the
deployer tests.
If the incoming message has a stream_routekey header, we use that
to route the message to a named function. Also adding the header
to messages coming out of suppliers.
The biggest change here is sort of orthogonal: it fixes a bug where
Spring Integration would subscribe twice to the same input channel
if the FunctionCatalog contains both functions and consumers. Then
when a message comes in it is dispatched to one or the other, but not
both. So the routing key couldn't work without fixing that
problem.
* If there is only one function, use it unconditionally
* If no functions match, ensure that null is not the selector
* If a conversion succeeds, check the type of the result
The FunctionInspector needs to be able to distinguish between a
function of Flux<Foo> and a function of Flux<Message<Foo>>. Then
it can do the conversion a level below.
Only supports Message->Message or POJO->POJO (no mixtures), so there
is only one new method in FunctionInspector.
No support in the web endpoints yet. But it's probably not so hard
to add.
If there is only one function you shouldn't have to set any configuration
to get a stream app to run. This also implementation supports multiple
functions, trying to guess which one to use based on the type of
the incoming message payload. In principle that could be strategized
as a simple router function (e.g. to look for a header with a function
name).
If there are functions and consumers in the same app, they will
subscribe to the same input channel (and hence by default Spring
Integration will load balance between them). This could also probably
use some more features, to specify the desired behaviour.
If user *does* supply spring.cloud.function.stream.endpoint then it
is used and overrides all other possible routes.
- use ProxyWrapper around a FluxConsumer as well
making it consistent with the behaviour of Flux
functions
- Enable introspection for scanned beans
- Fix failing tests by passing JSON string as input messages
(marshalled form expected from the binder)
