This topic explains how to filter events for multi-site (WAN) distribution in VMware Tanzu GemFire.
You can create gateway sender and gateway receiver filters to control which events are queued and distributed to a remote site, or to modify the data stream that is transmitted between GemFire sites.
You can implement and deploy two different types of filter for multi-site events:
GatewayEventFilter
. A GatewayEventFilter
implementation determines whether a region event is placed in a gateway sender queue and/or whether an event in a gateway queue is distributed to a remote site. You can optionally add one or more GatewayEventFilter
implementations to a gateway sender, either in the cache.xml
configuration file or using the Java API.
Tanzu GemFire makes a synchronous call to the filter’s beforeEnqueue
method before it places a region event in the gateway sender queue. The filter returns a boolean value that specifies whether the event should be added to the queue.
Tanzu GemFire asynchronously calls the filter’s beforeTransmit
method to determine whether the gateway sender dispatcher thread should distribute the event to a remote gateway receiver.
For events that are distributed to another site, Tanzu GemFire calls the listener’s afterAcknowledgement
method to indicate that is has received an ack from the remote site after the event was received.
GatewayTransportFilter. Use a GatewayTransportFilter
implementation to process the TCP stream that sends a batch of events that is distributed from one Tanzu GemFire cluster to another over a WAN. A GatewayTransportFilter
is typically used to perform encryption or compression on the data that distributed. You install the same GatewayTransportFilter
implementation on both a gateway sender and gateway receiver.
When a gateway sender processes a batch of events for distribution, Tanzu GemFire delivers the stream to the getInputStream
method of a configured GatewayTransportFilter
implementation. The filter processes and returns the stream, which is then transmitted to the gateway receiver. When the gateway receiver receives the batch, Tanzu GemFire calls the getOutputStream
method of a configured filter, which again processes and returns the stream so that the events can be applied in the local cluster.
You install a GatewayEventFilter
implementation to a configured gateway sender in order to decide which events are queued and distributed. You install a GatewayTransportFilter
implementation to both a gateway sender and a gateway receiver to process the stream of batched events that are distributed between two sites:
XML example
<cache>
<gateway-sender id="remoteA" parallel="true" remote-distributed-system-id="1">
<gateway-event-filter>
<class-name>org.apache.geode.util.SampleEventFilter</class-name>
<parameter name="param1">
<string>"value1"</string>
</parameter>
</gateway-event-filter>
<gateway-transport-filter>
<class-name>org.apache.geode.util.SampleTransportFilter</class-name>
<parameter name="param1">
<string>"value1"</string>
</parameter>
</gateway-transport-filter>
</gateway-sender>
</cache>
<cache>
...
<gateway-receiver start-port="1530" end-port="1551">
<gateway-transport-filter>
<class-name>org.apache.geode.util.SampleTransportFilter</class-name>
<parameter name="param1">
<string>"value1"</string>
</parameter>
</gateway-transport-filter>
</gateway-receiver>
</cache>
gfsh example
gfsh>create gateway-sender --id=remoteA --parallel=true --remote-distributed-id="1"
--gateway-event-filter=org.apache.geode.util.SampleEventFilter
--gateway-transport-filter=org.apache.geode.util.SampleTransportFilter
gfsh>create gateway-receiver --start-port=1530 --end-port=1551 \
--gateway-transport-filter=org.apache.geode.util.SampleTransportFilter
Note: You cannot specify parameters and values for the Java class you specify with the --gateway-transport-filter
option.
API example
Cache cache = new CacheFactory().create();
GatewayEventFilter efilter = new SampleEventFilter();
GatewayTransportFilter tfilter = new SampleTransportFilter();
GatewaySenderFactory gateway = cache.createGatewaySenderFactory();
gateway.setParallel(true);
gateway.addGatewayEventFilter(efilter);
gateway.addTransportFilter(tfilter);
GatewaySender sender = gateway.create("remoteA", "1");
sender.start();
Cache cache = new CacheFactory().create();
GatewayTransportFilter tfilter = new SampleTransportFilter();
GatewayReceiverFactory gateway = cache.createGatewayReceiverFactory();
gateway.setStartPort(1530);
gateway.setEndPort(1551);
gateway.addTransportFilter(tfilter);
GatewayReceiver receiver = gateway.create();
receiver.start();