[[hazelcast]]
= Hazelcast Support

Spring Integration provides channel adapters and other utility components to interact with an in-memory data grid https://hazelcast.com[Hazelcast].

You need to include this dependency into your project:

[tabs]
======
Maven::
+
[source, xml, subs="normal", role="primary"]
----
<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-hazelcast</artifactId>
    <version>{project-version}</version>
</dependency>
----

Gradle::
+
[source, groovy, subs="normal", role="secondary"]
----
compile "org.springframework.integration:spring-integration-hazelcast:{project-version}"
----
======

The XML namespace and schemaLocation definitions for Hazelcast components are:

[source,xml]
----
xmlns:int-hazelcast="http://www.springframework.org/schema/integration/hazelcast"
xsi:schemaLocation="http://www.springframework.org/schema/integration/hazelcast
          https://www.springframework.org/schema/integration/hazelcast/spring-integration-hazelcast.xsd"
----

[[hazelcast-inbound]]
== Hazelcast Event-driven Inbound Channel Adapter

Hazelcast provides distributed data structures such as:

* `com.hazelcast.map.IMap`
* `com.hazelcast.multimap.MultiMap`
* `com.hazelcast.collection.IList`
* `com.hazelcast.collection.ISet`
* `com.hazelcast.collection.IQueue`
* `com.hazelcast.topic.ITopic`
* `com.hazelcast.replicatedmap.ReplicatedMap`

It also provides event listeners in order to listen to modifications made to these data structures.

* `com.hazelcast.core.EntryListener<K, V>`
* `com.hazelcast.collection.ItemListener`
* `com.hazelcast.topic.MessageListener`

The Hazelcast Event-Driven Inbound Channel Adapter listens to related cache events and sends event messages to the defined channel.
It supports both XML and JavaConfig driven configurations.

[[xml-configuration-:]]
=== XML Configuration :

[source,xml]
----
<int-hazelcast:inbound-channel-adapter channel="mapChannel"
                      cache="map"
                      cache-events="UPDATED, REMOVED"
                      cache-listening-policy="SINGLE" />
----

The Hazelcast Event-Driven Inbound Channel Adapter requires the following attributes:

* `channel`: Specifies the channel to which messages are sent;
* `cache`: Specifies the distributed Object reference which is listened to.
It is a mandatory attribute;
* `cache-events`: Specifies cache events which are listened for.
It is an optional attribute and its default value is `ADDED`.
Its supported values are as follows :

* Supported cache event types for `IMap` and `MultiMap`: `ADDED`, `REMOVED`, `UPDATED`, `EVICTED`, `EVICT_ALL` and `CLEAR_ALL`;
* Supported cache event types for `ReplicatedMap`: `ADDED`, `REMOVED`, `UPDATED`, `EVICTED`;
* Supported cache event types for `IList`, `ISet` and `IQueue`: `ADDED`, `REMOVED`.
There are no cache event types for `ITopic`.

* `cache-listening-policy`: Specifies the cache listening policy as `SINGLE` or `ALL`.
It is an optional attribute and its default value is `SINGLE`.
Each Hazelcast inbound channel adapter listening to the same cache object with the same cache-events attribute, can receive a single event message or all event messages.
If it is `ALL`, all Hazelcast inbound channel adapters listening to the same cache object with the same cache-events attribute, will receive all event messages.
If it is `SINGLE`, they will receive unique event messages.

Some configuration samples:

[source,xml]
.Distributed Map
----
<int:channel id="mapChannel"/>

<int-hazelcast:inbound-channel-adapter channel="mapChannel"
                              cache="map"
                              cache-events="UPDATED, REMOVED" />

<bean id="map" factory-bean="instance" factory-method="getMap">
    <constructor-arg value="map"/>
</bean>

<bean id="instance" class="com.hazelcast.core.Hazelcast"
            factory-method="newHazelcastInstance">
    <constructor-arg>
        <bean class="com.hazelcast.config.Config" />
    </constructor-arg>
</bean>
----

