diff --git a/common/function-test-support/pom.xml b/common/function-test-support/pom.xml index e0262c11..9fd267dc 100644 --- a/common/function-test-support/pom.xml +++ b/common/function-test-support/pom.xml @@ -15,13 +15,16 @@ 1.6.0 + 1.14.2 + 1.1.1 + 1.2.7.RELEASE org.apache.ftpserver ftpserver-core - 1.1.1 + ${apache-ftpserver.version} compile @@ -44,6 +47,16 @@ sshd-core ${sshd-core.version} + + org.testcontainers + testcontainers + ${test-containers.version} + + + org.springframework.data + spring-data-geode + true + diff --git 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 new file mode 100644 index 00000000..524547e9 --- /dev/null +++ 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}. 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 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 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 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, ","); + } + } + } + +} diff --git 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 new file mode 100644 index 00000000..d3bb66ce --- /dev/null +++ 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} post processor if provided. + * @author David Turanski + */ +public class GeodeContainerIntializer { + + private int locatorPort; + + private int cacheServerPort; + + private GeodeContainer geode; + + private Optional> postProcessor; + + /** + * Create, start, and perform post processing on a {@link GeodeContainer}. + * @param postProcessor a {@code Consumer} to run after the container is started. + */ + public GeodeContainerIntializer(Consumer 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 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; + } + +} diff --git a/common/function-test-support/src/main/resources/geode/Dockerfile b/common/function-test-support/src/main/resources/geode/Dockerfile new file mode 100644 index 00000000..8f71755e --- /dev/null +++ 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 +# 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"] \ No newline at end of file diff --git a/common/geode-common/pom.xml b/common/geode-common/pom.xml new file mode 100644 index 00000000..b6268ddb --- /dev/null +++ b/common/geode-common/pom.xml @@ -0,0 +1,41 @@ + + + 4.0.0 + geode-common + 1.0.0-SNAPSHOT + geode-common + Geode Common Components + + + org.springframework.cloud.fn + spring-functions-parent + 1.0.0-SNAPSHOT + ../../spring-functions-parent + + + + 20200518 + + + + + org.springframework.boot + spring-boot-starter-validation + + + org.json + json + ${org-json.version} + + + org.springframework.data + spring-data-geode + + + org.springframework.boot + spring-boot-configuration-processor + provided + + + + diff --git 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 new file mode 100644 index 00000000..b64035c9 --- /dev/null +++ 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); + } + } +} diff --git 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 new file mode 100644 index 00000000..9a36c63b --- /dev/null +++ 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; + } +} diff --git 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 new file mode 100644 index 00000000..879f24ac --- /dev/null +++ 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 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; + } + +} diff --git 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 new file mode 100644 index 00000000..0bcba0fd --- /dev/null +++ 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; + } + +} diff --git 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 new file mode 100644 index 00000000..c4e94a53 --- /dev/null +++ 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; + } + +} diff --git 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 new file mode 100644 index 00000000..07c9c72c --- /dev/null +++ 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() { + } + } +} diff --git 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 new file mode 100644 index 00000000..49723e08 --- /dev/null +++ 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 Implementing SSL in Geode + * + * @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)); + } + +} diff --git 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 new file mode 100644 index 00000000..407781af --- /dev/null +++ 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 inetSocketAddressConverter() { + return new InetSocketAddressConverter(); + } + + public static class InetSocketAddressConverter implements Converter { + + 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)); + } + } +} diff --git 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 new file mode 100644 index 00000000..f637824c --- /dev/null +++ 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 jsonToPdx() { + return JSONFormatter::fromJSON; + } + + public static Function 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(); + }; + } +} diff --git a/consumer/geode-consumer/README.adoc b/consumer/geode-consumer/README.adoc new file mode 100644 index 00000000..3cfce0bb --- /dev/null +++ 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>`. +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. \ No newline at end of file diff --git a/consumer/geode-consumer/pom.xml b/consumer/geode-consumer/pom.xml new file mode 100644 index 00000000..44095248 --- /dev/null +++ b/consumer/geode-consumer/pom.xml @@ -0,0 +1,70 @@ + + + 4.0.0 + geode-consumer + 1.0.0-SNAPSHOT + geode-consumer + Geode consumer + + + org.springframework.cloud.fn + spring-functions-parent + 1.0.0-SNAPSHOT + ../../spring-functions-parent + + + + + org.springframework.integration + spring-integration-gemfire + + + org.springframework.boot + spring-boot-starter-validation + + + org.springframework.cloud.fn + geode-common + ${spring-cloud-fn.version} + + + org.springframework.cloud.fn + config-common + ${spring-cloud-fn.version} + + + org.springframework.boot + spring-boot-configuration-processor + provided + + + org.springframework.boot + spring-boot-starter-test + test + + + org.junit.vintage + junit-vintage-engine + + + + + io.projectreactor + reactor-test + test + + + org.springframework.cloud.fn + function-test-support + ${spring-cloud-fn.version} + test + + + org.projectlombok + lombok + test + + + + diff --git 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 new file mode 100644 index 00000000..b4a0e1e4 --- /dev/null +++ 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> geodeConsumer(Function, Message> geodeConsumerHandler, + CacheWritingMessageHandler cacheWriter) { + return message -> cacheWriter.handleMessage(geodeConsumerHandler.apply(message)); + } + + @Bean + Function, 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; + } +} diff --git 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 new file mode 100644 index 00000000..d7c5e660 --- /dev/null +++ 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> { + + private final Boolean convertToJson; + + private final Function 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(); + } +} diff --git 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 new file mode 100644 index 00000000..221da527 --- /dev/null +++ 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; + } +} diff --git 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 new file mode 100644 index 00000000..639be1e8 --- /dev/null +++ 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> geodeConsumer = context.getBean("geodeConsumer", Consumer.class); + + String json = objectMapper.writeValueAsString(new Stock("XXX", 100.00)); + geodeConsumer.accept(new GenericMessage<>(json)); + + Region 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> geodeConsumer = context.getBean("geodeConsumer", Consumer.class); + + geodeConsumer.accept(new GenericMessage<>("value")); + + Region 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 { + } + +} diff --git a/pom.xml b/pom.xml index a6ca032f..f582e55d 100644 --- a/pom.xml +++ b/pom.xml @@ -44,13 +44,16 @@ common/config-common common/ftp-common common/function-test-support - common/tcp-common + common/geode-common common/mqtt-common + common/tcp-common + consumer/cassandra-consumer consumer/counter-consumer consumer/file-consumer consumer/ftp-consumer + consumer/geode-consumer consumer/jdbc-consumer consumer/log-consumer consumer/mongodb-consumer @@ -68,6 +71,7 @@ function/splitter-function supplier/file-supplier + supplier/geode-supplier supplier/http-supplier supplier/jdbc-supplier supplier/mongodb-supplier diff --git a/spring-functions-parent/pom.xml b/spring-functions-parent/pom.xml index 4dcb87e2..cdeea218 100644 --- a/spring-functions-parent/pom.xml +++ b/spring-functions-parent/pom.xml @@ -17,6 +17,7 @@ 2.3.0.RELEASE 3.0.7.RELEASE + ${project.version} diff --git a/supplier/geode-supplier/README.adoc b/supplier/geode-supplier/README.adoc new file mode 100644 index 00000000..b1a5208b --- /dev/null +++ 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>`. +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>`, 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. \ No newline at end of file diff --git a/supplier/geode-supplier/pom.xml b/supplier/geode-supplier/pom.xml new file mode 100644 index 00000000..ea76900c --- /dev/null +++ b/supplier/geode-supplier/pom.xml @@ -0,0 +1,76 @@ + + + 4.0.0 + geode-supplier + 1.0.0-SNAPSHOT + geode-supplier + geode supplier + + + org.springframework.cloud.fn + spring-functions-parent + 1.0.0-SNAPSHOT + ../../spring-functions-parent + + + + org.springframework.boot + spring-boot-starter + + + + org.springframework.integration + spring-integration-gemfire + + + + org.springframework.cloud.fn + geode-common + ${spring-cloud-fn.version} + + + + org.springframework.cloud.fn + config-common + ${spring-cloud-fn.version} + + + + org.hibernate.validator + hibernate-validator + true + + + + org.springframework.boot + spring-boot-configuration-processor + + + + org.springframework.boot + spring-boot-starter-test + test + + + + org.springframework.cloud.fn + function-test-support + ${spring-cloud-fn.version} + test + + + + org.projectlombok + lombok + test + + + + io.projectreactor + reactor-test + + + + + diff --git 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 new file mode 100644 index 00000000..af27f90f --- /dev/null +++ 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> cacheEvents = EmitterProcessor.create(); + + @Bean + public Supplier> 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 pdxToJson() { + return JsonPdxFunctions.pdxToJson(); + } + + @Bean + IntegrationFlow convertToString(MessageChannel convertToStringChannel, FluxMessageChannel fluxChannel, + Function 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")); + } + } +} diff --git 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 new file mode 100644 index 00000000..9e08f86e --- /dev/null +++ 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; + } +} diff --git a/supplier/geode-supplier/src/main/resources/geode-client.properties b/supplier/geode-supplier/src/main/resources/geode-client.properties new file mode 100644 index 00000000..d4de889e --- /dev/null +++ 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 diff --git 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 new file mode 100644 index 00000000..66220bab --- /dev/null +++ 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> 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> geodeSupplier = context.getBean("geodeSupplier", Supplier.class); + // Using local region here + Region 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 region = context.getBean(Region.class); + + region.put("foo", "bar"); + Supplier> 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> geodeCqSupplier = context.getBean("geodeSupplier", Supplier.class); + // Using local region here + Region 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 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 { + } +}