From c64b259d914b7165849ab28c201ca1f08efc01b4 Mon Sep 17 00:00:00 2001 From: Corneil du Plessis Date: Fri, 7 Oct 2022 15:22:57 +0200 Subject: [PATCH] [SB3 Update] Remove Apache Geode (#345) - Also update SFTP usage --- .../fn/test/support/geode/GeodeContainer.java | 197 ------------ .../geode/GeodeContainerIntializer.java | 112 ------- .../fn/test/support/geode/SocketUtils.java | 302 ------------------ common/geode-common/pom.xml | 32 -- .../geode/GeodeClientCacheConfiguration.java | 147 --------- .../geode/GeodeClientCacheProperties.java | 43 --- .../geode/GeodeClientRegionConfiguration.java | 68 ---- .../fn/common/geode/GeodePoolProperties.java | 90 ------ .../common/geode/GeodeRegionProperties.java | 45 --- .../common/geode/GeodeSecurityProperties.java | 100 ------ .../fn/common/geode/GeodeSslProperties.java | 170 ---------- ...etSocketAddressConverterConfiguration.java | 49 --- .../fn/common/geode/JsonPdxFunctions.java | 49 --- common/metadata-store-common/pom.xml | 17 - .../store/ClientCacheAutoConfiguration.java | 44 --- .../store/MetadataStoreAutoConfiguration.java | 35 +- .../store/MetadataStoreProperties.java | 26 +- .../MetadataStoreAutoConfigurationTests.java | 3 +- common/pom.xml | 1 - consumer/geode-consumer/README.adoc | 48 --- consumer/geode-consumer/pom.xml | 44 --- .../geode/GeodeConsumerConfiguration.java | 68 ---- .../consumer/geode/GeodeConsumerHandler.java | 68 ---- .../geode/GeodeConsumerProperties.java | 57 ---- .../geode/GeodeConsumerApplicationTests.java | 118 ------- .../GeodeServerTestConfiguration.java | 45 --- .../src/test/resources/logback-include.xml | 16 - consumer/pom.xml | 1 - .../redis/RedisConsumerTopicTests.java | 5 +- .../sftp/SftpConsumerConfiguration.java | 9 +- .../consumer/sftp/SftpConsumerProperties.java | 5 +- ...tpConsumerSessionFactoryConfiguration.java | 7 +- function/aggregator-function/pom.xml | 9 - .../AggregatorFunctionConfiguration.java | 29 +- ...ConfigurationEnvironmentPostProcessor.java | 13 +- .../aggregator/MessageStoreConfiguration.java | 33 +- .../AbstractAggregatorFunctionTests.java | 6 +- .../fn/aggregator/DefaultAggregatorTests.java | 3 +- ...AndGemfireMessageStoreAggregatorTests.java | 69 ---- .../JdbcMessageStoreAggregatorTests.java | 4 +- supplier/geode-supplier/README.adoc | 68 ---- supplier/geode-supplier/pom.xml | 44 --- .../geode/GeodeSupplierConfiguration.java | 194 ----------- .../geode/GeodeSupplierProperties.java | 61 ---- .../main/resources/geode-client.properties | 4 - .../geode/GeodeSupplierApplicationTests.java | 176 ---------- .../GeodeServerTestConfiguration.java | 47 --- .../src/test/resources/logback-include.xml | 16 - supplier/pom.xml | 1 - .../sftp/SftpSupplierConfiguration.java | 29 +- .../SftpSupplierFactoryConfiguration.java | 17 +- .../supplier/sftp/SftpSupplierProperties.java | 15 +- 52 files changed, 74 insertions(+), 2785 deletions(-) delete mode 100644 common/function-test-support/src/main/java/org/springframework/cloud/fn/test/support/geode/GeodeContainer.java delete mode 100644 common/function-test-support/src/main/java/org/springframework/cloud/fn/test/support/geode/GeodeContainerIntializer.java delete mode 100644 common/function-test-support/src/main/java/org/springframework/cloud/fn/test/support/geode/SocketUtils.java delete mode 100644 common/geode-common/pom.xml delete mode 100644 common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeClientCacheConfiguration.java delete mode 100644 common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeClientCacheProperties.java delete mode 100644 common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeClientRegionConfiguration.java delete mode 100644 common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodePoolProperties.java delete mode 100644 common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeRegionProperties.java delete mode 100644 common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeSecurityProperties.java delete mode 100644 common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeSslProperties.java delete mode 100644 common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/InetSocketAddressConverterConfiguration.java delete mode 100644 common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/JsonPdxFunctions.java delete mode 100644 common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/ClientCacheAutoConfiguration.java delete mode 100644 consumer/geode-consumer/README.adoc delete mode 100644 consumer/geode-consumer/pom.xml delete mode 100644 consumer/geode-consumer/src/main/java/org/springframework/cloud/fn/consumer/geode/GeodeConsumerConfiguration.java delete mode 100644 consumer/geode-consumer/src/main/java/org/springframework/cloud/fn/consumer/geode/GeodeConsumerHandler.java delete mode 100644 consumer/geode-consumer/src/main/java/org/springframework/cloud/fn/consumer/geode/GeodeConsumerProperties.java delete mode 100644 consumer/geode-consumer/src/test/java/org/springframework/cloud/fn/consumer/geode/GeodeConsumerApplicationTests.java delete mode 100644 consumer/geode-consumer/src/test/java/org/springframework/cloud/fn/consumer/geodeserver/GeodeServerTestConfiguration.java delete mode 100644 consumer/geode-consumer/src/test/resources/logback-include.xml delete mode 100644 function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/GroupTimeOutAndGemfireMessageStoreAggregatorTests.java delete mode 100644 supplier/geode-supplier/README.adoc delete mode 100644 supplier/geode-supplier/pom.xml delete mode 100644 supplier/geode-supplier/src/main/java/org/springframework/cloud/fn/supplier/geode/GeodeSupplierConfiguration.java delete mode 100644 supplier/geode-supplier/src/main/java/org/springframework/cloud/fn/supplier/geode/GeodeSupplierProperties.java delete mode 100644 supplier/geode-supplier/src/main/resources/geode-client.properties delete mode 100644 supplier/geode-supplier/src/test/java/org/springframework/cloud/fn/supplier/geode/GeodeSupplierApplicationTests.java delete mode 100644 supplier/geode-supplier/src/test/java/org/springframework/cloud/fn/supplier/geodeserver/GeodeServerTestConfiguration.java delete mode 100644 supplier/geode-supplier/src/test/resources/logback-include.xml 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 deleted file mode 100644 index b00ff406..00000000 --- a/common/function-test-support/src/main/java/org/springframework/cloud/fn/test/support/geode/GeodeContainer.java +++ /dev/null @@ -1,197 +0,0 @@ -/* - * 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; - - private final boolean useLocator; - - /** - * 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. - * @param useLocator set to use a locator. - */ - public GeodeContainer(@NonNull String dockerImageName, int locatorPort, int cacheServerPort, boolean useLocator) { - super(dockerImageName); - this.locatorPort = locatorPort; - this.cacheServerPort = cacheServerPort; - this.useLocator = useLocator; - } - - public GeodeContainer(@NonNull String dockerImageName, int locatorPort, int cacheServerPort) { - this(dockerImageName, locatorPort, cacheServerPort, false); - } - - /** - * 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. - * @param useLocator set to use a locator. - */ - public GeodeContainer(@NonNull Future image, int locatorPort, int cacheServerPort, boolean useLocator) { - super(image); - this.locatorPort = locatorPort; - this.cacheServerPort = cacheServerPort; - this.useLocator = useLocator; - } - - public GeodeContainer(@NonNull Future image, int locatorPort, int cacheServerPort) { - this(image, locatorPort, cacheServerPort, false); - } - - /** - * A convenience method to connect to a locator with Gfsh. - * @return the connect command String. - */ - public String connect() { - return useLocator ? "connect --locator=" + locators() : "connect --jmx-manager=localhost[1099]"; - } - - /** - * 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 deleted file mode 100644 index 6186c119..00000000 --- a/common/function-test-support/src/main/java/org/springframework/cloud/fn/test/support/geode/GeodeContainerIntializer.java +++ /dev/null @@ -1,112 +0,0 @@ -/* - * 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.HostConfig; -import com.github.dockerjava.api.model.PortBinding; -import com.github.dockerjava.api.model.Ports; -import org.testcontainers.images.builder.ImageFromDockerfile; - -/** - * 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; - - private final boolean useLocator; - - /** - * 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) { - this(postProcessor, false); - } - - public GeodeContainerIntializer(Consumer postProcessor, boolean useLocator) { - this.useLocator = useLocator; - - 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, useLocator); - startContainer(); - } - - /** - * Create and start a {@link GeodeContainer}. - */ - public GeodeContainerIntializer() { - this(null, false); - } - - 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.withHostConfig(new HostConfig().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(); - - if (useLocator) { - geode.execGfsh("start locator --name=Locator1 --hostname-for-clients=localhost --port=" + locatorPort); - geode.execGfsh(geode.connect(), - "start server --name=Server1 --hostname-for-clients=localhost --server-port=" + cacheServerPort); - } - else { - geode.execGfsh( - "start server --name=Server1 --hostname-for-clients=localhost --server-port=" + cacheServerPort + - " --J=-Dgemfire.jmx-manager=true --J=-Dgemfire.jmx-manager-start=true"); - } - 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/java/org/springframework/cloud/fn/test/support/geode/SocketUtils.java b/common/function-test-support/src/main/java/org/springframework/cloud/fn/test/support/geode/SocketUtils.java deleted file mode 100644 index 7c83b76d..00000000 --- a/common/function-test-support/src/main/java/org/springframework/cloud/fn/test/support/geode/SocketUtils.java +++ /dev/null @@ -1,302 +0,0 @@ -/* - * Copyright 2002-2022 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.net.DatagramSocket; -import java.net.InetAddress; -import java.net.ServerSocket; -import java.util.Random; -import java.util.SortedSet; -import java.util.TreeSet; - -import javax.net.ServerSocketFactory; - -import org.springframework.util.Assert; - -/** - * Simple utility methods for working with network sockets — for example, - * for finding available ports on {@code localhost}. - * - *

Within this class, a TCP port refers to a port for a {@link ServerSocket}; - * whereas, a UDP port refers to a port for a {@link DatagramSocket}. - * - *

{@code SocketUtils} was introduced in Spring Framework 4.0, primarily to - * assist in writing integration tests which start an external server on an - * available random port. However, these utilities make no guarantee about the - * subsequent availability of a given port and are therefore unreliable. Instead - * of using {@code SocketUtils} to find an available local port for a server, it - * is recommended that you rely on a server's ability to start on a random port - * that it selects or is assigned by the operating system. To interact with that - * server, you should query the server for the port it is currently using. - * - * @author Sam Brannen - * @author Ben Hale - * @author Arjen Poutsma - * @author Gunnar Hillert - * @author Gary Russell - * @since 4.0 - * @deprecated as of Spring Framework 5.3.16, to be removed in 6.0; see - * {@link SocketUtils class-level Javadoc} for details. - */ -@Deprecated -public final class SocketUtils { - - /** - * The default minimum value for port ranges used when finding an available - * socket port. - */ - public static final int PORT_RANGE_MIN = 1024; - - /** - * The default maximum value for port ranges used when finding an available - * socket port. - */ - public static final int PORT_RANGE_MAX = 65535; - - private static final Random random = new Random(System.nanoTime()); - - private SocketUtils() { - } - - /** - * Find an available TCP port randomly selected from the range - * [{@value #PORT_RANGE_MIN}, {@value #PORT_RANGE_MAX}]. - * @return an available TCP port number - * @throws IllegalStateException if no available port could be found - */ - public static int findAvailableTcpPort() { - return findAvailableTcpPort(PORT_RANGE_MIN); - } - - /** - * Find an available TCP port randomly selected from the range - * [{@code minPort}, {@value #PORT_RANGE_MAX}]. - * @param minPort the minimum port number - * @return an available TCP port number - * @throws IllegalStateException if no available port could be found - */ - public static int findAvailableTcpPort(int minPort) { - return findAvailableTcpPort(minPort, PORT_RANGE_MAX); - } - - /** - * Find an available TCP port randomly selected from the range - * [{@code minPort}, {@code maxPort}]. - * @param minPort the minimum port number - * @param maxPort the maximum port number - * @return an available TCP port number - * @throws IllegalStateException if no available port could be found - */ - public static int findAvailableTcpPort(int minPort, int maxPort) { - return SocketType.TCP.findAvailablePort(minPort, maxPort); - } - - /** - * Find the requested number of available TCP ports, each randomly selected - * from the range [{@value #PORT_RANGE_MIN}, {@value #PORT_RANGE_MAX}]. - * @param numRequested the number of available ports to find - * @return a sorted set of available TCP port numbers - * @throws IllegalStateException if the requested number of available ports could not be found - */ - public static SortedSet findAvailableTcpPorts(int numRequested) { - return findAvailableTcpPorts(numRequested, PORT_RANGE_MIN, PORT_RANGE_MAX); - } - - /** - * Find the requested number of available TCP ports, each randomly selected - * from the range [{@code minPort}, {@code maxPort}]. - * @param numRequested the number of available ports to find - * @param minPort the minimum port number - * @param maxPort the maximum port number - * @return a sorted set of available TCP port numbers - * @throws IllegalStateException if the requested number of available ports could not be found - */ - public static SortedSet findAvailableTcpPorts(int numRequested, int minPort, int maxPort) { - return SocketType.TCP.findAvailablePorts(numRequested, minPort, maxPort); - } - - /** - * Find an available UDP port randomly selected from the range - * [{@value #PORT_RANGE_MIN}, {@value #PORT_RANGE_MAX}]. - * @return an available UDP port number - * @throws IllegalStateException if no available port could be found - */ - public static int findAvailableUdpPort() { - return findAvailableUdpPort(PORT_RANGE_MIN); - } - - /** - * Find an available UDP port randomly selected from the range - * [{@code minPort}, {@value #PORT_RANGE_MAX}]. - * @param minPort the minimum port number - * @return an available UDP port number - * @throws IllegalStateException if no available port could be found - */ - public static int findAvailableUdpPort(int minPort) { - return findAvailableUdpPort(minPort, PORT_RANGE_MAX); - } - - /** - * Find an available UDP port randomly selected from the range - * [{@code minPort}, {@code maxPort}]. - * @param minPort the minimum port number - * @param maxPort the maximum port number - * @return an available UDP port number - * @throws IllegalStateException if no available port could be found - */ - public static int findAvailableUdpPort(int minPort, int maxPort) { - return SocketType.UDP.findAvailablePort(minPort, maxPort); - } - - /** - * Find the requested number of available UDP ports, each randomly selected - * from the range [{@value #PORT_RANGE_MIN}, {@value #PORT_RANGE_MAX}]. - * @param numRequested the number of available ports to find - * @return a sorted set of available UDP port numbers - * @throws IllegalStateException if the requested number of available ports could not be found - */ - public static SortedSet findAvailableUdpPorts(int numRequested) { - return findAvailableUdpPorts(numRequested, PORT_RANGE_MIN, PORT_RANGE_MAX); - } - - /** - * Find the requested number of available UDP ports, each randomly selected - * from the range [{@code minPort}, {@code maxPort}]. - * @param numRequested the number of available ports to find - * @param minPort the minimum port number - * @param maxPort the maximum port number - * @return a sorted set of available UDP port numbers - * @throws IllegalStateException if the requested number of available ports could not be found - */ - public static SortedSet findAvailableUdpPorts(int numRequested, int minPort, int maxPort) { - return SocketType.UDP.findAvailablePorts(numRequested, minPort, maxPort); - } - - - private enum SocketType { - - TCP { - @Override - protected boolean isPortAvailable(int port) { - try { - ServerSocket serverSocket = ServerSocketFactory.getDefault().createServerSocket( - port, 1, InetAddress.getByName("localhost")); - serverSocket.close(); - return true; - } - catch (Exception ex) { - return false; - } - } - }, - - UDP { - @Override - protected boolean isPortAvailable(int port) { - try { - DatagramSocket socket = new DatagramSocket(port, InetAddress.getByName("localhost")); - socket.close(); - return true; - } - catch (Exception ex) { - return false; - } - } - }; - - /** - * Determine if the specified port for this {@code SocketType} is - * currently available on {@code localhost}. - */ - protected abstract boolean isPortAvailable(int port); - - /** - * Find a pseudo-random port number within the range - * [{@code minPort}, {@code maxPort}]. - * @param minPort the minimum port number - * @param maxPort the maximum port number - * @return a random port number within the specified range - */ - private int findRandomPort(int minPort, int maxPort) { - int portRange = maxPort - minPort; - return minPort + random.nextInt(portRange + 1); - } - - /** - * Find an available port for this {@code SocketType}, randomly selected - * from the range [{@code minPort}, {@code maxPort}]. - * @param minPort the minimum port number - * @param maxPort the maximum port number - * @return an available port number for this socket type - * @throws IllegalStateException if no available port could be found - */ - int findAvailablePort(int minPort, int maxPort) { - Assert.isTrue(minPort > 0, "'minPort' must be greater than 0"); - Assert.isTrue(maxPort >= minPort, "'maxPort' must be greater than or equal to 'minPort'"); - Assert.isTrue(maxPort <= PORT_RANGE_MAX, "'maxPort' must be less than or equal to " + PORT_RANGE_MAX); - - int portRange = maxPort - minPort; - int candidatePort; - int searchCounter = 0; - do { - if (searchCounter > portRange) { - throw new IllegalStateException(String.format( - "Could not find an available %s port in the range [%d, %d] after %d attempts", - name(), minPort, maxPort, searchCounter)); - } - candidatePort = findRandomPort(minPort, maxPort); - searchCounter++; - } - while (!isPortAvailable(candidatePort)); - - return candidatePort; - } - - /** - * Find the requested number of available ports for this {@code SocketType}, - * each randomly selected from the range [{@code minPort}, {@code maxPort}]. - * @param numRequested the number of available ports to find - * @param minPort the minimum port number - * @param maxPort the maximum port number - * @return a sorted set of available port numbers for this socket type - * @throws IllegalStateException if the requested number of available ports could not be found - */ - SortedSet findAvailablePorts(int numRequested, int minPort, int maxPort) { - Assert.isTrue(minPort > 0, "'minPort' must be greater than 0"); - Assert.isTrue(maxPort > minPort, "'maxPort' must be greater than 'minPort'"); - Assert.isTrue(maxPort <= PORT_RANGE_MAX, "'maxPort' must be less than or equal to " + PORT_RANGE_MAX); - Assert.isTrue(numRequested > 0, "'numRequested' must be greater than 0"); - Assert.isTrue((maxPort - minPort) >= numRequested, - "'numRequested' must not be greater than 'maxPort' - 'minPort'"); - - SortedSet availablePorts = new TreeSet<>(); - int attemptCount = 0; - while ((++attemptCount <= numRequested + 100) && availablePorts.size() < numRequested) { - availablePorts.add(findAvailablePort(minPort, maxPort)); - } - - if (availablePorts.size() != numRequested) { - throw new IllegalStateException(String.format( - "Could not find %d available %s ports in the range [%d, %d]", - numRequested, name(), minPort, maxPort)); - } - - return availablePorts; - } - } - -} diff --git a/common/geode-common/pom.xml b/common/geode-common/pom.xml deleted file mode 100644 index ea83d16a..00000000 --- a/common/geode-common/pom.xml +++ /dev/null @@ -1,32 +0,0 @@ - - - 4.0.0 - - - org.springframework.cloud.fn - spring-functions-parent - 4.0.0-SNAPSHOT - ../../spring-functions-parent/pom.xml - - - geode-common - geode-common - Geode Common Components - - - 20200518 - - - - - org.springframework.data - spring-data-geode - - - org.json - json - ${org-json.version} - - - - 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 deleted file mode 100644 index b64035c9..00000000 --- a/common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeClientCacheConfiguration.java +++ /dev/null @@ -1,147 +0,0 @@ -/* - * 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 deleted file mode 100644 index 9a36c63b..00000000 --- a/common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeClientCacheProperties.java +++ /dev/null @@ -1,43 +0,0 @@ -/* - * 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 deleted file mode 100644 index 879f24ac..00000000 --- a/common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeClientRegionConfiguration.java +++ /dev/null @@ -1,68 +0,0 @@ -/* - * 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 deleted file mode 100644 index a1b2be03..00000000 --- a/common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodePoolProperties.java +++ /dev/null @@ -1,90 +0,0 @@ -/* - * 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 jakarta.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 deleted file mode 100644 index 727891dc..00000000 --- a/common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeRegionProperties.java +++ /dev/null @@ -1,45 +0,0 @@ -/* - * 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 jakarta.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 deleted file mode 100644 index 07c9c72c..00000000 --- a/common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeSecurityProperties.java +++ /dev/null @@ -1,100 +0,0 @@ -/* - * 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 deleted file mode 100644 index 2544ff04..00000000 --- a/common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeSslProperties.java +++ /dev/null @@ -1,170 +0,0 @@ -/* - * 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 jakarta.validation.constraints.AssertTrue; -import jakarta.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 deleted file mode 100644 index 407781af..00000000 --- a/common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/InetSocketAddressConverterConfiguration.java +++ /dev/null @@ -1,49 +0,0 @@ -/* - * 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 deleted file mode 100644 index f637824c..00000000 --- a/common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/JsonPdxFunctions.java +++ /dev/null @@ -1,49 +0,0 @@ -/* - * 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/common/metadata-store-common/pom.xml b/common/metadata-store-common/pom.xml index bc7a7f97..1345d692 100644 --- a/common/metadata-store-common/pom.xml +++ b/common/metadata-store-common/pom.xml @@ -51,7 +51,6 @@ test - org.springframework.boot spring-boot-starter-logging @@ -63,22 +62,6 @@ - - org.springframework.integration - spring-integration-gemfire - true - - - org.springframework.data - spring-data-gemfire - - - - - org.springframework.data - spring-data-geode - true - diff --git a/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/ClientCacheAutoConfiguration.java b/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/ClientCacheAutoConfiguration.java deleted file mode 100644 index 22aed185..00000000 --- a/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/ClientCacheAutoConfiguration.java +++ /dev/null @@ -1,44 +0,0 @@ -/* - * 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.metadata.store; - -import org.apache.geode.cache.GemFireCache; -import org.apache.geode.cache.client.ClientCache; - -import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; -import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; -import org.springframework.context.annotation.Configuration; -import org.springframework.data.gemfire.client.ClientCacheFactoryBean; -import org.springframework.data.gemfire.config.annotation.ClientCacheApplication; -import org.springframework.data.gemfire.config.annotation.EnablePdx; - -/** - * TODO - *

- * This is the copy of {@code org.springframework.boot.data.geode.autoconfigure.ClientCacheAutoConfiguration} - * until {@code geode-spring-boot-starter} is released. - * - * @author John Blum - */ -@Configuration -@ConditionalOnClass({ClientCacheFactoryBean.class, ClientCache.class}) -@ConditionalOnMissingBean(GemFireCache.class) -@ClientCacheApplication -@EnablePdx -public class ClientCacheAutoConfiguration { - -} diff --git a/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreAutoConfiguration.java b/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreAutoConfiguration.java index f6ab135f..e92489f3 100644 --- a/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreAutoConfiguration.java +++ b/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreAutoConfiguration.java @@ -24,8 +24,6 @@ import io.awspring.cloud.core.region.RegionProvider; import org.apache.curator.framework.CuratorFramework; import org.apache.curator.framework.CuratorFrameworkFactory; import org.apache.curator.retry.RetryForever; -import org.apache.geode.cache.GemFireCache; -import org.apache.geode.cache.Region; import org.springframework.beans.factory.ObjectProvider; import org.springframework.boot.autoconfigure.AutoConfiguration; @@ -34,12 +32,9 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Import; -import org.springframework.data.gemfire.client.ClientRegionFactoryBean; import org.springframework.data.mongodb.core.MongoTemplate; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.integration.aws.metadata.DynamoDbMetadataStore; -import org.springframework.integration.gemfire.metadata.GemfireMetadataStore; import org.springframework.integration.hazelcast.metadata.HazelcastMetadataStore; import org.springframework.integration.jdbc.metadata.JdbcMetadataStore; import org.springframework.integration.metadata.ConcurrentMetadataStore; @@ -53,6 +48,7 @@ import org.springframework.jdbc.core.JdbcTemplate; /** * @author Artem Bilan * @author David Turanski + * @author Corneil du Plessis * @since 2.0.2 */ @AutoConfiguration @@ -93,35 +89,6 @@ public class MetadataStoreAutoConfiguration { } - @ConditionalOnProperty(prefix = "metadata.store", name = "type", havingValue = "gemfire") - @Import(ClientCacheAutoConfiguration.class) - static class Gemfire { - - @Bean - @ConditionalOnMissingBean - public ClientRegionFactoryBean gemfireRegion(GemFireCache cache, - MetadataStoreProperties metadataStoreProperties) { - - ClientRegionFactoryBean clientRegionFactoryBean = new ClientRegionFactoryBean<>(); - clientRegionFactoryBean.setCache(cache); - clientRegionFactoryBean.setName(metadataStoreProperties.getGemfire().getRegion()); - return clientRegionFactoryBean; - } - - @Bean - @ConditionalOnMissingBean - public ConcurrentMetadataStore gemfireMetadataStore(Region region, - ObjectProvider metadataStoreListenerObjectProvider) { - - @SuppressWarnings("unchecked") - GemfireMetadataStore gemfireMetadataStore = new GemfireMetadataStore((Region) region); - metadataStoreListenerObjectProvider.ifAvailable(gemfireMetadataStore::addListener); - - return gemfireMetadataStore; - } - - } - @ConditionalOnProperty(prefix = "metadata.store", name = "type", havingValue = "hazelcast") static class Hazelcast { diff --git a/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreProperties.java b/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreProperties.java index ac030cd7..85b0c4d9 100644 --- a/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreProperties.java +++ b/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreProperties.java @@ -21,20 +21,19 @@ import java.nio.charset.StandardCharsets; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.integration.aws.metadata.DynamoDbMetadataStore; -import org.springframework.integration.gemfire.metadata.GemfireMetadataStore; import org.springframework.integration.jdbc.metadata.JdbcMetadataStore; import org.springframework.integration.redis.metadata.RedisMetadataStore; /** * @author Artem Bilan * @author David Turanski + * @author Corneil du Plessis * @since 2.0.2 */ @ConfigurationProperties("metadata.store") public class MetadataStoreProperties { enum StoreType { mongodb, - gemfire, redis, dynamodb, jdbc, @@ -51,8 +50,6 @@ public class MetadataStoreProperties { private final Mongo mongoDb = new Mongo(); - private final Gemfire gemfire = new Gemfire(); - private final Redis redis = new Redis(); private final DynamoDb dynamoDb = new DynamoDb(); @@ -73,10 +70,6 @@ public class MetadataStoreProperties { return this.mongoDb; } - public Gemfire getGemfire() { - return this.gemfire; - } - public Redis getRedis() { return this.redis; } @@ -110,23 +103,6 @@ public class MetadataStoreProperties { } - public static class Gemfire { - - /** - * Gemfire region name for metadata. - */ - private String region = GemfireMetadataStore.KEY; - - public String getRegion() { - return this.region; - } - - public void setRegion(String region) { - this.region = region; - } - - } - public static class Redis { /** diff --git a/common/metadata-store-common/src/test/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreAutoConfigurationTests.java b/common/metadata-store-common/src/test/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreAutoConfigurationTests.java index c92e12e9..7c08cc66 100644 --- a/common/metadata-store-common/src/test/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreAutoConfigurationTests.java +++ b/common/metadata-store-common/src/test/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreAutoConfigurationTests.java @@ -40,7 +40,6 @@ import org.springframework.boot.test.context.runner.ApplicationContextRunner; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.integration.aws.metadata.DynamoDbMetadataStore; -import org.springframework.integration.gemfire.metadata.GemfireMetadataStore; import org.springframework.integration.hazelcast.metadata.HazelcastMetadataStore; import org.springframework.integration.jdbc.metadata.JdbcMetadataStore; import org.springframework.integration.metadata.ConcurrentMetadataStore; @@ -57,6 +56,7 @@ import static org.mockito.Mockito.mock; /** * @author Artem Bilan + * @author Corneil du Plessis * @since 2.0.2 */ @RunWith(Parameterized.class) @@ -67,7 +67,6 @@ public class MetadataStoreAutoConfigurationTests { Arrays.asList( RedisMetadataStore.class, MongoDbMetadataStore.class, - GemfireMetadataStore.class, JdbcMetadataStore.class, ZookeeperMetadataStore.class, HazelcastMetadataStore.class, diff --git a/common/pom.xml b/common/pom.xml index 80c5e784..c3f4a275 100644 --- a/common/pom.xml +++ b/common/pom.xml @@ -16,7 +16,6 @@ file-common ftp-common function-test-support - geode-common metadata-store-common mqtt-common redis-common diff --git a/consumer/geode-consumer/README.adoc b/consumer/geode-consumer/README.adoc deleted file mode 100644 index 3cfce0bb..00000000 --- a/consumer/geode-consumer/README.adoc +++ /dev/null @@ -1,48 +0,0 @@ -# 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 deleted file mode 100644 index 1f14e5ce..00000000 --- a/consumer/geode-consumer/pom.xml +++ /dev/null @@ -1,44 +0,0 @@ - - - 4.0.0 - - - org.springframework.cloud.fn - spring-functions-parent - 4.0.0-SNAPSHOT - ../../spring-functions-parent/pom.xml - - - geode-consumer - geode-consumer - Geode consumer - - - - org.springframework.integration - spring-integration-gemfire - - - org.springframework.cloud.fn - geode-common - ${project.version} - - - org.springframework.geode - spring-geode-starter-test - test - - - org.springframework.geode - spring-geode-starter-logging - 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 deleted file mode 100644 index 474c0c7a..00000000 --- a/consumer/geode-consumer/src/main/java/org/springframework/cloud/fn/consumer/geode/GeodeConsumerConfiguration.java +++ /dev/null @@ -1,68 +0,0 @@ -/* - * Copyright 2020-2022 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.config.ComponentCustomizer; -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.lang.Nullable; -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(proxyBeanMethods = false) -@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, - @Nullable ComponentCustomizer cacheWritingMessageHandlerCustomizer) { - - CacheWritingMessageHandler messageHandler = new CacheWritingMessageHandler(region); - messageHandler.setCacheEntries(Collections.singletonMap(properties.getKeyExpression(), "payload")); - if (cacheWritingMessageHandlerCustomizer != null) { - cacheWritingMessageHandlerCustomizer.customize(messageHandler); - } - 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 deleted file mode 100644 index d7c5e660..00000000 --- a/consumer/geode-consumer/src/main/java/org/springframework/cloud/fn/consumer/geode/GeodeConsumerHandler.java +++ /dev/null @@ -1,68 +0,0 @@ -/* - * 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 deleted file mode 100644 index 555e80a6..00000000 --- a/consumer/geode-consumer/src/main/java/org/springframework/cloud/fn/consumer/geode/GeodeConsumerProperties.java +++ /dev/null @@ -1,57 +0,0 @@ -/* - * 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 jakarta.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 deleted file mode 100644 index cc551a07..00000000 --- a/consumer/geode-consumer/src/test/java/org/springframework/cloud/fn/consumer/geode/GeodeConsumerApplicationTests.java +++ /dev/null @@ -1,118 +0,0 @@ -/* - * 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.io.IOException; -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.AfterAll; -import org.junit.jupiter.api.BeforeAll; -import org.junit.jupiter.api.Tag; -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.consumer.geodeserver.GeodeServerTestConfiguration; -import org.springframework.data.gemfire.tests.integration.ForkingClientServerIntegrationTestsSupport; -import org.springframework.messaging.Message; -import org.springframework.messaging.support.GenericMessage; - -import static org.assertj.core.api.Assertions.assertThat; - -@Tag("integration") -public class GeodeConsumerApplicationTests { - private static ApplicationContextRunner applicationContextRunner; - - private ObjectMapper objectMapper = new ObjectMapper(); - - @BeforeAll - static void setup() throws IOException { - ForkingClientServerIntegrationTestsSupport.startGemFireServer( - GeodeServerTestConfiguration.class); - - applicationContextRunner = new ApplicationContextRunner() - .withUserConfiguration(GeodeConsumerTestApplication.class); - } - - @AfterAll - static void stopServer() { - ForkingClientServerIntegrationTestsSupport.stopGemFireServer(); - ForkingClientServerIntegrationTestsSupport.clearCacheServerPortAndPoolPortProperties(); - } - - @Test - void consumeWithJsonEnabled() { - applicationContextRunner - .withPropertyValues( - "geode.region.regionName=Stocks", - "geode.consumer.json=true", - "geode.consumer.key-expression=payload.getField('symbol')", - "geode.pool.connectType=server", - "geode.pool.hostAddresses=" + "localhost:" + System.getProperty("spring.data.gemfire.cache.server.port")) - .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.connectType=server", - "geode.pool.hostAddresses=" + "localhost:" + System.getProperty("spring.data.gemfire.cache.server.port")) - .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/consumer/geode-consumer/src/test/java/org/springframework/cloud/fn/consumer/geodeserver/GeodeServerTestConfiguration.java b/consumer/geode-consumer/src/test/java/org/springframework/cloud/fn/consumer/geodeserver/GeodeServerTestConfiguration.java deleted file mode 100644 index 4ad3a629..00000000 --- a/consumer/geode-consumer/src/test/java/org/springframework/cloud/fn/consumer/geodeserver/GeodeServerTestConfiguration.java +++ /dev/null @@ -1,45 +0,0 @@ -/* - * 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.geodeserver; - -import org.apache.geode.cache.GemFireCache; - -import org.springframework.context.annotation.AnnotationConfigApplicationContext; -import org.springframework.context.annotation.Bean; -import org.springframework.data.gemfire.ReplicatedRegionFactoryBean; -import org.springframework.data.gemfire.config.annotation.CacheServerApplication; - -@CacheServerApplication -public class GeodeServerTestConfiguration { - - public static void main(String[] args) { - - AnnotationConfigApplicationContext applicationContext = - new AnnotationConfigApplicationContext(GeodeServerTestConfiguration.class); - - applicationContext.registerShutdownHook(); - } - - @Bean("Stocks") - public ReplicatedRegionFactoryBean stocksRegion(GemFireCache gemfireCache) { - - ReplicatedRegionFactoryBean regionFactoryBean = new ReplicatedRegionFactoryBean<>(); - regionFactoryBean.setCache(gemfireCache); - - return regionFactoryBean; - } -} diff --git a/consumer/geode-consumer/src/test/resources/logback-include.xml b/consumer/geode-consumer/src/test/resources/logback-include.xml deleted file mode 100644 index ae330ab3..00000000 --- a/consumer/geode-consumer/src/test/resources/logback-include.xml +++ /dev/null @@ -1,16 +0,0 @@ - - - - - - %d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n - - - - - - - - - - diff --git a/consumer/pom.xml b/consumer/pom.xml index a7d7c04c..08bbba41 100644 --- a/consumer/pom.xml +++ b/consumer/pom.xml @@ -17,7 +17,6 @@ analytics-consumer file-consumer ftp-consumer - geode-consumer jdbc-consumer log-consumer mongodb-consumer diff --git a/consumer/redis-consumer/src/test/java/org/springframework/cloud/fn/consumer/redis/RedisConsumerTopicTests.java b/consumer/redis-consumer/src/test/java/org/springframework/cloud/fn/consumer/redis/RedisConsumerTopicTests.java index f559f882..d2af40dc 100644 --- a/consumer/redis-consumer/src/test/java/org/springframework/cloud/fn/consumer/redis/RedisConsumerTopicTests.java +++ b/consumer/redis-consumer/src/test/java/org/springframework/cloud/fn/consumer/redis/RedisConsumerTopicTests.java @@ -23,15 +23,11 @@ import java.util.concurrent.TimeUnit; import org.awaitility.Awaitility; import org.junit.jupiter.api.Test; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.data.redis.connection.RedisConnection; -import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.listener.ChannelTopic; import org.springframework.data.redis.listener.RedisMessageListenerContainer; import org.springframework.data.redis.listener.Topic; import org.springframework.data.redis.listener.adapter.MessageListenerAdapter; import org.springframework.data.redis.serializer.StringRedisSerializer; -import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.support.MessageBuilder; import org.springframework.test.context.TestPropertySource; @@ -40,6 +36,7 @@ import static org.assertj.core.api.Assertions.assertThat; /** * @author Soby Chacko + * @author Corneil du Plessis */ @TestPropertySource(properties = "redis.consumer.topic = foo-topic") public class RedisConsumerTopicTests extends AbstractRedisConsumerTests { diff --git a/consumer/sftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerConfiguration.java b/consumer/sftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerConfiguration.java index 520419d4..4c267b0b 100644 --- a/consumer/sftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerConfiguration.java +++ b/consumer/sftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerConfiguration.java @@ -18,7 +18,7 @@ package org.springframework.cloud.fn.consumer.sftp; import java.util.function.Consumer; -import com.jcraft.jsch.ChannelSftp; +import org.apache.sshd.sftp.client.SftpClient; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.cloud.fn.common.config.ComponentCustomizer; @@ -35,6 +35,11 @@ import org.springframework.integration.sftp.session.SftpRemoteFileTemplate; import org.springframework.lang.Nullable; import org.springframework.messaging.Message; +/** + * Configuration for SFTP Consumer. + * @author Soby Chacko + * @author Corneil du Plessis + */ @Configuration(proxyBeanMethods = false) @EnableConfigurationProperties(SftpConsumerProperties.class) @Import(SftpConsumerSessionFactoryConfiguration.class) @@ -42,7 +47,7 @@ public class SftpConsumerConfiguration { @Bean public IntegrationFlow ftpOutboundFlow(SftpConsumerProperties properties, - SessionFactory ftpSessionFactory, + SessionFactory ftpSessionFactory, @Nullable ComponentCustomizer sftpMessageHandlerSpecCustomizer) { IntegrationFlowBuilder integrationFlowBuilder = diff --git a/consumer/sftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerProperties.java b/consumer/sftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerProperties.java index 34553057..57ecdcd2 100644 --- a/consumer/sftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerProperties.java +++ b/consumer/sftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerProperties.java @@ -29,6 +29,7 @@ import org.springframework.validation.annotation.Validated; /** * @author Gary Russell * @author Artem Bilan + * @author Corneil du Plessis */ @ConfigurationProperties("sftp.consumer") @Validated @@ -42,7 +43,7 @@ public class SftpConsumerProperties { private String temporaryRemoteDir = "/"; /** - * Whether or not to create the remote directory. + * Whether to create the remote directory. */ private boolean autoCreateDir = true; @@ -52,7 +53,7 @@ public class SftpConsumerProperties { private FileExistsMode mode = FileExistsMode.REPLACE; /** - * Whether or not to write to a temporary file and rename. + * Whether to write to a temporary file and rename. */ private boolean useTemporaryFilename = true; diff --git a/consumer/sftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerSessionFactoryConfiguration.java b/consumer/sftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerSessionFactoryConfiguration.java index 9986b635..92e826b6 100644 --- a/consumer/sftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerSessionFactoryConfiguration.java +++ b/consumer/sftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerSessionFactoryConfiguration.java @@ -18,7 +18,7 @@ package org.springframework.cloud.fn.consumer.sftp; import java.nio.charset.StandardCharsets; -import com.jcraft.jsch.ChannelSftp; +import org.apache.sshd.sftp.client.SftpClient; import org.springframework.beans.factory.BeanFactory; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; @@ -33,13 +33,14 @@ import org.springframework.integration.sftp.session.DefaultSftpSessionFactory; * Session factory configuration. * * @author Gary Russell + * @author Corneil du Plessis * */ public class SftpConsumerSessionFactoryConfiguration { @Bean @ConditionalOnMissingBean - public SessionFactory sftpSessionFactory(SftpConsumerProperties properties, BeanFactory beanFactory) { + public SessionFactory sftpSessionFactory(SftpConsumerProperties properties, BeanFactory beanFactory) { DefaultSftpSessionFactory sftpSessionFactory = new DefaultSftpSessionFactory(); SftpConsumerProperties.Factory factory = properties.getFactory(); sftpSessionFactory.setHost(factory.getHost()); @@ -57,7 +58,7 @@ public class SftpConsumerSessionFactoryConfiguration { sftpSessionFactory.setKnownHostsResource(knownHostsResource); } if (factory.getCacheSessions() != null) { - CachingSessionFactory csf = new CachingSessionFactory<>(sftpSessionFactory); + CachingSessionFactory csf = new CachingSessionFactory<>(sftpSessionFactory); return csf; } else { diff --git a/function/aggregator-function/pom.xml b/function/aggregator-function/pom.xml index a2cd72aa..6cdfd88a 100644 --- a/function/aggregator-function/pom.xml +++ b/function/aggregator-function/pom.xml @@ -40,15 +40,6 @@ spring-boot-starter-data-redis runtime - - - org.springframework.integration - spring-integration-gemfire - - - org.springframework.geode - spring-geode-starter - org.springframework.integration diff --git a/function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/AggregatorFunctionConfiguration.java b/function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/AggregatorFunctionConfiguration.java index 6bc87f85..1f28ce76 100644 --- a/function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/AggregatorFunctionConfiguration.java +++ b/function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/AggregatorFunctionConfiguration.java @@ -49,6 +49,7 @@ import org.springframework.messaging.MessageChannel; /** * @author Artem Bilan + * @author Corneil du Plessis */ @AutoConfiguration @EnableConfigurationProperties(AggregatorFunctionProperties.class) @@ -61,11 +62,12 @@ public class AggregatorFunctionConfiguration { private BeanFactory beanFactory; @Bean - public Function>, Flux>> aggregatorFunction(FluxMessageChannel inputChannel, - FluxMessageChannel outputChannel) { - + public Function>, Flux>> aggregatorFunction( + FluxMessageChannel inputChannel, + FluxMessageChannel outputChannel + ) { return input -> Flux.from(outputChannel) - .doOnRequest((request) -> inputChannel.subscribeTo(input)); + .doOnRequest((request) -> inputChannel.subscribeTo(input)); } @Bean @@ -81,12 +83,12 @@ public class AggregatorFunctionConfiguration { @Bean @ServiceActivator(inputChannel = "inputChannel") public AggregatorFactoryBean aggregator( - @Nullable CorrelationStrategy correlationStrategy, - @Nullable ReleaseStrategy releaseStrategy, - @Nullable MessageGroupProcessor messageGroupProcessor, - @Nullable MessageGroupStore messageStore, - @Qualifier("outputChannel") MessageChannel outputChannel, - @Nullable ComponentCustomizer aggregatorCustomizer) { + @Nullable CorrelationStrategy correlationStrategy, + @Nullable ReleaseStrategy releaseStrategy, + @Nullable MessageGroupProcessor messageGroupProcessor, + @Nullable MessageGroupStore messageStore, + @Qualifier("outputChannel") MessageChannel outputChannel, + @Nullable ComponentCustomizer aggregatorCustomizer) { AggregatorFactoryBean aggregator = new AggregatorFactoryBean(); aggregator.setExpireGroupsUponCompletion(true); @@ -144,8 +146,11 @@ public class AggregatorFunctionConfiguration { @Configuration @ConditionalOnMissingBean(MessageGroupStore.class) - @Import({ MessageStoreConfiguration.Mongo.class, MessageStoreConfiguration.Redis.class, - MessageStoreConfiguration.Gemfire.class, MessageStoreConfiguration.Jdbc.class }) + @Import({ + MessageStoreConfiguration.Mongo.class, + MessageStoreConfiguration.Redis.class, + MessageStoreConfiguration.Jdbc.class + }) protected static class MessageStoreAutoConfiguration { } diff --git a/function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/ExcludeStoresAutoConfigurationEnvironmentPostProcessor.java b/function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/ExcludeStoresAutoConfigurationEnvironmentPostProcessor.java index 5743a0d4..60614059 100644 --- a/function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/ExcludeStoresAutoConfigurationEnvironmentPostProcessor.java +++ b/function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/ExcludeStoresAutoConfigurationEnvironmentPostProcessor.java @@ -30,14 +30,13 @@ import org.springframework.boot.env.EnvironmentPostProcessor; import org.springframework.core.env.ConfigurableEnvironment; import org.springframework.core.env.MutablePropertySources; import org.springframework.core.env.PropertiesPropertySource; -import org.springframework.geode.boot.autoconfigure.ClientCacheAutoConfiguration; -import org.springframework.geode.boot.autoconfigure.SslAutoConfiguration; /** * An {@link EnvironmentPostProcessor} to add {@code spring.autoconfigure.exclude} property * since we can't use {@code application.properties} from the library perspective. * * @author Artem Bilan + * @author Corneil du Plessis */ public class ExcludeStoresAutoConfigurationEnvironmentPostProcessor implements EnvironmentPostProcessor { @@ -52,18 +51,10 @@ public class ExcludeStoresAutoConfigurationEnvironmentPostProcessor implements E MongoAutoConfiguration.class.getName() + ", " + MongoDataAutoConfiguration.class.getName() + ", " + MongoRepositoriesAutoConfiguration.class.getName() + ", " + - ClientCacheAutoConfiguration.class.getName() + ", " + RedisAutoConfiguration.class.getName() + ", " + RedisRepositoriesAutoConfiguration.class.getName()); - String messageStoreType = environment.getProperty(AggregatorFunctionProperties.PREFIX + ".message-store-type"); - if (!AggregatorFunctionProperties.MessageStoreType.GEMFIRE.equals(messageStoreType)) { - properties.setProperty(SslAutoConfiguration.SECURITY_SSL_ENVIRONMENT_POST_PROCESSOR_ENABLED_PROPERTY, - "false"); - } - - propertySources.addLast( - new PropertiesPropertySource("aggregator.exclude.stores.auto-configuration", properties)); + propertySources.addLast(new PropertiesPropertySource("aggregator.exclude.stores.auto-configuration", properties)); } } diff --git a/function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/MessageStoreConfiguration.java b/function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/MessageStoreConfiguration.java index 460ea914..c823a8c3 100644 --- a/function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/MessageStoreConfiguration.java +++ b/function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/MessageStoreConfiguration.java @@ -18,11 +18,7 @@ package org.springframework.cloud.fn.aggregator; import java.util.Arrays; -import org.apache.geode.cache.GemFireCache; -import org.apache.geode.cache.Region; - import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; -import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.autoconfigure.data.mongo.MongoDataAutoConfiguration; import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration; @@ -32,13 +28,9 @@ import org.springframework.boot.autoconfigure.mongo.MongoAutoConfiguration; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Import; import org.springframework.context.annotation.Primary; -import org.springframework.data.gemfire.client.ClientRegionFactoryBean; -import org.springframework.data.gemfire.config.annotation.EnablePdx; import org.springframework.data.mongodb.core.MongoTemplate; import org.springframework.data.mongodb.core.convert.MongoCustomConversions; import org.springframework.data.redis.core.RedisTemplate; -import org.springframework.geode.boot.autoconfigure.ClientCacheAutoConfiguration; -import org.springframework.integration.gemfire.store.GemfireMessageStore; import org.springframework.integration.jdbc.store.JdbcMessageStore; import org.springframework.integration.mongodb.store.ConfigurableMongoDbMessageStore; import org.springframework.integration.mongodb.support.BinaryToMessageConverter; @@ -55,6 +47,7 @@ import org.springframework.util.StringUtils; * via matched configuration properties. * * @author Artem Bilan + * @author Corneil du Plessis */ class MessageStoreConfiguration { @@ -99,30 +92,6 @@ class MessageStoreConfiguration { } - @ConditionalOnClass(GemfireMessageStore.class) - @ConditionalOnProperty(prefix = AggregatorFunctionProperties.PREFIX, - name = "message-store-type", - havingValue = AggregatorFunctionProperties.MessageStoreType.GEMFIRE) - @Import(ClientCacheAutoConfiguration.class) - @EnablePdx - static class Gemfire { - - @Bean - @ConditionalOnMissingBean - public ClientRegionFactoryBean gemfireRegion(GemFireCache cache, AggregatorFunctionProperties properties) { - ClientRegionFactoryBean clientRegionFactoryBean = new ClientRegionFactoryBean<>(); - clientRegionFactoryBean.setCache(cache); - clientRegionFactoryBean.setName(properties.getMessageStoreEntity()); - return clientRegionFactoryBean; - } - - @Bean - public MessageGroupStore messageStore(Region region) { - return new GemfireMessageStore(region); - } - - } - @ConditionalOnClass(JdbcMessageStore.class) @ConditionalOnProperty(prefix = AggregatorFunctionProperties.PREFIX, name = "message-store-type", diff --git a/function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/AbstractAggregatorFunctionTests.java b/function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/AbstractAggregatorFunctionTests.java index e4e65721..05e6f8c5 100644 --- a/function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/AbstractAggregatorFunctionTests.java +++ b/function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/AbstractAggregatorFunctionTests.java @@ -21,10 +21,8 @@ import java.util.function.Function; import reactor.core.publisher.Flux; import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.geode.boot.autoconfigure.SslAutoConfiguration; import org.springframework.integration.aggregator.AggregatingMessageHandler; import org.springframework.integration.store.MessageGroupStore; import org.springframework.messaging.Message; @@ -32,6 +30,7 @@ import org.springframework.test.annotation.DirtiesContext; /** * @author Artem Bilan + * @author Corneil du Plessis */ @SpringBootTest @DirtiesContext @@ -46,9 +45,6 @@ public abstract class AbstractAggregatorFunctionTests { @Autowired protected AggregatingMessageHandler aggregatingMessageHandler; - @Value("${" + SslAutoConfiguration.SECURITY_SSL_ENVIRONMENT_POST_PROCESSOR_ENABLED_PROPERTY + ":true}") - protected Boolean geodeSslEnable; - @SpringBootApplication static class AggregatorFunctionTestApplication { diff --git a/function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/DefaultAggregatorTests.java b/function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/DefaultAggregatorTests.java index 7af1cdad..32a8e997 100644 --- a/function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/DefaultAggregatorTests.java +++ b/function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/DefaultAggregatorTests.java @@ -32,6 +32,7 @@ import static org.assertj.core.api.Assertions.assertThat; /** * @author Artem Bilan + * @author Corneil du Plessis */ public class DefaultAggregatorTests extends AbstractAggregatorFunctionTests { @@ -65,8 +66,6 @@ public class DefaultAggregatorTests extends AbstractAggregatorFunctionTests { assertThat(this.messageGroupStore).isNull(); assertThat(this.aggregatingMessageHandler.getMessageStore()).isInstanceOf(SimpleMessageStore.class); - // Also verify geode ssl flag not enabled for default message store (non-geode) - assertThat(this.geodeSslEnable).isFalse(); } } diff --git a/function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/GroupTimeOutAndGemfireMessageStoreAggregatorTests.java b/function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/GroupTimeOutAndGemfireMessageStoreAggregatorTests.java deleted file mode 100644 index 850a9ded..00000000 --- a/function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/GroupTimeOutAndGemfireMessageStoreAggregatorTests.java +++ /dev/null @@ -1,69 +0,0 @@ -/* - * Copyright 2020-2022 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.aggregator; - -import java.time.Duration; -import java.util.List; - -import org.junit.jupiter.api.Test; -import reactor.core.publisher.Flux; -import reactor.test.StepVerifier; - -import org.springframework.integration.IntegrationMessageHeaderAccessor; -import org.springframework.integration.gemfire.store.GemfireMessageStore; -import org.springframework.integration.support.MessageBuilder; -import org.springframework.messaging.Message; -import org.springframework.test.context.TestPropertySource; - -import static org.assertj.core.api.Assertions.assertThat; - -/** - * @author Artem Bilan - */ -@TestPropertySource(properties = { - "aggregator.message-store-type=gemfire", - "aggregator.groupTimeout=10" }) -public class GroupTimeOutAndGemfireMessageStoreAggregatorTests extends AbstractAggregatorFunctionTests { - - @Test - public void test() { - Flux> input = - Flux.just(MessageBuilder.withPayload("1") - .setHeader(IntegrationMessageHeaderAccessor.CORRELATION_ID, "my_correlation") - .build()); - - Flux> output = this.aggregatorFunction.apply(input); - - output.as(StepVerifier::create) - .assertNext((message) -> - assertThat(message) - .extracting(Message::getPayload) - .isInstanceOf(List.class) - .asList() - .hasSize(1) - .contains("1")) - .thenCancel() - .verify(Duration.ofSeconds(10)); - - assertThat(this.messageGroupStore).isInstanceOf(GemfireMessageStore.class); - assertThat(this.aggregatingMessageHandler.getMessageStore()).isSameAs(this.messageGroupStore); - - // Also verify geode ssl flag enabled for geode message stores - assertThat(this.geodeSslEnable).isTrue(); - } - -} diff --git a/function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/JdbcMessageStoreAggregatorTests.java b/function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/JdbcMessageStoreAggregatorTests.java index ac87a5c5..d009270f 100644 --- a/function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/JdbcMessageStoreAggregatorTests.java +++ b/function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/JdbcMessageStoreAggregatorTests.java @@ -33,6 +33,7 @@ import static org.assertj.core.api.Assertions.assertThat; /** * @author Artem Bilan + * @author Corneil du Plessis */ @TestPropertySource(properties = "aggregator.message-store-type=jdbc") public class JdbcMessageStoreAggregatorTests extends AbstractAggregatorFunctionTests { @@ -67,9 +68,6 @@ public class JdbcMessageStoreAggregatorTests extends AbstractAggregatorFunctionT assertThat(this.messageGroupStore).isInstanceOf(JdbcMessageStore.class); assertThat(this.aggregatingMessageHandler.getMessageStore()).isSameAs(this.messageGroupStore); - - // Also verify geode ssl flag not enabled for non-geode message stores - assertThat(this.geodeSslEnable).isFalse(); } } diff --git a/supplier/geode-supplier/README.adoc b/supplier/geode-supplier/README.adoc deleted file mode 100644 index b1a5208b..00000000 --- a/supplier/geode-supplier/README.adoc +++ /dev/null @@ -1,68 +0,0 @@ -# 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 deleted file mode 100644 index 2c91f6dd..00000000 --- a/supplier/geode-supplier/pom.xml +++ /dev/null @@ -1,44 +0,0 @@ - - - 4.0.0 - - - org.springframework.cloud.fn - spring-functions-parent - 4.0.0-SNAPSHOT - ../../spring-functions-parent/pom.xml - - - geode-supplier - geode-supplier - geode supplier - - - - org.springframework.integration - spring-integration-gemfire - - - org.springframework.cloud.fn - geode-common - ${project.version} - - - org.springframework.geode - spring-geode-starter-logging - test - - - org.springframework.geode - spring-geode-starter-test - test - - - org.projectlombok - lombok - 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 deleted file mode 100644 index af27f90f..00000000 --- a/supplier/geode-supplier/src/main/java/org/springframework/cloud/fn/supplier/geode/GeodeSupplierConfiguration.java +++ /dev/null @@ -1,194 +0,0 @@ -/* - * 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 deleted file mode 100644 index 9e08f86e..00000000 --- a/supplier/geode-supplier/src/main/java/org/springframework/cloud/fn/supplier/geode/GeodeSupplierProperties.java +++ /dev/null @@ -1,61 +0,0 @@ -/* - * 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 deleted file mode 100644 index d4de889e..00000000 --- a/supplier/geode-supplier/src/main/resources/geode-client.properties +++ /dev/null @@ -1,4 +0,0 @@ -# -# 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 deleted file mode 100644 index e07e0816..00000000 --- a/supplier/geode-supplier/src/test/java/org/springframework/cloud/fn/supplier/geode/GeodeSupplierApplicationTests.java +++ /dev/null @@ -1,176 +0,0 @@ -/* - * 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.io.IOException; -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.AfterAll; -import org.junit.jupiter.api.BeforeAll; -import org.junit.jupiter.api.Tag; -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.supplier.geodeserver.GeodeServerTestConfiguration; -import org.springframework.data.gemfire.tests.integration.ForkingClientServerIntegrationTestsSupport; - -import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.fail; - -@Tag("integration") -public class GeodeSupplierApplicationTests { - - private static ApplicationContextRunner applicationContextRunner; - - private final ObjectMapper objectMapper = new ObjectMapper(); - - @BeforeAll - static void setup() throws IOException { - ForkingClientServerIntegrationTestsSupport.startGemFireServer( - GeodeServerTestConfiguration.class); - - applicationContextRunner = new ApplicationContextRunner() - .withUserConfiguration(GeodeSupplierTestApplication.class); - } - - @AfterAll - static void stopServer() { - ForkingClientServerIntegrationTestsSupport.stopGemFireServer(); - ForkingClientServerIntegrationTestsSupport.clearCacheServerPortAndPoolPortProperties(); - } - - @Test - void getServerEntryEvents() { - applicationContextRunner - .withPropertyValues("geode.region.regionName=myRegion", - "geode.supplier.event-expression=#root", - "geode.pool.connectType=server", - "geode.pool.hostAddresses=" + "localhost:" + System.getProperty("spring.data.gemfire.cache.server.port")) - .run(context -> { - Region region = context.getBean(Region.class); - region.put("hello", "world"); - region.put("foo", "bar"); - region.replace("hello", "dave"); - - 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.connectType=server", - "geode.pool.hostAddresses=" + "localhost:" + System.getProperty("spring.data.gemfire.cache.server.port")) - .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 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.connectType=server", - "geode.pool.hostAddresses=" + "localhost:" + System.getProperty("spring.data.gemfire.cache.server.port")) - .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 { - } -} diff --git a/supplier/geode-supplier/src/test/java/org/springframework/cloud/fn/supplier/geodeserver/GeodeServerTestConfiguration.java b/supplier/geode-supplier/src/test/java/org/springframework/cloud/fn/supplier/geodeserver/GeodeServerTestConfiguration.java deleted file mode 100644 index f517b42e..00000000 --- a/supplier/geode-supplier/src/test/java/org/springframework/cloud/fn/supplier/geodeserver/GeodeServerTestConfiguration.java +++ /dev/null @@ -1,47 +0,0 @@ -/* - * 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.geodeserver; - -import org.apache.geode.cache.GemFireCache; -import org.apache.geode.cache.Scope; - -import org.springframework.context.annotation.AnnotationConfigApplicationContext; -import org.springframework.context.annotation.Bean; -import org.springframework.data.gemfire.ReplicatedRegionFactoryBean; -import org.springframework.data.gemfire.config.annotation.CacheServerApplication; - -@CacheServerApplication -public class GeodeServerTestConfiguration { - - public static void main(String[] args) { - - AnnotationConfigApplicationContext applicationContext = - new AnnotationConfigApplicationContext(GeodeServerTestConfiguration.class); - - applicationContext.registerShutdownHook(); - } - - @Bean("myRegion") - public ReplicatedRegionFactoryBean myRegion(GemFireCache gemfireCache) { - - ReplicatedRegionFactoryBean regionFactoryBean = new ReplicatedRegionFactoryBean<>(); - regionFactoryBean.setScope(Scope.DISTRIBUTED_ACK); - regionFactoryBean.setCache(gemfireCache); - - return regionFactoryBean; - } -} diff --git a/supplier/geode-supplier/src/test/resources/logback-include.xml b/supplier/geode-supplier/src/test/resources/logback-include.xml deleted file mode 100644 index ae330ab3..00000000 --- a/supplier/geode-supplier/src/test/resources/logback-include.xml +++ /dev/null @@ -1,16 +0,0 @@ - - - - - - %d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n - - - - - - - - - - diff --git a/supplier/pom.xml b/supplier/pom.xml index 34cd96ff..53edf497 100644 --- a/supplier/pom.xml +++ b/supplier/pom.xml @@ -13,7 +13,6 @@ file-supplier ftp-supplier - geode-supplier http-supplier jdbc-supplier jms-supplier diff --git a/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierConfiguration.java b/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierConfiguration.java index 5d537096..67b118c8 100644 --- a/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierConfiguration.java +++ b/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierConfiguration.java @@ -23,7 +23,7 @@ import java.util.function.Supplier; import java.util.regex.Pattern; import java.util.stream.Stream; -import com.jcraft.jsch.ChannelSftp.LsEntry; +import org.apache.sshd.sftp.client.SftpClient; import org.reactivestreams.Publisher; import reactor.core.publisher.Flux; import reactor.core.publisher.MonoProcessor; @@ -84,6 +84,7 @@ import org.springframework.util.StringUtils; * @author Chris Schaefer * @author Christian Tzolov * @author David Turanski + * @author Corneil du Plessis */ @Configuration @@ -140,10 +141,12 @@ public class SftpSupplierConfiguration { * Configure the standard filters for SFTP inbound adapters. */ @Bean - public FileListFilter chainFilter(SftpSupplierProperties sftpSupplierProperties, - ConcurrentMetadataStore metadataStore) { + public FileListFilter chainFilter( + SftpSupplierProperties sftpSupplierProperties, + ConcurrentMetadataStore metadataStore + ) { - ChainFileListFilter chainFilter = new ChainFileListFilter<>(); + ChainFileListFilter chainFilter = new ChainFileListFilter<>(); if (StringUtils.hasText(sftpSupplierProperties.getFilenamePattern())) { chainFilter.addFilter(new SftpSimplePatternFileListFilter(sftpSupplierProperties.getFilenamePattern())); @@ -192,7 +195,7 @@ public class SftpSupplierConfiguration { @Bean public MessageSource targetMessageSource(SftpRemoteFileTemplate sftpTemplate, SftpSupplierProperties sftpSupplierProperties, - FileListFilter fileListFilter) { + FileListFilter fileListFilter) { return Sftp.inboundStreamingAdapter(sftpTemplate) .remoteDirectory(remoteDirectory(sftpSupplierProperties)) @@ -279,7 +282,7 @@ public class SftpSupplierConfiguration { @Bean public SftpInboundChannelAdapterSpec targetMessageSource(SftpSupplierProperties sftpSupplierProperties, SftpSupplierFactoryConfiguration.DelegatingFactoryWrapper delegatingFactoryWrapper, - FileListFilter fileListFilter) { + FileListFilter fileListFilter) { return Sftp .inboundAdapter(delegatingFactoryWrapper.getFactory()) @@ -377,13 +380,13 @@ public class SftpSupplierConfiguration { public MessageProcessor> lsEntryToStringTransformer() { return (Message message) -> { - LsEntry lsEntry = (LsEntry) message.getPayload(); + SftpClient.DirEntry dirEntry = (SftpClient.DirEntry) message.getPayload(); - String fileName = message.getHeaders().get(FileHeaders.REMOTE_DIRECTORY) + lsEntry.getFilename(); + String fileName = message.getHeaders().get(FileHeaders.REMOTE_DIRECTORY) + dirEntry.getFilename(); return MessageBuilder.withPayload(fileName) .copyHeaders(message.getHeaders()) - .setHeader(FILE_MODIFIED_TIME_HEADER, String.valueOf(lsEntry.getAttrs().getMTime())) + .setHeader(FILE_MODIFIED_TIME_HEADER, String.valueOf(dirEntry.getAttributes().getModifyTime())) .setHeader(MessageHeaders.CONTENT_TYPE, MediaType.TEXT_PLAIN) .build(); }; @@ -415,13 +418,13 @@ public class SftpSupplierConfiguration { private final String remoteDirectory; - private final SessionFactory sessionFactory; + private final SessionFactory sessionFactory; private final String remoteFileSeparator; private final SftpSupplierProperties.SortSpec sort; - SftpListingMessageProducer(SessionFactory sessionFactory, String remoteDirectory, + SftpListingMessageProducer(SessionFactory sessionFactory, String remoteDirectory, String remoteFileSeparator, SftpSupplierProperties.SortSpec sort) { this.sessionFactory = sessionFactory; @@ -431,10 +434,10 @@ public class SftpSupplierConfiguration { } public void listNames() { - Stream stream; + Stream stream; try { stream = Stream.of(this.sessionFactory.getSession().list(this.remoteDirectory)) - .filter(x -> !(x.getAttrs().isDir() || x.getAttrs().isLink())); + .filter(x -> !(x.getAttributes().isDirectory() || x.getAttributes().isSymbolicLink())); if (sort != null) { stream = stream.sorted(sort.comparator()); diff --git a/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierFactoryConfiguration.java b/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierFactoryConfiguration.java index 868617fb..3068bb6d 100644 --- a/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierFactoryConfiguration.java +++ b/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierFactoryConfiguration.java @@ -19,7 +19,7 @@ package org.springframework.cloud.fn.supplier.sftp; import java.util.HashMap; import java.util.Map; -import com.jcraft.jsch.ChannelSftp.LsEntry; +import org.apache.sshd.sftp.client.SftpClient; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.DisposableBean; @@ -40,19 +40,20 @@ import org.springframework.lang.Nullable; * @author Gary Russell * @author Artem Bilan * @author David Turanski + * @author Corneil du Plessis * */ public class SftpSupplierFactoryConfiguration { @Bean @ConditionalOnMissingBean - public SessionFactory sftpSessionFactory(SftpSupplierProperties properties, BeanFactory beanFactory) { + public SessionFactory sftpSessionFactory(SftpSupplierProperties properties, BeanFactory beanFactory) { return buildFactory(beanFactory, properties.getFactory()); } @Bean public DelegatingFactoryWrapper delegatingFactoryWrapper(SftpSupplierProperties properties, - SessionFactory defaultFactory, BeanFactory beanFactory) { + SessionFactory defaultFactory, BeanFactory beanFactory) { return new DelegatingFactoryWrapper(properties, defaultFactory, beanFactory); } @@ -73,7 +74,7 @@ public class SftpSupplierFactoryConfiguration { : null; } - static SessionFactory buildFactory(BeanFactory beanFactory, SftpSupplierProperties.Factory factory) { + static SessionFactory buildFactory(BeanFactory beanFactory, SftpSupplierProperties.Factory factory) { DefaultSftpSessionFactory sftpSessionFactory = new DefaultSftpSessionFactory(true); sftpSessionFactory.setHost(factory.getHost()); sftpSessionFactory.setPort(factory.getPort()); @@ -93,11 +94,11 @@ public class SftpSupplierFactoryConfiguration { public final static class DelegatingFactoryWrapper implements DisposableBean { - private final DelegatingSessionFactory delegatingSessionFactory; + private final DelegatingSessionFactory delegatingSessionFactory; - private final Map> factories = new HashMap<>(); + private final Map> factories = new HashMap<>(); - DelegatingFactoryWrapper(SftpSupplierProperties properties, SessionFactory defaultFactory, + DelegatingFactoryWrapper(SftpSupplierProperties properties, SessionFactory defaultFactory, BeanFactory beanFactory) { properties.getFactories().forEach((key, factory) -> { this.factories.put(key, SftpSupplierFactoryConfiguration.buildFactory(beanFactory, factory)); @@ -105,7 +106,7 @@ public class SftpSupplierFactoryConfiguration { this.delegatingSessionFactory = new DelegatingSessionFactory<>(this.factories, defaultFactory); } - public DelegatingSessionFactory getFactory() { + public DelegatingSessionFactory getFactory() { return this.delegatingSessionFactory; } diff --git a/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierProperties.java b/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierProperties.java index 547e3ec3..f81f4904 100644 --- a/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierProperties.java +++ b/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierProperties.java @@ -25,11 +25,11 @@ import java.util.List; import java.util.Map; import java.util.regex.Pattern; -import com.jcraft.jsch.ChannelSftp; import jakarta.validation.Valid; import jakarta.validation.constraints.AssertTrue; import jakarta.validation.constraints.NotBlank; import jakarta.validation.constraints.NotNull; +import org.apache.sshd.sftp.client.SftpClient; import org.hibernate.validator.constraints.Range; import org.springframework.boot.context.properties.ConfigurationProperties; @@ -44,6 +44,7 @@ import org.springframework.validation.annotation.Validated; * @author Artem Bilan * @author Chris Schaefer * @author David Turanski + * @author Corneil du Plessis */ @ConfigurationProperties("sftp.supplier") @Validated @@ -495,21 +496,21 @@ public class SftpSupplierProperties { DESC } - private Comparator getAttributeComparator() { + private Comparator getAttributeComparator() { switch (attribute) { case FILENAME: - return Comparator.comparing(ChannelSftp.LsEntry::getFilename); + return Comparator.comparing(SftpClient.DirEntry::getFilename); case ATIME: - return Comparator.comparing(x -> x.getAttrs().getATime()); + return Comparator.comparing(x -> x.getAttributes().getAccessTime()); case MTIME: - return Comparator.comparing(x -> x.getAttrs().getMTime()); + return Comparator.comparing(x -> x.getAttributes().getModifyTime()); } throw new UnsupportedOperationException("Unsupported sortBy attribute: " + attribute); } - public Comparator comparator() { - Comparator comparator = getAttributeComparator(); + public Comparator comparator() { + Comparator comparator = getAttributeComparator(); return dir == Dir.ASC ? comparator : comparator.reversed(); } }