[source,xml]
.Distributed MultiMap
----
<int-hazelcast:inbound-channel-adapter channel="multiMapChannel"
                              cache="multiMap"
                              cache-events="ADDED, REMOVED, CLEAR_ALL" />

<bean id="multiMap" factory-bean="instance" factory-method="getMultiMap">
    <constructor-arg value="multiMap"/>
</bean>
----

[source,xml]
.Distributed List
----
<int-hazelcast:inbound-channel-adapter  channel="listChannel"
                               cache="list"
                               cache-events="ADDED, REMOVED"
                               cache-listening-policy="ALL" />

<bean id="list" factory-bean="instance" factory-method="getList">
    <constructor-arg value="list"/>
</bean>
----

[source,xml]
.Distributed Set
----
<int-hazelcast:inbound-channel-adapter channel="setChannel" cache="set" />

<bean id="set" factory-bean="instance" factory-method="getSet">
    <constructor-arg value="set"/>
</bean>
----

[source,xml]
.Distributed Queue
----
<int-hazelcast:inbound-channel-adapter  channel="queueChannel"
                               cache="queue"
                               cache-events="REMOVED"
                               cache-listening-policy="ALL" />

<bean id="queue" factory-bean="instance" factory-method="getQueue">
    <constructor-arg value="queue"/>
</bean>
----

[source,xml]
.Distributed Topic
----
<int-hazelcast:inbound-channel-adapter channel="topicChannel" cache="topic" />

<bean id="topic" factory-bean="instance" factory-method="getTopic">
    <constructor-arg value="topic"/>
</bean>
----

[source,xml]
.Replicated Map
----
<int-hazelcast:inbound-channel-adapter channel="replicatedMapChannel"
                              cache="replicatedMap"
                              cache-events="ADDED, UPDATED, REMOVED"
                              cache-listening-policy="SINGLE"  />

<bean id="replicatedMap" factory-bean="instance" factory-method="getReplicatedMap">
    <constructor-arg value="replicatedMap"/>
</bean>
----

[[java-configuration-sample:]]
=== Java Configuration Sample:

The following sample shows a `DistributedMap` configuration.
The same configuration can be used for other distributed data structures(`IMap`, `MultiMap`, `ReplicatedMap`, `IList`, `ISet`, `IQueue` and `ITopic`):

[source,java]
----
@Bean
public PollableChannel distributedMapChannel() {
    return new QueueChannel();
}

@Bean
public IMap<Integer, String> distributedMap() {
    return hazelcastInstance().getMap("Distributed_Map");
}

@Bean
public HazelcastInstance hazelcastInstance() {
    return Hazelcast.newHazelcastInstance();
}

@Bean
public HazelcastEventDrivenMessageProducer hazelcastEventDrivenMessageProducer() {
    final HazelcastEventDrivenMessageProducer producer = new HazelcastEventDrivenMessageProducer(distributedMap());
    producer.setOutputChannel(distributedMapChannel());
    producer.setCacheEventTypes("ADDED,REMOVED,UPDATED,CLEAR_ALL");
    producer.setCacheListeningPolicy(CacheListeningPolicyType.SINGLE);

    return producer;
}
----

[[hazelcast-continuous-query]]
== Hazelcast Continuous Query Inbound Channel Adapter

Hazelcast Continuous Query enables listening to modifications performed on specific map entries.
The Hazelcast Continuous Query Inbound Channel Adapter is an event-driven channel adapter which listens to the related distributed map events in the light of the defined predicate.

[tabs]
======
Java::
+
[source, java, role="primary"]
----
@Bean
public PollableChannel cqDistributedMapChannel() {
    return new QueueChannel();
}

@Bean
public IMap<Integer, String> cqDistributedMap() {
    return hazelcastInstance().getMap("CQ_Distributed_Map");
}

@Bean
public HazelcastInstance hazelcastInstance() {
    return Hazelcast.newHazelcastInstance();
}

