stream-applications

Files

Artem Bilan 61f1412365 GH-276 Apply ComponentCustomizer in other modules (#284 )

* GH-276 Apply ComponentCustomizer in other modules

Fixes https://github.com/spring-cloud/stream-applications/issues/276

Some changes are straightforward as just a `ComponentCustomizer<?>`
injection and its optional usage for the target component customization.
Other changes are more drastic since it is better to use `proxyBeanMethods = false`,
so some direct bean method calls had to be reworked to the bean method argument
injections

* Fix some deprecations from Project Reactor
* Remove `UpdatingMongoDbMessageSource` since `MongoDbMessageSource`
supports now an `update` option in Spring Integration

NOTE: The `ZeroMqSupplierConfiguration` has its own customizer already
like `Consumer<ZMQ.Socket>`, so it can be revised to a more broader solution,
but in the future version.

Some modules just does not make sense to modify for more advance customization
since all their configuration is covered with plain configuration properties.
See `splitter-function` or `wavefront-consumer`

* * Use `ObjectProvider<WebClientCustomizer>` for `HttpRequestFunctionConfiguration`
instead of our own `ComponentCustomizer`.
Although this has to be revised in favor of fully auto-configured `WebClient.Builder`.
Even that `maximumBufferSize` is included into codecs configuration properties.
* The `MongoDbMessageSource` has now an `update` option in Spring Integration
therefore remove redundant already `UpdatingMongoDbMessageSource`
and populate `this.properties.getUpdateExpression()` directly ot the
`MongoDbMessageSource` bean
* Modify supplier configurations to avoid manual starts for endpoints on
reactive `Publisher` subscriptions.
Spring Integration provides now a `toReactivePublisher(true)` operator
to have all the endpoints stopped on start, but started automatically
when subscription happens to the provided `Publisher`
* Update Copyright to the current year for all the affected classes
* Use `@Nullable` instead of `ObjectProvider` in the `AggregatorFunctionConfiguration`
dependency injection for consistency
* Fix `LogConsumerConfiguration` to not have a `log()` in the end - it is not terminal
in Spring Integration anymore.
Use recommended `nullChannel()` terminating operator for use-cases like this.

* * Remove `beanName` from `ComponentCustomizer` contract

Co-authored-by: Chris Bono <cbono@vmware.com>

2022-07-15 19:36:21 -05:00

src

GH-276 Apply ComponentCustomizer in other modules (#284 )

2022-07-15 19:36:21 -05:00

pom.xml

GH-271: Remove duplication from pom.xml files

2022-05-26 13:57:44 -04:00

README.adoc

Implement geode-supplier and geode-source along with common config and test support

2020-06-03 16:05:28 -04:00

README.adoc

# Geode Consumer

This module provides a `java.util.function.Consumer` that can be reused and composed in other applications.
The `Consumer` configures an Apache Geode client that connects to an external Apache Geode cache server or locator to write Message contents to an existing region.
The consumer uses the `CacheWritingMessageHandler` from `Spring Integration` which writes entries, as key-value pairs to the given region.
A SpEl Expression, given by the property `geode.consumer.key-expression` is used to derive the desired key from the inbound Message.
The value is the message payload.

## PDX Serialization

The supplier works with PDX serialized cache objects of type https://geode.apache.org/releases/latest/javadoc/org/apache/geode/pdx/PdxInstance.html[PdxInstance], which Geode uses to store objects that can be represented as JSON.
If the target region uses PDX serialization ,you should set `geode.consumer.json` to `true`.
In this case, the expected payload is Json, serialized as a byte[] or String, and will be converted to PdxInstance before writing to the region.
The root object for evaluating `key-expression` will be of type PDXInstance so values are referenced using the `getField(...)` method.


## Beans for injection

You can import the `GeodeConsumerConfiguration` configuration in a Spring Boot application and then inject the `geodeConsumer` bean as type `Consumer<Message<?>>`.
If necessary, can use the bean name `geodeCqSupplier` as a qualifier.

Once injected, you can invoke the `accept` method of the `Consumer`.

## Configuration Options

Required properties:

* `geode.region.region-name` - The name of the existing remote region.
* `geode.pool.host-addresses` - A comma delimited list of `host:port` pairs. By default these are locator addresses but are cache server addresses if you set `geode.pool.connect-type=server`.

For more information on the various options available, please see:

* link:src/main/java/org/springframework/cloud/fn/consumer/geode/cq/GeodeConsumerProperties.java[GeodeConsumerProperties.java] (`geode.consumer`)

Many of the options, common to functions that use Apache Geode, are configured by several `@ConfigurationProperties` classes which are included as needed:

* link:../../common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeRegionProperties.java[GeodeRegionProperties.java] (`geode.region`)
* link:../../common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodePoolProperties.java[GeodePoolProperties.java] (`geode.pool`)
* link:../../common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeSecurityProperties.java[GeodeSecurityProperties.java] (`geode.security`)
* link:../../common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeSslProperties.java[GeodeSslProperties.java] (`geode.security.ssl`)

## Examples

See this link:src/test/java/org/springframework/cloud/fn/consumer/geode/GeodeConsumerApplicationTests.java[test suite] for examples of how this consumer is used.

## Other usage

See this link:../../../applications/sink/geode-sink/README.adoc[README] where this consumer is used to create a Spring Cloud Stream application to write to a Geode region.