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

Implement geodeCqSupplier Implement Geode CQ Source Implement Geode Sink Removed @TestInstance Combine CQ Source and Source Clean up per review Update README Fix Javadoc
2020-05-27 08:47:09 -04:00
parent 783d0dbc03
commit 1051a40d40
28 changed files with 2202 additions and 2 deletions
--- a/common/function-test-support/pom.xml
+++ b/common/function-test-support/pom.xml
@@ -15,13 +15,16 @@

    <properties>
        <sshd-core.version>1.6.0</sshd-core.version>
+        <test-containers.version> 1.14.2</test-containers.version>
+        <apache-ftpserver.version>1.1.1</apache-ftpserver.version>
+        <geode-starter.version>1.2.7.RELEASE</geode-starter.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.apache.ftpserver</groupId>
            <artifactId>ftpserver-core</artifactId>
-            <version>1.1.1</version>
+            <version>${apache-ftpserver.version}</version>
            <scope>compile</scope>
        </dependency>
        <dependency>
@@ -44,6 +47,16 @@
            <artifactId>sshd-core</artifactId>
            <version>${sshd-core.version}</version>
        </dependency>
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>testcontainers</artifactId>
+            <version>${test-containers.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.data</groupId>
+            <artifactId>spring-data-geode</artifactId>
+            <optional>true</optional>
+        </dependency>
    </dependencies>

 </project>
--- a/common/function-test-support/src/main/java/org/springframework/cloud/fn/test/support/geode/GeodeContainer.java
+++ b/common/function-test-support/src/main/java/org/springframework/cloud/fn/test/support/geode/GeodeContainer.java
@@ -0,0 +1,183 @@
+/*
+ * Copyright 2020-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.test.support.geode;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.concurrent.Future;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.GenericContainer;
+
+import org.springframework.lang.NonNull;
+import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
+
+/**
+ * A Test Container that starts a Geode Locator and Server on configured ports. This also
+ * provides methods for executing one or more Gfsh commands.
+ */
+public class GeodeContainer extends GenericContainer {
+	private static Logger logger = LoggerFactory.getLogger(GeodeContainer.class);
+
+	private final int locatorPort;
+
+	private final int cacheServerPort;
+
+	/**
+	 * Create a Geode container from a Docker image.
+	 * @param dockerImageName the name of the image.
+	 * @param locatorPort the locator port.
+	 * @param cacheServerPort the cache server port.
+	 */
+	public GeodeContainer(@NonNull String dockerImageName, int locatorPort, int cacheServerPort) {
+		super(dockerImageName);
+		this.locatorPort = locatorPort;
+		this.cacheServerPort = cacheServerPort;
+	}
+
+	/**
+	 * Create a Geode Container from a {@code Future<String>}. Test containers provides some
+	 * implementations as image builders, such as
+	 * {@link org.testcontainers.images.builder.ImageFromDockerfile}.
+	 * @param image the image builder.
+	 * @param locatorPort the locator port.
+	 * @param cacheServerPort the server port.
+	 */
+	public GeodeContainer(@NonNull Future<String> image, int locatorPort, int cacheServerPort) {
+		super(image);
+		this.locatorPort = locatorPort;
+		this.cacheServerPort = cacheServerPort;
+	}
+
+	/**
+	 * A convenience method to connect to a locator with Gfsh.
+	 * @return the connect command String.
+	 */
+	public String connect() {
+		return "connect --locator=" + locators();
+	}
+
+	/**
+	 * Get the locator port.
+	 * @return the locator port.
+	 */
+	public int getLocatorPort() {
+		return locatorPort;
+	}
+
+	/**
+	 * Get the cache server port.
+	 * @return the cache server port.
+	 */
+	public int getCacheServerPort() {
+		return cacheServerPort;
+	}
+
+	/**
+	 *
+	 * @return Geode locators as host[port],...
+	 */
+	public String locators() {
+		return "localhost[" + locatorPort + "]";
+	}
+
+	/**
+	 * Invoke the `gfsh` shell, Connect to the locator and execute the commands.
+	 * @param command a list of commands to execute in a single `gfsh` invocation.
+	 * @return the {@link org.testcontainers.containers.Container.ExecResult}
+	 */
+	public ExecResult connectAndExecGfsh(String... command) {
+		ArrayList<String> args = new ArrayList<>(Arrays.asList(command));
+		args.add(0, connect());
+		return execInContainer(Gfsh.command(args.toArray(new String[args.size()])).commandParts());
+	}
+
+	/**
+	 * Invoke the `gfsh` shell, and execute the commands.
+	 * @param command a list of commands to execute in a single `gfsh` invocation.
+	 * @return the {@link org.testcontainers.containers.Container.ExecResult}
+	 */
+	public ExecResult execGfsh(String... command) {
+		return execInContainer(Gfsh.command(command).commandParts());
+	}
+
+	/**
+	 * Executes a command in the container, logging stdout and stderr and wrapping checked
+	 * exceptions.
+	 * @see GenericContainer#execInContainer(String...)
+	 * @param command the command to execute.
+	 * @return the {@link org.testcontainers.containers.Container.ExecResult}
+	 */
+	@Override
+	public ExecResult execInContainer(String... command) {
+		try {
+			ExecResult execResult = super.execInContainer(command);
+			logger.debug("stdout: {}", execResult.getStdout());
+			if (execResult.getExitCode() != 0) {
+				logger.warn("stdout: {}", execResult.getStdout());
+				logger.warn("stderr: {}", execResult.getStderr());
+			}
+			return execResult;
+		}
+		catch (IOException e) {
+			throw new RuntimeException(e.getMessage(), e);
+		}
+		catch (InterruptedException e) {
+			throw new RuntimeException(e.getMessage(), e);
+		}
+	}
+
+	/**
+	 * Builds a Gfsh command.
+	 */
+	public final static class Gfsh {
+
+		public static Command command(String... gfshCommands) {
+			return new Command(gfshCommands);
+		}
+
+		public final static class Command {
+			private final List<String> commandParts = new LinkedList<>();
+
+			private Command(String... gfshCommands) {
+				Assert.notEmpty(gfshCommands, "at least one command is required");
+				for (String gfshCommand : gfshCommands) {
+					Assert.hasText(gfshCommand, "command must contain text");
+					if (commandParts.size() == 0) {
+						commandParts.add("gfsh");
+					}
+					commandParts.add("-e");
+					commandParts.add(gfshCommand);
+				}
+			}
+
+			public String[] commandParts() {
+				return commandParts.toArray(new String[commandParts.size()]);
+			}
+
+			public String toString() {
+				return StringUtils.collectionToDelimitedString(commandParts, ",");
+			}
+		}
+	}
+
+}
--- a/common/function-test-support/src/main/java/org/springframework/cloud/fn/test/support/geode/GeodeContainerIntializer.java
+++ b/common/function-test-support/src/main/java/org/springframework/cloud/fn/test/support/geode/GeodeContainerIntializer.java
@@ -0,0 +1,94 @@
+/*
+ * Copyright 2020-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.test.support.geode;
+
+import java.util.Optional;
+import java.util.function.Consumer;
+
+import com.github.dockerjava.api.command.CreateContainerCmd;
+import com.github.dockerjava.api.model.ExposedPort;
+import com.github.dockerjava.api.model.PortBinding;
+import com.github.dockerjava.api.model.Ports;
+import org.testcontainers.images.builder.ImageFromDockerfile;
+
+import org.springframework.util.SocketUtils;
+
+/**
+ * Creates and starts a {@link GeodeContainer} using available random ports for locator
+ * and server. Runs a {@code Consumer<GeodeContainer>} post processor if provided.
+ * @author David Turanski
+ */
+public class GeodeContainerIntializer {
+
+	private int locatorPort;
+
+	private int cacheServerPort;
+
+	private GeodeContainer geode;
+
+	private Optional<Consumer<GeodeContainer>> postProcessor;
+
+	/**
+	 * Create, start, and perform post processing on a {@link GeodeContainer}.
+	 * @param postProcessor a {@code Consumer<GeodeContainer>} to run after the container is started.
+	 */
+	public GeodeContainerIntializer(Consumer<GeodeContainer> postProcessor) {
+		cacheServerPort = SocketUtils.findAvailableTcpPort();
+
+		locatorPort = SocketUtils.findAvailableTcpPort();
+
+		this.postProcessor = Optional.ofNullable(postProcessor);
+
+		geode = new GeodeContainer(new ImageFromDockerfile()
+				.withFileFromClasspath("Dockerfile", "geode/Dockerfile")
+				.withBuildArg("CACHE_SERVER_PORT", String.valueOf(cacheServerPort))
+				.withBuildArg("LOCATOR_PORT", String.valueOf(locatorPort)),
+				locatorPort, cacheServerPort);
+		startContainer();
+	}
+
+	/**
+	 * Create and start a {@link GeodeContainer}.
+	 */
+	public GeodeContainerIntializer() {
+		this(null);
+	}
+
+	private void startContainer() {
+		// There is apparently no way to initialize Geode with random port mapping. Ports
+		// must be the same on client and server.
+		Consumer<CreateContainerCmd> cmd = e -> e.withPortBindings(
+				new PortBinding(Ports.Binding.bindPort(cacheServerPort),
+						new ExposedPort(cacheServerPort)),
+				new PortBinding(Ports.Binding.bindPort(locatorPort), new ExposedPort(locatorPort)));
+		// Wait forever
+		geode.withCommand("tail", "-f", "/dev/null").withCreateContainerCmdModifier(cmd).start();
+
+		geode.execGfsh("start locator --name=Locator1 --hostname-for-clients=localhost --port=" + locatorPort);
+		geode.execGfsh("connect --locator=" + geode.locators(),
+				"start server --name=Server1 --hostname-for-clients=localhost --server-port=" + cacheServerPort);
+		postProcessor.ifPresent(geodeContainerConsumer -> geodeContainerConsumer.accept(geode));
+	}
+
+	/**
+	 * @return the {@link GeodeContainer} instance.
+	 */
+	public GeodeContainer geodeContainer() {
+		return geode;
+	}
+
+}
--- a/common/function-test-support/src/main/resources/geode/Dockerfile
+++ b/common/function-test-support/src/main/resources/geode/Dockerfile
@@ -0,0 +1,93 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+FROM openjdk:8-jre-alpine
+
+# runtime dependencies
+RUN apk add --no-cache \
+		bash \
+		ncurses
+
+# pub   4096R/ABF4396F 2018-04-12 [expires: 2022-04-12]
+#       8763 31B4 5A97 E382 D1BD  FB44 4482 0F9C ABF4 396F
+# uid       [  undef ] Mike Stolz <mikestolz@apache.org>
+# sub   4096R/3871E6AD 2018-04-12 [expires: 2022-04-12]
+ENV GEODE_GPG D2AE0CA42736AA78E473774DEBC77DA34AADD408
+# TODO does this change per-release like other Apache projects? (and thus needs to be a list of full fingerprints from a KEYS file instead?)
+
+ENV GEODE_HOME /geode
+ENV PATH $PATH:$GEODE_HOME/bin
+
+# https://geode.apache.org/releases/
+ENV GEODE_VERSION 1.12.0
+# Binaries TGZ SHA-256
+# https://dist.apache.org/repos/dist/release/geode/VERSION/apache-geode-VERSION.tgz.sha256
+ENV GEODE_SHA256 063b473dac914aca53c09326487cc96c63ef84eecc8b053c8cc3d5110e82f179
+
+# http://apache.org/dyn/closer.cgi/geode/1.3.0/apache-geode-1.3.0.tgz
+
+RUN set -eux; \
+	apk add --no-cache --virtual .fetch-deps \
+		libressl \
+		gnupg \
+	; \
+	for file in \
+		"geode/$GEODE_VERSION/apache-geode-$GEODE_VERSION.tgz" \
+		"geode/$GEODE_VERSION/apache-geode-$GEODE_VERSION.tgz.asc" \
+	; do \
+		target="$(basename "$file")"; \
+		for url in \
+# https://issues.apache.org/jira/browse/INFRA-8753?focusedCommentId=14735394#comment-14735394
+			"https://www.apache.org/dyn/closer.cgi?action=download&filename=$file" \
+			"https://www-us.apache.org/dist/$file" \
+			"https://www.apache.org/dist/$file" \
+			"https://archive.apache.org/dist/$file" \
+		; do \
+			if wget -O "$target" "$url"; then \
+				break; \
+			fi; \
+		done; \
+	done; \
+	[ -s "apache-geode-$GEODE_VERSION.tgz" ]; \
+	[ -s "apache-geode-$GEODE_VERSION.tgz.asc" ]; \
+	echo "$GEODE_SHA256 *apache-geode-$GEODE_VERSION.tgz" | sha256sum -c -; \
+	export GNUPGHOME="$(mktemp -d)"; \
+	gpg --keyserver ha.pool.sks-keyservers.net --recv-keys "$GEODE_GPG"; \
+	gpg --batch --verify "apache-geode-$GEODE_VERSION.tgz.asc" "apache-geode-$GEODE_VERSION.tgz"; \
+	rm -rf "$GNUPGHOME"; \
+	mkdir /geode; \
+	tar --extract \
+		--file "apache-geode-$GEODE_VERSION.tgz" \
+		--directory /geode \
+		--strip-components 1 \
+	; \
+	rm -rf /geode/javadoc "apache-geode-$GEODE_VERSION.tgz" "apache-geode-$GEODE_VERSION.tgz.asc"; \
+	apk del .fetch-deps; \
+# smoke test to ensure the shell can still run properly after removing temporary deps
+	gfsh version
+
+# Default ports:
+# RMI/JMX 1099
+# REST 8080
+# PULSE 7070
+# LOCATOR 10334
+# CACHESERVER 40404
+ARG CACHE_SERVER_PORT=40404
+ARG LOCATOR_PORT=10334
+
+EXPOSE  8080 ${LOCATOR_PORT} ${CACHE_SERVER_PORT} 1099 7070
+VOLUME ["/data"]
+#CMD ["gfsh"]
--- a/common/geode-common/pom.xml
+++ b/common/geode-common/pom.xml
@@ -0,0 +1,41 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <artifactId>geode-common</artifactId>
+    <version>1.0.0-SNAPSHOT</version>
+    <name>geode-common</name>
+    <description>Geode Common Components</description>
+
+    <parent>
+        <groupId>org.springframework.cloud.fn</groupId>
+        <artifactId>spring-functions-parent</artifactId>
+        <version>1.0.0-SNAPSHOT</version>
+        <relativePath>../../spring-functions-parent</relativePath>
+    </parent>
+    <properties>
+<!--        <geode-starter.version>1.2.7.RELEASE</geode-starter.version>-->
+        <org-json.version>20200518</org-json.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-validation</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.json</groupId>
+            <artifactId>json</artifactId>
+            <version>${org-json.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.data</groupId>
+            <artifactId>spring-data-geode</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-configuration-processor</artifactId>
+            <scope>provided</scope>
+        </dependency>
+    </dependencies>
+
+</project>
--- a/common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeClientCacheConfiguration.java
+++ b/common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeClientCacheConfiguration.java
@@ -0,0 +1,147 @@
+/*
+ * Copyright 2016-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.common.geode;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.Properties;
+
+import org.apache.geode.cache.client.ClientCache;
+import org.apache.geode.cache.client.ClientCacheFactory;
+import org.apache.geode.pdx.ReflectionBasedAutoSerializer;
+
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Import;
+import org.springframework.core.io.Resource;
+import org.springframework.data.gemfire.util.PropertiesBuilder;
+import org.springframework.util.FileCopyUtils;
+import org.springframework.util.StringUtils;
+
+/**
+ * @author David Turanski
+ * @author Christian Tzolov
+ */
+@EnableConfigurationProperties({ GeodeClientCacheProperties.class, GeodeSecurityProperties.class,
+		GeodeSslProperties.class, GeodePoolProperties.class })
+@Import(InetSocketAddressConverterConfiguration.class)
+public class GeodeClientCacheConfiguration {
+
+	private static final String SECURITY_CLIENT = "security-client-auth-init";
+
+	private static final String SECURITY_USERNAME = "security-username";
+
+	private static final String SECURITY_PASSWORD = "security-password";
+
+	@Bean
+	public ClientCache clientCache(GeodeClientCacheProperties clientCacheProperties,
+			GeodeSecurityProperties securityProperties, GeodeSslProperties sslProperties,
+			GeodePoolProperties geodePoolProperties) {
+
+		Properties properties = new Properties();
+		PropertiesBuilder pb = new PropertiesBuilder();
+		if (StringUtils.hasText(securityProperties.getUsername())
+				&& StringUtils.hasText(securityProperties.getPassword())) {
+
+			properties
+					.setProperty(SECURITY_CLIENT,
+							GeodeSecurityProperties.UserAuthInitialize.class.getName() + ".create");
+			properties.setProperty(SECURITY_USERNAME, securityProperties.getUsername());
+			properties.setProperty(SECURITY_PASSWORD, securityProperties.getPassword());
+
+		}
+
+		if (sslProperties.isSslEnabled()) {
+			pb.add(properties);
+			pb.add(this.toGeodeSslProperties(sslProperties));
+		}
+
+		ClientCacheFactory clientCacheFactory = new ClientCacheFactory(pb.build());
+
+		if (clientCacheProperties.isPdxReadSerialized()) {
+			clientCacheFactory.setPdxSerializer(new ReflectionBasedAutoSerializer(".*"));
+			clientCacheFactory.setPdxReadSerialized(true);
+		}
+
+		if (geodePoolProperties.getConnectType().equals(GeodePoolProperties.ConnectType.locator)) {
+			for (InetSocketAddress address : geodePoolProperties.getHostAddresses()) {
+				clientCacheFactory.addPoolLocator(address.getHostName(), address.getPort());
+			}
+		}
+		else {
+			for (InetSocketAddress address : geodePoolProperties.getHostAddresses()) {
+				clientCacheFactory.addPoolServer(address.getHostName(), address.getPort());
+			}
+		}
+
+		clientCacheFactory.setPoolSubscriptionEnabled(geodePoolProperties.isSubscriptionEnabled());
+		ClientCache clientCache = clientCacheFactory.create();
+		clientCache.readyForEvents();
+		return clientCache;
+	}
+
+	/**
+	 * Converts the App Starter properties into Geode native SSL properties.
+	 * @param sslProperties App starter properties.
+	 * @return Returns the geode native SSL properties.
+	 */
+	private Properties toGeodeSslProperties(GeodeSslProperties sslProperties) {
+
+		PropertiesBuilder pb = new PropertiesBuilder();
+
+		// locator - SSL communication with and between locators
+		// server - SSL communication between clients and servers
+		pb.setProperty("ssl-enabled-components", "server,locator");
+
+		pb.setProperty("ssl-keystore", this.resolveRemoteStore(sslProperties.getKeystoreUri(),
+				sslProperties.getUserHomeDirectory(), GeodeSslProperties.LOCAL_KEYSTORE_FILE_NAME));
+		pb.setProperty("ssl-keystore-password", sslProperties.getSslKeystorePassword());
+		pb.setProperty("ssl-keystore-type", sslProperties.getKeystoreType());
+
+		pb.setProperty("ssl-truststore", this.resolveRemoteStore(sslProperties.getTruststoreUri(),
+				sslProperties.getUserHomeDirectory(), GeodeSslProperties.LOCAL_TRUSTSTORE_FILE_NAME));
+		pb.setProperty("ssl-truststore-password", sslProperties.getSslTruststorePassword());
+		pb.setProperty("ssl-truststore-type", sslProperties.getTruststoreType());
+
+		pb.setProperty("ssl-ciphers", sslProperties.getCiphers());
+
+		return pb.build();
+	}
+
+	/**
+	 * Copy the Trust store specified in the URI into a local accessible file.
+	 *
+	 * @param storeUri Either Keystore or Truststore remote resource URI
+	 * @param userHomeDirectory local root directory to store the keystore and localsore files
+	 * @param localStoreFileName local keystore or truststore file name
+	 * @return Returns the absolute path of the local trust or keys store file copy
+	 */
+	private String resolveRemoteStore(Resource storeUri, String userHomeDirectory, String localStoreFileName) {
+
+		File localStoreFile = new File(userHomeDirectory, localStoreFileName);
+		try {
+			FileCopyUtils.copy(storeUri.getInputStream(), new FileOutputStream(localStoreFile));
+			return localStoreFile.getAbsolutePath();
+		}
+		catch (IOException e) {
+			throw new IllegalStateException(String.format("Failed to copy the store from [%s] into %s",
+					storeUri.getDescription(), localStoreFile.getAbsolutePath()), e);
+		}
+	}
+}
--- a/common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeClientCacheProperties.java
+++ b/common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeClientCacheProperties.java
@@ -0,0 +1,43 @@
+/*
+ * Copyright 2019-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.common.geode;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.validation.annotation.Validated;
+
+/**
+ * Geode client pool configuration properties.
+ *
+ * @author Christian Tzolov
+ */
+@ConfigurationProperties("geode.client")
+@Validated
+public class GeodeClientCacheProperties {
+
+	/**
+	 * Deserialize the Geode objects into PdxInstance instead of the domain class.
+	 */
+	private boolean pdxReadSerialized = false;
+
+	public boolean isPdxReadSerialized() {
+		return pdxReadSerialized;
+	}
+
+	public void setPdxReadSerialized(boolean pdxReadSerialized) {
+		this.pdxReadSerialized = pdxReadSerialized;
+	}
+}
--- a/common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeClientRegionConfiguration.java
+++ b/common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeClientRegionConfiguration.java
@@ -0,0 +1,68 @@
+/*
+ * Copyright 2015-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.common.geode;
+
+import java.util.List;
+
+import org.apache.geode.cache.DataPolicy;
+import org.apache.geode.cache.client.ClientCache;
+
+import org.springframework.beans.factory.BeanCreationException;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.data.gemfire.client.ClientRegionFactoryBean;
+import org.springframework.data.gemfire.client.Interest;
+import org.springframework.lang.Nullable;
+import org.springframework.util.CollectionUtils;
+
+/**
+ * Client region configuration common to Geode functions. This configures the
+ * 'regionName', 'spring.application.name' by default and injects the pool. Also, any
+ * beans of type {@link Interest} will be registered to the client region to control which
+ * keys will be automatically synced to the client. At least one of these is required for
+ * Geode Suppliers.
+ *
+ * @author David Turanski
+ */
+@Configuration
+@Import(GeodeClientCacheConfiguration.class)
+@EnableConfigurationProperties(GeodeRegionProperties.class)
+public class GeodeClientRegionConfiguration {
+
+	@Bean(name = "clientRegion")
+	@SuppressWarnings({ "rawtype", "unchecked" })
+	public ClientRegionFactoryBean clientRegionFactoryBean(ClientCache clientCache,
+			@Nullable List<Interest> keyInterests, GeodeRegionProperties properties) {
+		ClientRegionFactoryBean clientRegionFactoryBean = new ClientRegionFactoryBean();
+		clientRegionFactoryBean.setRegionName(properties.getRegionName());
+		clientRegionFactoryBean.setDataPolicy(DataPolicy.EMPTY);
+		if (!CollectionUtils.isEmpty(keyInterests)) {
+			clientRegionFactoryBean.setInterests(keyInterests.toArray(new Interest[keyInterests.size()]));
+		}
+
+		try {
+			clientRegionFactoryBean.setCache(clientCache);
+		}
+		catch (Exception e) {
+			throw new BeanCreationException(e.getMessage(), e);
+		}
+		return clientRegionFactoryBean;
+	}
+
+}
--- a/common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodePoolProperties.java
+++ b/common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodePoolProperties.java
@@ -0,0 +1,90 @@
+/*
+ * Copyright 2015-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.common.geode;
+
+import java.net.InetSocketAddress;
+
+import javax.validation.constraints.NotEmpty;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.validation.annotation.Validated;
+
+/**
+ * Geode client pool configuration properties.
+ *
+ * @author David Turanski
+ */
+@ConfigurationProperties("geode.pool")
+@Validated
+public class GeodePoolProperties {
+
+	/**
+	 * Connection Type locator or server.
+	 */
+	public enum ConnectType {
+		/**
+		 * use locator.
+		 */
+		locator,
+		/**
+		 * use server.
+		 */
+		server
+	}
+
+	/**
+	 * Specifies one or more Gemfire locator or server addresses formatted as [host]:[port].
+	 */
+	private InetSocketAddress[] hostAddresses = { new InetSocketAddress("localhost", 10334) };
+
+	/**
+	 * Specifies connection type: 'server' or 'locator'.
+	 */
+	private ConnectType connectType = ConnectType.locator;
+
+	/**
+	 * Set to true to enable subscriptions for the client pool. Required to sync updates to
+	 * the client cache.
+	 */
+	private boolean subscriptionEnabled;
+
+	@NotEmpty
+	public InetSocketAddress[] getHostAddresses() {
+		return hostAddresses;
+	}
+
+	public void setHostAddresses(InetSocketAddress[] hostAddresses) {
+		this.hostAddresses = hostAddresses;
+	}
+
+	public ConnectType getConnectType() {
+		return connectType;
+	}
+
+	public void setConnectType(ConnectType connectType) {
+		this.connectType = connectType;
+	}
+
+	public boolean isSubscriptionEnabled() {
+		return subscriptionEnabled;
+	}
+
+	public void setSubscriptionEnabled(boolean subscriptionEnabled) {
+		this.subscriptionEnabled = subscriptionEnabled;
+	}
+
+}
--- a/common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeRegionProperties.java
+++ b/common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeRegionProperties.java
@@ -0,0 +1,45 @@
+/*
+ * Copyright 2015-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.common.geode;
+
+import javax.validation.constraints.NotBlank;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.validation.annotation.Validated;
+
+/**
+ * Region configuration properties.
+ * @author David Turanski
+ */
+@ConfigurationProperties("geode.region")
+@Validated
+public class GeodeRegionProperties {
+	/**
+	 * The region name.
+	 */
+	private String regionName;
+
+	@NotBlank(message = "Region name is required")
+	public String getRegionName() {
+		return regionName;
+	}
+
+	public void setRegionName(String regionName) {
+		this.regionName = regionName;
+	}
+
+}
--- a/common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeSecurityProperties.java
+++ b/common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeSecurityProperties.java
@@ -0,0 +1,100 @@
+/*
+ * Copyright 2017-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.common.geode;
+
+import java.util.Properties;
+
+import org.apache.geode.LogWriter;
+import org.apache.geode.distributed.DistributedMember;
+import org.apache.geode.security.AuthInitialize;
+import org.apache.geode.security.AuthenticationFailedException;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+/**
+ * Configuration properties for Geode username/password authentication.
+ * @author David Turanski
+ **/
+@ConfigurationProperties("geode.security")
+public class GeodeSecurityProperties {
+
+	/**
+	 * The cache username.
+	 */
+	private String username;
+
+	/**
+	 * The cache password.
+	 */
+	private String password;
+
+	public String getUsername() {
+		return username;
+	}
+
+	public void setUsername(String username) {
+		this.username = username;
+	}
+
+	public String getPassword() {
+		return password;
+	}
+
+	public void setPassword(String password) {
+		this.password = password;
+	}
+
+	@SuppressWarnings("unused")
+	public static class UserAuthInitialize implements AuthInitialize {
+
+		private LogWriter securitylog;
+		private LogWriter systemlog;
+
+		public static AuthInitialize create() {
+			return new UserAuthInitialize();
+		}
+
+		@Override
+		public void init(LogWriter systemLogger, LogWriter securityLogger) throws AuthenticationFailedException {
+			this.systemlog = systemLogger;
+			this.securitylog = securityLogger;
+		}
+
+		@Override
+		public Properties getCredentials(Properties props, DistributedMember server, boolean isPeer) throws AuthenticationFailedException {
+
+			String username = props.getProperty(SECURITY_USERNAME);
+			if (username == null) {
+				throw new AuthenticationFailedException("UserAuthInitialize: username not set.");
+			}
+
+			String password = props.getProperty(SECURITY_PASSWORD);
+			if (password == null) {
+				throw new AuthenticationFailedException("UserAuthInitialize: password not set.");
+			}
+
+			Properties properties = new Properties();
+			properties.setProperty(SECURITY_USERNAME, username);
+			properties.setProperty(SECURITY_PASSWORD, password);
+			return properties;
+		}
+
+		@Override
+		public void close() {
+		}
+	}
+}
--- a/common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeSslProperties.java
+++ b/common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeSslProperties.java
@@ -0,0 +1,170 @@
+/*
+ * Copyright 2018-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.common.geode;
+
+import javax.validation.constraints.AssertTrue;
+import javax.validation.constraints.NotBlank;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.core.io.Resource;
+import org.springframework.validation.annotation.Validated;
+
+/**
+ * SSL configuration for Geode client applications.
+ * @see <a href=https://geode.apache.org/docs/guide/14/managing/security/implementing_ssl.html>Implementing SSL in Geode</a>
+ *
+ * @author Christian Tzolov
+ */
+@ConfigurationProperties("geode.security.ssl")
+@Validated
+public class GeodeSslProperties {
+
+	private static final String USER_HOME_DIRECTORY = System.getProperty("user.home");
+	/**
+	 * Name of the local trust store file copied in the local file system.
+	 */
+	public static final String LOCAL_TRUSTSTORE_FILE_NAME = "trusted.keystore";
+	/**
+	 * Name of the local trust store file copied in the local file system.
+	 */
+	public static final String LOCAL_KEYSTORE_FILE_NAME = "keystore.keystore";
+
+	/**
+	 * Local directory to cache the truststore and keystore files downloaded form the truststoreUri and keystoreUri locations.
+	 */
+	@NotBlank
+	private String userHomeDirectory = USER_HOME_DIRECTORY;
+
+	/**
+	 * Location of the pre-created truststore URI to be used for connecting to the Geode cluster.
+	 */
+	private Resource truststoreUri;
+
+	/**
+	 * Password for accessing the trust store.
+	 */
+	private String sslTruststorePassword;
+
+	/**
+	 * Identifies the type of truststore used for SSL communications (e.g. JKS, PKCS11, etc.).
+	 */
+	@NotBlank
+	private String truststoreType = "JKS";
+
+	/**
+	 * Location of the pre-created Keystore URI to be used for connecting to the Geode cluster.
+	 */
+	private Resource keystoreUri;
+
+	/**
+	 * Password for accessing the keys truststore.
+	 */
+	private String sslKeystorePassword;
+
+	/**
+	 * Identifies the type of Keystore used for SSL communications (e.g. JKS, PKCS11, etc.).
+	 */
+	@NotBlank
+	private String keystoreType = "JKS";
+
+	/**
+	 * Configures the SSL ciphers used for secure Socket connections as an array of valid cipher names.
+	 */
+	@NotBlank
+	private String ciphers = "any";
+
+	public Resource getTruststoreUri() {
+		return truststoreUri;
+	}
+
+	public void setTruststoreUri(Resource truststoreUri) {
+		this.truststoreUri = truststoreUri;
+	}
+
+	public String getSslKeystorePassword() {
+		return sslKeystorePassword;
+	}
+
+	public void setSslKeystorePassword(String sslKeystorePassword) {
+		this.sslKeystorePassword = sslKeystorePassword;
+	}
+
+	public String getSslTruststorePassword() {
+		return sslTruststorePassword;
+	}
+
+	public void setSslTruststorePassword(String sslTruststorePassword) {
+		this.sslTruststorePassword = sslTruststorePassword;
+	}
+
+	public String getUserHomeDirectory() {
+		return userHomeDirectory;
+	}
+
+	public void setUserHomeDirectory(String userHomeDirectory) {
+		this.userHomeDirectory = userHomeDirectory;
+	}
+
+	public String getTruststoreType() {
+		return truststoreType;
+	}
+
+	public void setTruststoreType(String truststoreType) {
+		this.truststoreType = truststoreType;
+	}
+
+	public Resource getKeystoreUri() {
+		return keystoreUri;
+	}
+
+	public void setKeystoreUri(Resource keystoreUri) {
+		this.keystoreUri = keystoreUri;
+	}
+
+	public String getKeystoreType() {
+		return keystoreType;
+	}
+
+	public void setKeystoreType(String keystoreType) {
+		this.keystoreType = keystoreType;
+	}
+
+	public String getCiphers() {
+		return ciphers;
+	}
+
+	public void setCiphers(String ciphers) {
+		this.ciphers = ciphers;
+	}
+
+	public boolean isSslEnabled() {
+		return this.truststoreUri != null && this.keystoreUri != null;
+	}
+
+	@AssertTrue(message = "The truststoreUri and keystoreUri should together be either empty or not!")
+	private boolean isStoreUrisConsistent() {
+		return ((this.truststoreUri == null) && (this.keystoreUri == null)) ||
+				((this.truststoreUri != null) && (this.keystoreUri != null));
+	}
+
+	@AssertTrue(message = "The sslKeystorePassword and sslKeystorePassword must not be empty for non empty store URIs!")
+	private boolean isStorePasswordRequiredForValidStoreUri() {
+		return (!this.isSslEnabled()) ||
+				((this.sslKeystorePassword != null) && (this.sslTruststorePassword != null));
+	}
+
+}
--- a/common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/InetSocketAddressConverterConfiguration.java
+++ b/common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/InetSocketAddressConverterConfiguration.java
@@ -0,0 +1,49 @@
+/*
+ * Copyright 2018-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.common.geode;
+
+import java.net.InetSocketAddress;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import org.springframework.boot.context.properties.ConfigurationPropertiesBinding;
+import org.springframework.context.annotation.Bean;
+import org.springframework.core.convert.converter.Converter;
+
+public class InetSocketAddressConverterConfiguration {
+	@Bean
+	@ConfigurationPropertiesBinding
+	public Converter<String, InetSocketAddress> inetSocketAddressConverter() {
+		return new InetSocketAddressConverter();
+	}
+
+	public static class InetSocketAddressConverter implements Converter<String, InetSocketAddress> {
+
+		private static final Pattern HOST_AND_PORT_PATTERN = Pattern.compile("^\\s*(.*?):(\\d+)\\s*$");
+
+		@Override
+		public InetSocketAddress convert(String hostAddress) {
+			Matcher m = HOST_AND_PORT_PATTERN.matcher(hostAddress);
+			if (m.matches()) {
+				String host = m.group(1);
+				int port = Integer.parseInt(m.group(2));
+				return new InetSocketAddress(host, port);
+			}
+			throw new IllegalArgumentException(String.format("%s is not a valid [host]:[port] value.", hostAddress));
+		}
+	}
+}
--- a/common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/JsonPdxFunctions.java
+++ b/common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/JsonPdxFunctions.java
@@ -0,0 +1,49 @@
+/*
+ * Copyright 2016-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.common.geode;
+
+import java.util.function.Function;
+
+import org.apache.geode.pdx.JSONFormatter;
+import org.apache.geode.pdx.PdxInstance;
+
+/**
+ * @author David Turanski
+ * @author Christian Tzolov
+ *
+ */
+public abstract class JsonPdxFunctions {
+
+	public static Function<String, PdxInstance> jsonToPdx() {
+		return JSONFormatter::fromJSON;
+	}
+
+	public static Function<PdxInstance, String> pdxToJson() {
+		return obj -> {
+			if (obj == null) {
+				return null;
+			}
+			if (obj instanceof PdxInstance) {
+				String json = JSONFormatter.toJSON(obj);
+				// de-pretty
+				return json.replaceAll("\\r\\n\\s*", "").replaceAll("\\n\\s*", "")
+						.replaceAll("\\s*:\\s*", ":").trim();
+			}
+			return obj.toString();
+		};
+	}
+}
--- a/consumer/geode-consumer/README.adoc
+++ b/consumer/geode-consumer/README.adoc
@@ -0,0 +1,48 @@
+# 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.
--- a/consumer/geode-consumer/pom.xml
+++ b/consumer/geode-consumer/pom.xml
@@ -0,0 +1,70 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <artifactId>geode-consumer</artifactId>
+    <version>1.0.0-SNAPSHOT</version>
+    <name>geode-consumer</name>
+    <description>Geode consumer</description>
+
+    <parent>
+        <groupId>org.springframework.cloud.fn</groupId>
+        <artifactId>spring-functions-parent</artifactId>
+        <version>1.0.0-SNAPSHOT</version>
+        <relativePath>../../spring-functions-parent</relativePath>
+    </parent>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.integration</groupId>
+            <artifactId>spring-integration-gemfire</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-validation</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.cloud.fn</groupId>
+            <artifactId>geode-common</artifactId>
+            <version>${spring-cloud-fn.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.cloud.fn</groupId>
+            <artifactId>config-common</artifactId>
+            <version>${spring-cloud-fn.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-configuration-processor</artifactId>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.junit.vintage</groupId>
+                    <artifactId>junit-vintage-engine</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>io.projectreactor</groupId>
+            <artifactId>reactor-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.cloud.fn</groupId>
+            <artifactId>function-test-support</artifactId>
+            <version>${spring-cloud-fn.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+</project>
--- a/consumer/geode-consumer/src/main/java/org/springframework/cloud/fn/consumer/geode/GeodeConsumerConfiguration.java
+++ b/consumer/geode-consumer/src/main/java/org/springframework/cloud/fn/consumer/geode/GeodeConsumerConfiguration.java
@@ -0,0 +1,61 @@
+/*
+ * Copyright 2020-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.consumer.geode;
+
+import java.util.Collections;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+import org.apache.geode.cache.Region;
+
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.cloud.fn.common.geode.GeodeClientRegionConfiguration;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.integration.gemfire.outbound.CacheWritingMessageHandler;
+import org.springframework.messaging.Message;
+
+/**
+ * Configuration for Geode consumer that writes an entry to a {@link Region} for a
+ * Message, using a SpEL expression for a key, and the payload for the value.
+ * @author David Turanski
+ */
+@Configuration
+@EnableConfigurationProperties(GeodeConsumerProperties.class)
+@Import(GeodeClientRegionConfiguration.class)
+public class GeodeConsumerConfiguration {
+
+	@Bean
+	public Consumer<Message<?>> geodeConsumer(Function<Message<?>, Message<?>> geodeConsumerHandler,
+			CacheWritingMessageHandler cacheWriter) {
+		return message -> cacheWriter.handleMessage(geodeConsumerHandler.apply(message));
+	}
+
+	@Bean
+	Function<Message<?>, Message<?>> geodeConsumerHandler(GeodeConsumerProperties properties) {
+		return new GeodeConsumerHandler(properties.isJson());
+	}
+
+	@Bean
+	CacheWritingMessageHandler cacheWriter(Region<?, ?> region, GeodeConsumerProperties properties) {
+		CacheWritingMessageHandler messageHandler = new CacheWritingMessageHandler(region);
+		messageHandler.setCacheEntries(
+				Collections.singletonMap(properties.getKeyExpression(), "payload"));
+		return messageHandler;
+	}
+}
--- a/consumer/geode-consumer/src/main/java/org/springframework/cloud/fn/consumer/geode/GeodeConsumerHandler.java
+++ b/consumer/geode-consumer/src/main/java/org/springframework/cloud/fn/consumer/geode/GeodeConsumerHandler.java
@@ -0,0 +1,68 @@
+/*
+ * Copyright 2016-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.consumer.geode;
+
+import java.util.function.Function;
+
+import org.apache.geode.pdx.PdxInstance;
+
+import org.springframework.cloud.fn.common.geode.JsonPdxFunctions;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.converter.MessageConversionException;
+
+/**
+ * @author David Turanski
+ * @author Christian Tzolov
+ **/
+class GeodeConsumerHandler implements Function<Message<?>, Message<?>> {
+
+	private final Boolean convertToJson;
+
+	private final Function<String, PdxInstance> transformer = JsonPdxFunctions.jsonToPdx();
+
+	GeodeConsumerHandler(Boolean convertToJson) {
+		this.convertToJson = convertToJson;
+	}
+
+	@Override
+	public Message<?> apply(Message<?> message) {
+		Message<?> transformedMessage = message;
+		Object transformedPayload = message.getPayload();
+		if (convertToJson) {
+
+			Object payload = message.getPayload();
+
+			if (payload instanceof byte[]) {
+				transformedPayload = transformer.apply(new String((byte[]) payload));
+			}
+			else if (payload instanceof String) {
+				transformedPayload = transformer.apply((String) payload);
+			}
+			else {
+				throw new MessageConversionException(String.format(
+						"Cannot convert object of type %s", payload.getClass()
+								.getName()));
+			}
+		}
+
+		return MessageBuilder
+				.fromMessage(message)
+				.withPayload(transformedPayload)
+				.build();
+	}
+}
--- a/consumer/geode-consumer/src/main/java/org/springframework/cloud/fn/consumer/geode/GeodeConsumerProperties.java
+++ b/consumer/geode-consumer/src/main/java/org/springframework/cloud/fn/consumer/geode/GeodeConsumerProperties.java
@@ -0,0 +1,57 @@
+/*
+ * Copyright 2015-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.consumer.geode;
+
+import javax.validation.constraints.NotEmpty;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.validation.annotation.Validated;
+
+/**
+ * @author David Turanski
+ */
+@ConfigurationProperties("geode.consumer")
+@Validated
+public class GeodeConsumerProperties {
+
+	/**
+	 * SpEL expression to use as a cache key.
+	 */
+	private String keyExpression;
+
+	/**
+	 * Indicates if the Geode region stores json objects as PdxInstance.
+	 */
+	private boolean json;
+
+	@NotEmpty(message = "A valid key expression is required")
+	public String getKeyExpression() {
+		return keyExpression;
+	}
+
+	public void setKeyExpression(String keyExpression) {
+		this.keyExpression = keyExpression;
+	}
+
+	public boolean isJson() {
+		return json;
+	}
+
+	public void setJson(boolean json) {
+		this.json = json;
+	}
+}
--- a/consumer/geode-consumer/src/test/java/org/springframework/cloud/fn/consumer/geode/GeodeConsumerApplicationTests.java
+++ b/consumer/geode-consumer/src/test/java/org/springframework/cloud/fn/consumer/geode/GeodeConsumerApplicationTests.java
@@ -0,0 +1,113 @@
+/*
+ * Copyright 2020-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.consumer.geode;
+
+import java.util.function.Consumer;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import org.apache.geode.cache.Region;
+import org.apache.geode.pdx.PdxInstance;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.cloud.fn.test.support.geode.GeodeContainer;
+import org.springframework.cloud.fn.test.support.geode.GeodeContainerIntializer;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.GenericMessage;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class GeodeConsumerApplicationTests {
+
+	private static ApplicationContextRunner applicationContextRunner;
+
+	private static GeodeContainer geode;
+
+	private ObjectMapper objectMapper = new ObjectMapper();
+
+	@BeforeAll
+	static void setup() {
+		GeodeContainerIntializer initializer = new GeodeContainerIntializer(
+				geodeContainer -> {
+					geodeContainer.connectAndExecGfsh("create region --name=Stocks --type=REPLICATE");
+				});
+
+		applicationContextRunner = new ApplicationContextRunner()
+				.withUserConfiguration(GeodeConsumerTestApplication.class);
+
+		geode = initializer.geodeContainer();
+	}
+
+	@Test
+	void consumeWithJsonEnabled() {
+		applicationContextRunner
+				.withPropertyValues(
+						"geode.region.regionName=Stocks",
+						"geode.consumer.json=true",
+						"geode.consumer.key-expression=payload.getField('symbol')",
+						"geode.pool.hostAddresses=" + "localhost:" + geode.getLocatorPort())
+				.run(context -> {
+					Consumer<Message<?>> geodeConsumer = context.getBean("geodeConsumer", Consumer.class);
+
+					String json = objectMapper.writeValueAsString(new Stock("XXX", 100.00));
+					geodeConsumer.accept(new GenericMessage<>(json));
+
+					Region<String, PdxInstance> region = context.getBean(Region.class);
+					PdxInstance instance = region.get("XXX");
+					assertThat(instance.getField("price")).isEqualTo(100.00);
+					region.close();
+				});
+	}
+
+	@Test
+	void consumeWithoutJsonEnabled() {
+		applicationContextRunner
+				.withPropertyValues(
+						"geode.region.regionName=Stocks",
+						"geode.consumer.key-expression='key'",
+						"geode.pool.hostAddresses=" + "localhost:" + geode.getLocatorPort())
+				.run(context -> {
+					Consumer<Message<?>> geodeConsumer = context.getBean("geodeConsumer", Consumer.class);
+
+					geodeConsumer.accept(new GenericMessage<>("value"));
+
+					Region<String, String> region = context.getBean(Region.class);
+					String value = region.get("key");
+					assertThat(value).isEqualTo("value");
+					region.close();
+				});
+	}
+
+	@Data
+	@AllArgsConstructor
+	@NoArgsConstructor
+	static class Stock {
+		private String symbol;
+
+		private double price;
+	}
+
+	@SpringBootApplication
+	static class GeodeConsumerTestApplication {
+	}
+
+}
--- a/pom.xml
+++ b/pom.xml
@@ -44,13 +44,16 @@
 		<module>common/config-common</module>
 		<module>common/ftp-common</module>
 		<module>common/function-test-support</module>
-		<module>common/tcp-common</module>
+		<module>common/geode-common</module>
 		<module>common/mqtt-common</module>
+		<module>common/tcp-common</module>
+

 		<module>consumer/cassandra-consumer</module>
 		<module>consumer/counter-consumer</module>
 		<module>consumer/file-consumer</module>
 		<module>consumer/ftp-consumer</module>
+		<module>consumer/geode-consumer</module>
 		<module>consumer/jdbc-consumer</module>
 		<module>consumer/log-consumer</module>
 		<module>consumer/mongodb-consumer</module>
@@ -68,6 +71,7 @@
 		<module>function/splitter-function</module>

 		<module>supplier/file-supplier</module>
+		<module>supplier/geode-supplier</module>
 		<module>supplier/http-supplier</module>
 		<module>supplier/jdbc-supplier</module>
 		<module>supplier/mongodb-supplier</module>
--- a/spring-functions-parent/pom.xml
+++ b/spring-functions-parent/pom.xml
@@ -17,6 +17,7 @@
 	<properties>
 		<spring-boot.version>2.3.0.RELEASE</spring-boot.version>
 		<spring-cloud-function.version>3.0.7.RELEASE</spring-cloud-function.version>
+		<spring-cloud-fn.version>${project.version}</spring-cloud-fn.version>
 	</properties>
 	<dependencyManagement>
 		<dependencies>
--- a/supplier/geode-supplier/README.adoc
+++ b/supplier/geode-supplier/README.adoc
@@ -0,0 +1,68 @@
+# Geode Supplier
+
+This module provides a `java.util.function.Supplier` that can be reused and composed in other applications.
+The `Supplier` configures an Apache Geode client that connects to an external Apache Geode cache server or locator to monitor an existing region.
+If `geode.supplier.query` is provided, the supplier will create a continuous query on the region and publish any events that meet the select criteria.
+If no query is provided, the supplier will publish all create and update events on the region.
+
+## Continuous Query
+If a query is provided, the supplier uses the `ContinuousQueryMessageProducer` from `Spring Integration` which wraps a https://docs.spring.io/spring-data/gemfire/docs/current/api/org/springframework/data/gemfire/listener/ContinuousQueryListenerContainer.html[ContinuousQueryListenerContainer]
+and emits a reactive stream of objects, extracted from a Geode https://geode.apache.org/releases/latest/javadoc/org/apache/geode/cache/CqEvent.html[CqEvent], which holds all of the
+event details.
+A SpEl Expression, given by the property `geode.cq.supplier.event-expression` is used to extract desired fields from the CQEvent payload.
+The default expression is `newValue` which returns the current value from the configured Region.
+
+## Cache Listener
+
+If no query is provided, the supplier uses the `CacheListeningMessageProducer` from `Spring Integration` which wraps a https://geode.apache.org/releases/latest/javadoc/org/apache/geode/cache/CacheListener.html[CacheListener]
+and emits a reactive stream of objects, extracted from a Geode https://geode.apache.org/releases/latest/javadoc/org/apache/geode/cache/EntryEvent.html[EntryEvent], which holds all of the
+event details.
+A SpEl Expression, given by the property `geode.supplier.entry-event-expression` is used to extract desired fields from the EntryEvent payload.
+The default expression is `newValue` which returns the current value from the configured Region.
+
+NOTE: Retrieving the value by itself is not always sufficient, especially if it does not contain the key value, or any additional context.
+The key is referenced by the field `key`.
+If the cached key and value types are primitives, a simple expression like `key + ':' +newValue` may be useful.
+To access the entire EntryEvent, set the expression to `#root` or `#this`.
+
+The configured MessageProducer emits objects to the supplier implemented as `Supplier<Flux<?>>`.
+Users have to subscribe to the returned `Flux` to receive the data.
+
+## 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 and you set  `geode.client.pdx-read-serialized` to `true`, PdxInstance objects will be returned as JSON strings.
+
+## Beans for injection
+
+You can import the `GeodeSupplierConfiguration` configuration in a Spring Boot application and then inject the `geodeSupplier` bean as type `Supplier<Flux<T>>`, where `T` is the expected return type.
+If necessary, can use the bean name `geodeSupplier` as a qualifier.
+
+Once injected, you can invoke the `get` method of the `Supplier` and then subscribe to the returned `Flux` to initiate the stream.
+
+## 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/supplier/geode/GeodeSupplierProperties.java[GeodeSupplierProperties.java] (`geode.supplier`)
+
+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/GeodeClientCacheProperties.java[GeodeClientCacheProperties.java] (`geode.client`)
+* 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/supplier/geode/GeodeSupplierApplicationTests.java[test suite] for examples of how this supplier is used.
+
+## Other usage
+
+See this link:../../../applications/source/geode-source/README.adoc[README] where this supplier is used to create a Spring Cloud Stream application to emit entry event data.
--- a/supplier/geode-supplier/pom.xml
+++ b/supplier/geode-supplier/pom.xml
@@ -0,0 +1,76 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://maven.apache.org/POM/4.0.0"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <artifactId>geode-supplier</artifactId>
+    <version>1.0.0-SNAPSHOT</version>
+    <name>geode-supplier</name>
+    <description>geode supplier</description>
+
+    <parent>
+        <groupId>org.springframework.cloud.fn</groupId>
+        <artifactId>spring-functions-parent</artifactId>
+        <version>1.0.0-SNAPSHOT</version>
+        <relativePath>../../spring-functions-parent</relativePath>
+    </parent>
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.integration</groupId>
+            <artifactId>spring-integration-gemfire</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.cloud.fn</groupId>
+            <artifactId>geode-common</artifactId>
+            <version>${spring-cloud-fn.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.cloud.fn</groupId>
+            <artifactId>config-common</artifactId>
+            <version>${spring-cloud-fn.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.hibernate.validator</groupId>
+            <artifactId>hibernate-validator</artifactId>
+            <optional>true</optional>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-configuration-processor</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.cloud.fn</groupId>
+            <artifactId>function-test-support</artifactId>
+            <version>${spring-cloud-fn.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>io.projectreactor</groupId>
+            <artifactId>reactor-test</artifactId>
+		</dependency>
+
+	</dependencies>
+
+</project>
--- a/supplier/geode-supplier/src/main/java/org/springframework/cloud/fn/supplier/geode/GeodeSupplierConfiguration.java
+++ b/supplier/geode-supplier/src/main/java/org/springframework/cloud/fn/supplier/geode/GeodeSupplierConfiguration.java
@@ -0,0 +1,194 @@
+/*
+ * Copyright 2015-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.supplier.geode;
+
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+import org.apache.geode.cache.Region;
+import org.apache.geode.cache.client.ClientCache;
+import org.apache.geode.pdx.PdxInstance;
+import reactor.core.publisher.EmitterProcessor;
+import reactor.core.publisher.Flux;
+
+import org.springframework.beans.factory.BeanCreationException;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.cloud.fn.common.geode.GeodeClientRegionConfiguration;
+import org.springframework.cloud.fn.common.geode.JsonPdxFunctions;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Condition;
+import org.springframework.context.annotation.ConditionContext;
+import org.springframework.context.annotation.Conditional;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.context.annotation.PropertySource;
+import org.springframework.core.type.AnnotatedTypeMetadata;
+import org.springframework.data.gemfire.client.Interest;
+import org.springframework.data.gemfire.listener.ContinuousQueryListenerContainer;
+import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.channel.FluxMessageChannel;
+import org.springframework.integration.core.MessageProducer;
+import org.springframework.integration.dsl.IntegrationFlow;
+import org.springframework.integration.dsl.IntegrationFlows;
+import org.springframework.integration.gemfire.inbound.CacheListeningMessageProducer;
+import org.springframework.integration.gemfire.inbound.ContinuousQueryMessageProducer;
+import org.springframework.integration.router.PayloadTypeRouter;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.util.StringUtils;
+
+/**
+ * The Geode Supplier which publishes a message for each event matching the criteria for a
+ * configured Region. If a query is provided, this will use a
+ * {@link ContinuousQueryMessageProducer} to publish
+ * {@link org.apache.geode.cache.query.CqEvent}s that match the query selection criteria.
+ * If no query is provided, this will use a {@link CacheListeningMessageProducer} to
+ * publsh {@link org.apache.geode.cache.EntryEvent}s.
+ *
+ * A SpEl Expression, given by the property 'eventExpression' is evaluated to extract
+ * desired information from the payload. The default expression is 'newValue' which
+ * returns the current object value. This may not be ideal for every use case especially
+ * if it does not provide the key value. The key is referenced by the field 'key'. If the
+ * cached key and value types are primitives, an simple expression like "key + ':' +
+ * newValue" will work. Additional available depend on the specific event type.
+ *
+ * More complex transformations, such as Json, will require customization. To access the
+ * original object, set 'entryExpression' to '#root' or "#this'.
+ *
+ * This converts payloads of type {@link PdxInstance}, which Geode uses to store JSON
+ * content (the type of 'newValue' for instance), to a JSON String.
+ *
+ *
+ * @author David Turanski
+ */
+
+@Import(GeodeClientRegionConfiguration.class)
+@Configuration
+@PropertySource("classpath:geode-client.properties")
+@EnableConfigurationProperties(GeodeSupplierProperties.class)
+public class GeodeSupplierConfiguration {
+
+	private EmitterProcessor<Message<?>> cacheEvents = EmitterProcessor.create();
+
+	@Bean
+	public Supplier<Flux<?>> geodeSupplier() {
+		return () -> cacheEvents.map(Message::getPayload);
+	}
+
+	@Bean
+	FluxMessageChannel fluxChannel() {
+		FluxMessageChannel fluxChannel = new FluxMessageChannel();
+		fluxChannel.subscribe(cacheEvents);
+		return fluxChannel;
+	}
+
+	@Bean
+	MessageChannel convertToStringChannel() {
+		return new DirectChannel();
+	}
+
+	@Bean
+	MessageChannel routerChannel() {
+		return new DirectChannel();
+	}
+
+	@Bean
+	PayloadTypeRouter payloadTypeRouter(FluxMessageChannel fluxChannel) {
+		PayloadTypeRouter router = new PayloadTypeRouter();
+		router.setDefaultOutputChannel(fluxChannel);
+		router.setChannelMapping(PdxInstance.class.getName(), "convertToStringChannel");
+
+		return router;
+	}
+
+	@ConditionalOnMissingBean(Interest.class)
+	@Bean
+	public Interest<?> allKeysInterest() {
+		return new Interest<>(Interest.ALL_KEYS);
+	}
+
+	@Bean
+	IntegrationFlow startFlow(MessageChannel routerChannel, PayloadTypeRouter payloadTypeRouter) {
+		return IntegrationFlows.from(routerChannel)
+				.route(payloadTypeRouter)
+				.get();
+	}
+
+	@Bean
+	Function<PdxInstance, String> pdxToJson() {
+		return JsonPdxFunctions.pdxToJson();
+	}
+
+	@Bean
+	IntegrationFlow convertToString(MessageChannel convertToStringChannel, FluxMessageChannel fluxChannel,
+			Function<PdxInstance, String> pdxToJson) {
+		return IntegrationFlows.from(convertToStringChannel)
+				.transform(pdxToJson)
+				.channel(fluxChannel)
+				.get();
+	}
+
+	@Bean
+	@Conditional({ CQNotEnabled.class }) // ConditionalOnExpression causing SpEL errors.
+	public MessageProducer cacheListeningMessageProducer(
+			MessageChannel routerChannel,
+			GeodeSupplierProperties properties, Region<?, ?> region) {
+		CacheListeningMessageProducer cacheListeningMessageProducer = new CacheListeningMessageProducer(region);
+		cacheListeningMessageProducer.setOutputChannel(routerChannel);
+		cacheListeningMessageProducer.setPayloadExpression(
+				properties.getEventExpression());
+		return cacheListeningMessageProducer;
+	}
+
+	@ConditionalOnProperty(prefix = "geode.supplier", name = "query")
+	static class CQConfiguration {
+		@Bean
+		MessageProducer continuousQueryListener(
+				MessageChannel routerChannel,
+				ContinuousQueryListenerContainer continuousQueryListenerContainer,
+				GeodeSupplierProperties properties) {
+			ContinuousQueryMessageProducer continuousQueryMessageProducer = new ContinuousQueryMessageProducer(
+					continuousQueryListenerContainer,
+					properties.getQuery());
+			continuousQueryMessageProducer.setPayloadExpression(properties.getEventExpression());
+			continuousQueryMessageProducer.setOutputChannel(routerChannel);
+			return continuousQueryMessageProducer;
+		}
+
+		@Bean
+		ContinuousQueryListenerContainer continuousQueryListenerContainer(ClientCache clientCache) {
+			ContinuousQueryListenerContainer continuousQueryListenerContainer = new ContinuousQueryListenerContainer();
+			try {
+				continuousQueryListenerContainer.setCache(clientCache);
+			}
+			catch (Exception e) {
+				throw new BeanCreationException(e.getLocalizedMessage(), e);
+			}
+			return continuousQueryListenerContainer;
+		}
+	}
+
+	static class CQNotEnabled implements Condition {
+
+		@Override
+		public boolean matches(ConditionContext context, AnnotatedTypeMetadata metadata) {
+			return StringUtils.isEmpty(context.getEnvironment().getProperty("geode.supplier.query"));
+		}
+	}
+}
--- a/supplier/geode-supplier/src/main/java/org/springframework/cloud/fn/supplier/geode/GeodeSupplierProperties.java
+++ b/supplier/geode-supplier/src/main/java/org/springframework/cloud/fn/supplier/geode/GeodeSupplierProperties.java
@@ -0,0 +1,61 @@
+/*
+ * Copyright 2020-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.supplier.geode;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.expression.Expression;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
+
+/**
+ * This represents the configuration properties for the Gemfire Supplier.
+ *
+ * @author David Turanski
+ */
+@ConfigurationProperties("geode.supplier")
+public class GeodeSupplierProperties {
+
+	private static final String DEFAULT_EXPRESSION = "newValue";
+
+	private final SpelExpressionParser parser = new SpelExpressionParser();
+
+	/**
+	 * SpEL expression to extract data from an {@link org.apache.geode.cache.EntryEvent} or
+	 * {@link org.apache.geode.cache.query.CqEvent}.
+	 */
+	private Expression eventExpression = parser.parseExpression(DEFAULT_EXPRESSION);
+
+	/**
+	 * An OQL query. This will enable continuous query if provided.
+	 */
+	private String query;
+
+	public Expression getEventExpression() {
+		return eventExpression;
+	}
+
+	public void setEventExpression(Expression eventExpression) {
+		this.eventExpression = eventExpression;
+	}
+
+	public String getQuery() {
+		return query;
+	}
+
+	public void setQuery(String query) {
+		this.query = query;
+	}
+}
--- a/supplier/geode-supplier/src/main/resources/geode-client.properties
+++ b/supplier/geode-supplier/src/main/resources/geode-client.properties
@@ -0,0 +1,4 @@
+#
+# Geode client pool must be configured with subscriptions enabled for this component.
+#
+geode.pool.subscriptionEnabled=true
--- a/supplier/geode-supplier/src/test/java/org/springframework/cloud/fn/supplier/geode/GeodeSupplierApplicationTests.java
+++ b/supplier/geode-supplier/src/test/java/org/springframework/cloud/fn/supplier/geode/GeodeSupplierApplicationTests.java
@@ -0,0 +1,190 @@
+/*
+ * Copyright 2020-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.supplier.geode;
+
+import java.time.Duration;
+import java.util.function.Supplier;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import org.apache.geode.cache.EntryEvent;
+import org.apache.geode.cache.Region;
+import org.apache.geode.pdx.PdxInstance;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Flux;
+import reactor.test.StepVerifier;
+
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.cloud.fn.common.geode.JsonPdxFunctions;
+import org.springframework.cloud.fn.test.support.geode.GeodeContainer;
+import org.springframework.cloud.fn.test.support.geode.GeodeContainerIntializer;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.fail;
+
+public class GeodeSupplierApplicationTests {
+
+	private static ApplicationContextRunner applicationContextRunner;
+
+	private static GeodeContainer geode;
+
+	private ObjectMapper objectMapper = new ObjectMapper();
+
+	@BeforeAll
+	static void setup() {
+		GeodeContainerIntializer initializer = new GeodeContainerIntializer(
+				geodeContainer -> {
+					geodeContainer.connectAndExecGfsh("create region --name=myRegion --type=REPLICATE");
+				});
+
+		applicationContextRunner = new ApplicationContextRunner()
+				.withUserConfiguration(GeodeSupplierTestApplication.class);
+
+		geode = initializer.geodeContainer();
+	}
+
+	@Test
+	void getServerEntryEvents() {
+		applicationContextRunner
+				.withPropertyValues("geode.region.regionName=myRegion",
+						"geode.supplier.event-expression=#root",
+						"geode.pool.hostAddresses=" + "localhost:" + geode.getLocatorPort())
+				.run(context -> {
+					geode.connectAndExecGfsh(
+							"put --key=hello --value=world --region=myRegion",
+							"put --key=foo --value=bar --region=myRegion",
+							"put --key=hello --value=dave --region=myRegion");
+
+					Supplier<Flux<EntryEvent>> geodeSupplier = context.getBean("geodeSupplier", Supplier.class);
+
+					StepVerifier.create(geodeSupplier.get()).assertNext(cacheEvent -> {
+						assertThat(cacheEvent.getOperation().isCreate()).isTrue();
+						assertThat(cacheEvent.getKey()).isEqualTo("hello");
+						assertThat(cacheEvent.getNewValue()).isEqualTo("world");
+					}).assertNext(cacheEvent -> {
+						assertThat(cacheEvent.getOperation().isCreate()).isTrue();
+						assertThat(cacheEvent.getKey()).isEqualTo("foo");
+						assertThat(cacheEvent.getNewValue()).isEqualTo("bar");
+					}).assertNext(cacheEvent -> {
+						assertThat(cacheEvent.getOperation().isUpdate()).isTrue();
+						assertThat(cacheEvent.getKey()).isEqualTo("hello");
+						assertThat(cacheEvent.getNewValue()).isEqualTo("dave");
+
+					}).thenCancel().verify(Duration.ofSeconds(10));
+				});
+	}
+
+	@Test
+	void pdxReadSerialized() {
+		applicationContextRunner
+				.withPropertyValues(
+						"geode.region.regionName=myRegion",
+						"geode.client.pdx-read-serialized=true",
+						"geode.pool.hostAddresses=" + "localhost:" + geode.getLocatorPort())
+				.run(context -> {
+					Supplier<Flux<String>> geodeSupplier = context.getBean("geodeSupplier", Supplier.class);
+					// Using local region here
+					Region<String, PdxInstance> region = context.getBean(Region.class);
+					Stock stock = new Stock("XXX", 140.00);
+					ObjectMapper objectMapper = new ObjectMapper();
+					String json = objectMapper.writeValueAsString(stock);
+					region.put(stock.getSymbol(), JsonPdxFunctions.jsonToPdx().apply(json));
+
+					StepVerifier.create(geodeSupplier.get()).assertNext(val -> {
+						try {
+							assertThat(objectMapper.readValue(val, Stock.class)).isEqualTo(stock);
+						}
+						catch (JsonProcessingException e) {
+							fail(e.getMessage());
+						}
+					}).thenCancel().verify(Duration.ofSeconds(10));
+				});
+	}
+
+	@Test
+	void connectTypeServer() {
+		applicationContextRunner
+				.withPropertyValues("geode.region.regionName=myRegion",
+						"geode.pool.connect-type=server",
+						"geode.supplier.event-expression=key+':'+newValue",
+						"geode.pool.hostAddresses=" + "localhost:" + geode.getCacheServerPort())
+				.run(context -> {
+
+					// Using local region here since it's faster
+					Region<String, String> region = context.getBean(Region.class);
+
+					region.put("foo", "bar");
+					Supplier<Flux<String>> geodeSupplier = context.getBean("geodeSupplier", Supplier.class);
+
+					StepVerifier.create(geodeSupplier.get()).assertNext(val -> {
+						assertThat(val).isEqualTo("foo:bar");
+					}).thenCancel().verify(Duration.ofSeconds(10));
+				});
+	}
+
+	@Test
+	void continuousQuery() {
+		applicationContextRunner
+				.withPropertyValues(
+						"geode.region.regionName=myRegion",
+						"geode.client.pdx-read-serialized=true",
+						"geode.supplier.query=Select * from /myRegion where symbol='XXX' and price > 140",
+						"geode.pool.hostAddresses=" + "localhost:" + geode.getLocatorPort())
+				.run(context -> {
+					Supplier<Flux<String>> geodeCqSupplier = context.getBean("geodeSupplier", Supplier.class);
+					// Using local region here
+					Region<String, PdxInstance> region = context.getBean(Region.class);
+					putStockEvent(region, new Stock("XXX", 140.00));
+					putStockEvent(region, new Stock("XXX", 140.20));
+					putStockEvent(region, new Stock("YYY", 110.00));
+					putStockEvent(region, new Stock("YYY", 110.01));
+					putStockEvent(region, new Stock("XXX", 139.80));
+
+					StepVerifier.create(geodeCqSupplier.get()).assertNext(val -> {
+						try {
+							assertThat(objectMapper.readValue(val, Stock.class)).isEqualTo(new Stock("XXX", 140.20));
+						}
+						catch (JsonProcessingException e) {
+							fail(e.getMessage());
+						}
+					}).thenCancel().verify(Duration.ofSeconds(10));
+				});
+	}
+
+	private void putStockEvent(Region<String, PdxInstance> region, Stock stock) throws JsonProcessingException {
+		String json = objectMapper.writeValueAsString(stock);
+		region.put(stock.getSymbol(), JsonPdxFunctions.jsonToPdx().apply(json));
+	}
+
+	@Data
+	@AllArgsConstructor
+	@NoArgsConstructor
+	static class Stock {
+		private String symbol;
+
+		private double price;
+	}
+
+	@SpringBootApplication
+	static class GeodeSupplierTestApplication {
+	}
+}