@Bean
public HazelcastContinuousQueryMessageProducer hazelcastContinuousQueryMessageProducer() {
    final HazelcastContinuousQueryMessageProducer producer =
        new HazelcastContinuousQueryMessageProducer(cqDistributedMap(), "surname=TestSurname");
    producer.setOutputChannel(cqDistributedMapChannel());
    producer.setCacheEventTypes("UPDATED");
    producer.setIncludeValue(false);

    return producer;
}
----

XML::
+
[source, xml, role="secondary"]
----
<int:channel id="cqMapChannel"/>

<int-hazelcast:cq-inbound-channel-adapter
                channel="cqMapChannel"
                cache="cqMap"
                cache-events="UPDATED, REMOVED"
                predicate="name=TestName AND surname=TestSurname"
                include-value="true"
                cache-listening-policy="SINGLE"/>

<bean id="cqMap" factory-bean="instance" factory-method="getMap">
    <constructor-arg value="cqMap"/>
</bean>

<bean id="instance" class="com.hazelcast.core.Hazelcast"
            factory-method="newHazelcastInstance">
    <constructor-arg>
        <bean class="com.hazelcast.config.Config" />
    </constructor-arg>
</bean>
----
======

It supports six attributes as follows:

* `channel`: Specifies the channel to which messages are sent;
* `cache`: Specifies the distributed Map reference which is listened to.
Mandatory;
* `cache-events`: Specifies cache events which are listened for.
Optional attribute with `ADDED` being its default value.
Supported values are `ADDED`, `REMOVED`, `UPDATED`, `EVICTED`, `EVICT_ALL` and `CLEAR_ALL`;
* `predicate`: Specifies a predicate to listen to the modifications performed on specific map entries.
Mandatory;
* `include-value`: Specifies including the value and oldValue in a continuous query result.
Optional with `true` being the default;
* `cache-listening-policy`: Specifies the cache listening policy as `SINGLE` or `ALL`.
Optional with the default value being `SINGLE`.
Each Hazelcast CQ inbound channel adapter listening to the same cache object with the same cache-events attribute, can receive a single event message or all event messages.
If it is `ALL`, all Hazelcast CQ inbound channel adapters listening to the same cache object with the same cache-events attribute, will receive all event messages.
If it is `SINGLE`, they will receive unique event messages.

[[hazelcast-cluster-monitor]]
== Hazelcast Cluster Monitor Inbound Channel Adapter

A Hazelcast Cluster Monitor supports listening to modifications performed on the cluster.
The Hazelcast Cluster Monitor Inbound Channel Adapter is an event-driven channel adapter and listens to related Membership, Distributed Object, Migration, Lifecycle and Client events:

[tabs]
======
Java::
+
[source, java, role="primary"]
----
@Bean
public PollableChannel eventChannel() {
    return new QueueChannel();
}

@Bean
public HazelcastInstance hazelcastInstance() {
    return Hazelcast.newHazelcastInstance();
}

@Bean
public HazelcastClusterMonitorMessageProducer hazelcastClusterMonitorMessageProducer() {
    HazelcastClusterMonitorMessageProducer producer = new HazelcastClusterMonitorMessageProducer(hazelcastInstance());
    producer.setOutputChannel(eventChannel());
    producer.setMonitorEventTypes("DISTRIBUTED_OBJECT");

    return producer;
}
----

XML::
+
[source, xml, role="secondary"]
----
<int:channel id="monitorChannel"/>

<int-hazelcast:cm-inbound-channel-adapter
                 channel="monitorChannel"
                 hazelcast-instance="instance"
                 monitor-types="MEMBERSHIP, DISTRIBUTED_OBJECT" />

<bean id="instance" class="com.hazelcast.core.Hazelcast"
            factory-method="newHazelcastInstance">
    <constructor-arg>
        <bean class="com.hazelcast.config.Config" />
    </constructor-arg>
</bean>
----
======

It supports three attributes as follows :

* `channel`: Specifies the channel to which messages are sent;
* `hazelcast-instance`: Specifies the Hazelcast Instance reference to listen for cluster events.
It is a mandatory attribute;
* `monitor-types`: Specifies the monitor types which are listened for.
It is an optional attribute with `MEMBERSHIP` being the default value.
Supported values are `MEMBERSHIP`, `DISTRIBUTED_OBJECT`, `MIGRATION`, `LIFECYCLE`, `CLIENT`.

[[hazelcast-distributed-sql]]
== Hazelcast Distributed SQL Inbound Channel Adapter

Hazelcast allows running distributed queries on the distributed map.
The Hazelcast Distributed SQL Inbound Channel Adapter is a polling inbound channel adapter.
It runs the defined distributed-sql command and returns results depending on the iteration type.

[tabs]
======
Java::
+
[source, java, role="primary"]
----
@Bean
public PollableChannel dsDistributedMapChannel() {
    return new QueueChannel();
}

@Bean
public IMap<Integer, String> dsDistributedMap() {
    return hazelcastInstance().getMap("DS_Distributed_Map");
}

@Bean
public HazelcastInstance hazelcastInstance() {
    return Hazelcast.newHazelcastInstance();
}

@Bean
@InboundChannelAdapter(value = "dsDistributedMapChannel", poller = @Poller(maxMessagesPerPoll = "1"))
public HazelcastDistributedSQLMessageSource hazelcastDistributedSQLMessageSource() {
    final HazelcastDistributedSQLMessageSource messageSource =
        new HazelcastDistributedSQLMessageSource(dsDistributedMap(),
            "name='TestName' AND surname='TestSurname'");
    messageSource.setIterationType(DistributedSQLIterationType.ENTRY);

    return messageSource;
}
----

XML::
+
[source, xml, role="secondary"]
----
<int:channel id="dsMapChannel"/>

<int-hazelcast:ds-inbound-channel-adapter
            channel="dsMapChannel"
            cache="dsMap"
            iteration-type="ENTRY"
            distributed-sql="active=false OR age >= 25 OR name = 'TestName'">
    <int:poller fixed-delay="100"/>
</int-hazelcast:ds-inbound-channel-adapter>

<bean id="dsMap" factory-bean="instance" factory-method="getMap">
    <constructor-arg value="dsMap"/>
</bean>

<bean id="instance" class="com.hazelcast.core.Hazelcast"
            factory-method="newHazelcastInstance">
    <constructor-arg>
        <bean class="com.hazelcast.config.Config" />
    </constructor-arg>
</bean>
----
======

It requires a poller and supports four attributes:

* `channel`: Specifies the channel to which messages are sent.
It is a mandatory attribute;
* `cache`: Specifies the distributed `IMap` reference which is queried.
It is mandatory attribute;
* `iteration-type`: Specifies result type.
Distributed SQL can be run on `EntrySet`, `KeySet`, `LocalKeySet` or `Values`.
It is an optional attribute with `VALUE` being the default.
Supported values are `ENTRY, `KEY`, `LOCAL_KEY` and `VALUE`;
* `distributed-sql`: Specifies the where clause of the sql statement.
It is a mandatory attribute.

[[hazelcast-outbound-channel-adapter]]
== Hazelcast Outbound Channel Adapter

The Hazelcast Outbound Channel Adapter listens to its defined channel and writes incoming messages to related distributed cache.
It expects one of `cache`, `cache-expression` or `HazelcastHeaders.CACHE_NAME` for distributed object definition.
Supported Distributed Objects are: `IMap`, `MultiMap`, `ReplicatedMap`, `IList`, `ISet`, `IQueue` and `ITopic`.

[tabs]
======
Java::
+
[source, java, role="primary"]
----
@Bean
public MessageChannel distributedMapChannel() {
    return new DirectChannel();
}

@Bean
public IMap<Integer, String> distributedMap() {
    return hzInstance().getMap("Distributed_Map");
}

@Bean
public HazelcastInstance hzInstance() {
    return Hazelcast.newHazelcastInstance();
}

@Bean
@ServiceActivator(inputChannel = "distributedMapChannel")
public HazelcastCacheWritingMessageHandler hazelcastCacheWritingMessageHandler() {
    HazelcastCacheWritingMessageHandler handler = new HazelcastCacheWritingMessageHandler();
    handler.setDistributedObject(distributedMap());
    handler.setKeyExpression(new SpelExpressionParser().parseExpression("payload.id"));
    handler.setExtractPayload(true);
    return handler;
}
----

XML::
+
[source, xml, role="secondary"]
----
<int-hazelcast:outbound-channel-adapter channel="mapChannel"
                    cache-expression="headers['CACHE_HEADER']"
                    key-expression="payload.key"
                    extract-payload="true"/>
----
======

It requires the following attributes :

* `channel`: Specifies the channel to which messages are sent;
* `cache`: Specifies the distributed object reference.
Optional;
* `cache-expression`: Specifies the distributed object via Spring Expression Language (SpEL).
Optional;
* `key-expression`: Specifies the key of a key-value pair via Spring Expression Language (SpEL).
Optional and required for only for `IMap`, `MultiMap` and `ReplicatedMap` distributed data structures.
* `extract-payload`: Specifies whether to send the whole message or just the payload.
Optional attribute with `true` being the default.
If it is true, just the payload will be written to the distributed object.
Otherwise, the whole message will be written by converting both message headers and payload.

By setting distributed object name in the header, messages can be written to different distributed objects via same channel.
If `cache` or `cache-expression` attributes are not defined, a `HazelcastHeaders.CACHE_NAME` header has to be set in a request `Message`.

[[hazelcast-leader-election]]
== Hazelcast Leader Election

If leader election is needed (e.g. for highly available message consumer where only one node should receive messages) a Hazelcast-based `LeaderInitiator` can be used:

[source,java]
----
@Bean
public HazelcastInstance hazelcastInstance() {
    return Hazelcast.newHazelcastInstance();
}

@Bean
public LeaderInitiator initiator() {
    return new LeaderInitiator(hazelcastInstance());
}
----

When a node is elected leader it will send an `OnGrantedEvent` to all application listeners.

[[hazelcast-message-store]]
== Hazelcast Message Store

For distributed messaging state management, for example for a persistent `QueueChannel` or tracking `Aggregator` message groups, the `HazelcastMessageStore` implementation is provided:

[source,java]
----
@Bean
public HazelcastInstance hazelcastInstance() {
    return Hazelcast.newHazelcastInstance();
}

@Bean
public MessageGroupStore messageStore() {
    return new HazelcastMessageStore(hazelcastInstance());
}
----

By default, the `SPRING_INTEGRATION_MESSAGE_STORE` `IMap` is used to store messages and groups as a key/value.
Any custom `IMap` can be provided to the `HazelcastMessageStore`.

[[hazelcast-metadata-store]]
== Hazelcast Metadata Store

An implementation of a `ListenableMetadataStore` is available using a backing Hazelcast `IMap`.
The default map is created with a name `SPRING_INTEGRATION_METADATA_STORE` which can be customized.

[source,java]
----
@Bean
public HazelcastInstance hazelcastInstance() {
    return Hazelcast.newHazelcastInstance();
}

@Bean
public MetadataStore metadataStore() {
    return new HazelcastMetadataStore(hazelcastInstance());
}
----

The `HazelcastMetadataStore` implements `ListenableMetadataStore` which allows you to register your own listeners of type `MetadataStoreListener` to listen for events via `addListener(MetadataStoreListener callback)`.


[[hazelcast-lock-registry]]
== Hazelcast Lock Registry

An implementation of a `LockRegistry` is available using a backing Hazelcast distributed `ILock` support:

[source,java]
----
@Bean
public HazelcastInstance hazelcastInstance() {
    return Hazelcast.newHazelcastInstance();
}

@Bean
public LockRegistry lockRegistry() {
    return new HazelcastLockRegistry(hazelcastInstance());
}
----

When used with a shared `MessageGroupStore` (e.g. `Aggregator` store management), the `HazelcastLockRegistry` can be used to provide this functionality across multiple application instances, such that only one instance can manipulate the group at a time.

NOTE: For all the distributed operations the CP Subsystem must be enabled on `HazelcastInstance`.

[[hazelcast-message-channels]]
== Message Channels with Hazelcast

The Hazelcast `IQueue` and `ITopic` distributed objects are, essentially, messaging primitives and can be use with Spring Integration core components without extra implementations in this Hazelcast module.

The xref:channel/implementations.adoc#channel-implementations-queuechannel[`QueueChannel`] can be supplied by any `java.util.Queue`, including the mentioned Hazelcast distributed `IQueue`:

[source,java]
----
@Bean
PollableChannel hazelcastQueueChannel(HazelcastInstance hazelcastInstance) {
    return new QueueChannel(hazelcastInstance.getQueue("springIntegrationQueue"));
}
----

Placing this config on several nodes in Hazelcast cluster of the application, will make the `QueueChannel` as distributed and only one node will be able to poll a single `Message` from that `IQueue`.
This works similar to xref:jms.adoc#jms-channel[`PollableJmsChannel`], xref:kafka.adoc#kafka-channels[`PollableKafkaChannel`] or xref:amqp/channels.adoc[`PollableAmqpChannel`].

If the producer side is not a Spring Integration application, there is no way to configure a `QueueChannel`, and therefore the plain Hazelcast `IQueue` API is used to produce the data.
In this case, the `QueueChannel` approach is wrong on the consumer side: an xref:channel-adapter.adoc#channel-adapter-namespace-inbound[Inbound Channel Adapter] solution must be used instead:

[source,java]
----
@Bean
public IQueue<String> myStringHzQueue(HazelcastInstance hazelcastInstance) {
    return hazelcastInstance.getQueue("springIntegrationQueue");
}

@Bean
@InboundChannelAdapter(channel = "stringValuesFromHzQueueChannel")
Supplier<String> fromHzIQueueSource(IQueue<String> myStringHzQueue) {
    return myStringHzQueue::poll;
}
----

The `ITopic` abstraction in Hazelcast has similar semantics to a `Topic` in JMS: all subscribers receive published messages.
With a pair of simple `MessageChannel` beans this mechanism is supported as an out-of-the-box feature:

[source,java]
----
@Bean
public ITopic<Message<?>> springIntegrationTopic(HazelcastInstance hazelcastInstance,
        MessageChannel fromHazelcastTopicChannel) {

    ITopic<Message<?>> topic = hazelcastInstance.getTopic("springIntegrationTopic");
	topic.addMessageListener(m -> fromHazelcastTopicChannel.send(m.getMessageObject()));
	return topic;
}

@Bean
public MessageChannel publishToHazelcastTopicChannel(ITopic<Message<?>> springIntegrationTopic) {
    return new FixedSubscriberChannel(springIntegrationTopic::publish);
}

@Bean
public MessageChannel fromHazelcastTopicChannel() {
    return new DirectChannel();
}
----

The `FixedSubscriberChannel` is an optimized variant of `DirectChannel`, which requires a `MessageHandler` on initialization.
Since the `MessageHandler` is a functional interface a simple lambda for the `handleMessage` method can be provided.
When a message is sent to the `publishToHazelcastTopicChannel` it is just published onto the Hazelcast `ITopic`.
The `com.hazelcast.topic.MessageListener` is a functional interface, too, hence a lambda to the `ITopic#addMessageListener` can be provided.
So, a subscriber to the `fromHazelcastTopicChannel` will consume all messages sent to the mentioned `ITopic`.

An `ExecutorChannel` can be supplied with an `IExecutorService`.
For example, with respective configuration a cluster-wide singleton can be achieved:

[source,java]
----
@Bean
public HazelcastInstance hazelcastInstance() {
    return Hazelcast.newHazelcastInstance(
                new Config()
                    .addExecutorConfig(new ExecutorConfig()
                         .setName("singletonExecutor")
                         .setPoolSize(1)));
}

@Bean
public MessageChannel hazelcastSingletonExecutorChannel(HazelcastInstance hazelcastInstance) {
    return new ExecutorChannel(hazelcastInstance.getExecutorService("singletonExecutor"));
}
----
