SpringCloud/spring-functions-catalog
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

[SB3 Update] Remove Apache Geode (#345)

Browse Source 
- Also update SFTP usage
This commit is contained in:
Corneil du Plessis
2022-10-07 15:22:57 +02:00
committed by GitHub
parent b5ed0bdf1e
commit c64b259d91
52 changed files with 74 additions and 2785 deletions
Download Patch File Download Diff File Expand all files Collapse all files

197
common/function-test-support/src/main/java/org/springframework/cloud/fn/test/support/geode/GeodeContainer.java
View File

@@ -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<String>}. Test containers provides some
* implementations as image builders, such as
* {@link org.testcontainers.images.builder.ImageFromDockerfile}.
* @param image the image builder.
* @param locatorPort the locator port.
* @param cacheServerPort the server port.
* @param useLocator set to use a locator.
*/
public GeodeContainer(@NonNull Future<String> image, int locatorPort, int cacheServerPort, boolean useLocator) {
super(image);
this.locatorPort = locatorPort;
this.cacheServerPort = cacheServerPort;
this.useLocator = useLocator;
}
public GeodeContainer(@NonNull Future<String> 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<String> args = new ArrayList<>(Arrays.asList(command));
args.add(0, connect());
return execInContainer(Gfsh.command(args.toArray(new String[args.size()])).commandParts());
}
/**
* Invoke the `gfsh` shell, and execute the commands.
* @param command a list of commands to execute in a single `gfsh` invocation.
* @return the {@link org.testcontainers.containers.Container.ExecResult}
*/
public ExecResult execGfsh(String... command) {
return execInContainer(Gfsh.command(command).commandParts());
}
/**
* Executes a command in the container, logging stdout and stderr and wrapping checked
* exceptions.
* @see GenericContainer#execInContainer(String...)
* @param command the command to execute.
* @return the {@link org.testcontainers.containers.Container.ExecResult}
*/
@Override
public ExecResult execInContainer(String... command) {
try {
ExecResult execResult = super.execInContainer(command);
logger.debug("stdout: {}", execResult.getStdout());
if (execResult.getExitCode() != 0) {
logger.warn("stdout: {}", execResult.getStdout());
logger.warn("stderr: {}", execResult.getStderr());
}
return execResult;
}
catch (IOException e) {
throw new RuntimeException(e.getMessage(), e);
}
catch (InterruptedException e) {
throw new RuntimeException(e.getMessage(), e);
}
}
/**
* Builds a Gfsh command.
*/
public final static class Gfsh {
public static Command command(String... gfshCommands) {
return new Command(gfshCommands);
}
public final static class Command {
private final List<String> commandParts = new LinkedList<>();
private Command(String... gfshCommands) {
Assert.notEmpty(gfshCommands, "at least one command is required");
for (String gfshCommand : gfshCommands) {
Assert.hasText(gfshCommand, "command must contain text");
if (commandParts.size() == 0) {
commandParts.add("gfsh");
}
commandParts.add("-e");
commandParts.add(gfshCommand);
}
}
public String[] commandParts() {
return commandParts.toArray(new String[commandParts.size()]);
}
public String toString() {
return StringUtils.collectionToDelimitedString(commandParts, ",");
}
}
}
}

112
common/function-test-support/src/main/java/org/springframework/cloud/fn/test/support/geode/GeodeContainerIntializer.java
View File

@@ -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<GeodeContainer>} post processor if provided.
* @author David Turanski
*/
public class GeodeContainerIntializer {
private int locatorPort;
private int cacheServerPort;
private GeodeContainer geode;
private Optional<Consumer<GeodeContainer>> postProcessor;
private final boolean useLocator;
/**
* Create, start, and perform post processing on a {@link GeodeContainer}.
* @param postProcessor a {@code Consumer<GeodeContainer>} to run after the container is
* started.
*/
public GeodeContainerIntializer(Consumer<GeodeContainer> postProcessor) {
this(postProcessor, false);
}
public GeodeContainerIntializer(Consumer<GeodeContainer> 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<CreateContainerCmd> 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;
}
}

302
common/function-test-support/src/main/java/org/springframework/cloud/fn/test/support/geode/SocketUtils.java
View File

@@ -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 &mdash; for example,
* for finding available ports on {@code localhost}.
*
* <p>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}.
*
* <p>{@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<Integer> 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<Integer> 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<Integer> 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<Integer> 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<Integer> 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<Integer> 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;
}
}
}

32
common/geode-common/pom.xml
View File

@@ -1,32 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>spring-functions-parent</artifactId>
<version>4.0.0-SNAPSHOT</version>
<relativePath>../../spring-functions-parent/pom.xml</relativePath>
</parent>
<artifactId>geode-common</artifactId>
<name>geode-common</name>
<description>Geode Common Components</description>
<properties>
<org-json.version>20200518</org-json.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.data</groupId>
<artifactId>spring-data-geode</artifactId>
</dependency>
<dependency>
<groupId>org.json</groupId>
<artifactId>json</artifactId>
<version>${org-json.version}</version>
</dependency>
</dependencies>
</project>

147
common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeClientCacheConfiguration.java
View File

@@ -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);
}
}
}

43
common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeClientCacheProperties.java
View File

@@ -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;
}
}

68
common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeClientRegionConfiguration.java
View File

@@ -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<Interest> keyInterests, GeodeRegionProperties properties) {
ClientRegionFactoryBean clientRegionFactoryBean = new ClientRegionFactoryBean();
clientRegionFactoryBean.setRegionName(properties.getRegionName());
clientRegionFactoryBean.setDataPolicy(DataPolicy.EMPTY);
if (!CollectionUtils.isEmpty(keyInterests)) {
clientRegionFactoryBean.setInterests(keyInterests.toArray(new Interest[keyInterests.size()]));
}
try {
clientRegionFactoryBean.setCache(clientCache);
}
catch (Exception e) {
throw new BeanCreationException(e.getMessage(), e);
}
return clientRegionFactoryBean;
}
}

90
common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodePoolProperties.java
View File

@@ -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;
}
}

45
common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeRegionProperties.java
View File

@@ -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;
}
}

100
common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeSecurityProperties.java
View File

@@ -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() {
}
}
}

170
common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeSslProperties.java
View File

@@ -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 <a href=https://geode.apache.org/docs/guide/14/managing/security/implementing_ssl.html>Implementing SSL in Geode</a>
*
* @author Christian Tzolov
*/
@ConfigurationProperties("geode.security.ssl")
@Validated
public class GeodeSslProperties {
private static final String USER_HOME_DIRECTORY = System.getProperty("user.home");
/**
* Name of the local trust store file copied in the local file system.
*/
public static final String LOCAL_TRUSTSTORE_FILE_NAME = "trusted.keystore";
/**
* Name of the local trust store file copied in the local file system.
*/
public static final String LOCAL_KEYSTORE_FILE_NAME = "keystore.keystore";
/**
* Local directory to cache the truststore and keystore files downloaded form the truststoreUri and keystoreUri locations.
*/
@NotBlank
private String userHomeDirectory = USER_HOME_DIRECTORY;
/**
* Location of the pre-created truststore URI to be used for connecting to the Geode cluster.
*/
private Resource truststoreUri;
/**
* Password for accessing the trust store.
*/
private String sslTruststorePassword;
/**
* Identifies the type of truststore used for SSL communications (e.g. JKS, PKCS11, etc.).
*/
@NotBlank
private String truststoreType = "JKS";
/**
* Location of the pre-created Keystore URI to be used for connecting to the Geode cluster.
*/
private Resource keystoreUri;
/**
* Password for accessing the keys truststore.
*/
private String sslKeystorePassword;
/**
* Identifies the type of Keystore used for SSL communications (e.g. JKS, PKCS11, etc.).
*/
@NotBlank
private String keystoreType = "JKS";
/**
* Configures the SSL ciphers used for secure Socket connections as an array of valid cipher names.
*/
@NotBlank
private String ciphers = "any";
public Resource getTruststoreUri() {
return truststoreUri;
}
public void setTruststoreUri(Resource truststoreUri) {
this.truststoreUri = truststoreUri;
}
public String getSslKeystorePassword() {
return sslKeystorePassword;
}
public void setSslKeystorePassword(String sslKeystorePassword) {
this.sslKeystorePassword = sslKeystorePassword;
}
public String getSslTruststorePassword() {
return sslTruststorePassword;
}
public void setSslTruststorePassword(String sslTruststorePassword) {
this.sslTruststorePassword = sslTruststorePassword;
}
public String getUserHomeDirectory() {
return userHomeDirectory;
}
public void setUserHomeDirectory(String userHomeDirectory) {
this.userHomeDirectory = userHomeDirectory;
}
public String getTruststoreType() {
return truststoreType;
}
public void setTruststoreType(String truststoreType) {
this.truststoreType = truststoreType;
}
public Resource getKeystoreUri() {
return keystoreUri;
}
public void setKeystoreUri(Resource keystoreUri) {
this.keystoreUri = keystoreUri;
}
public String getKeystoreType() {
return keystoreType;
}
public void setKeystoreType(String keystoreType) {
this.keystoreType = keystoreType;
}
public String getCiphers() {
return ciphers;
}
public void setCiphers(String ciphers) {
this.ciphers = ciphers;
}
public boolean isSslEnabled() {
return this.truststoreUri != null && this.keystoreUri != null;
}
@AssertTrue(message = "The truststoreUri and keystoreUri should together be either empty or not!")
private boolean isStoreUrisConsistent() {
return ((this.truststoreUri == null) && (this.keystoreUri == null)) ||
((this.truststoreUri != null) && (this.keystoreUri != null));
}
@AssertTrue(message = "The sslKeystorePassword and sslKeystorePassword must not be empty for non empty store URIs!")
private boolean isStorePasswordRequiredForValidStoreUri() {
return (!this.isSslEnabled()) ||
((this.sslKeystorePassword != null) && (this.sslTruststorePassword != null));
}
}

49
common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/InetSocketAddressConverterConfiguration.java
View File

@@ -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<String, InetSocketAddress> inetSocketAddressConverter() {
return new InetSocketAddressConverter();
}
public static class InetSocketAddressConverter implements Converter<String, InetSocketAddress> {
private static final Pattern HOST_AND_PORT_PATTERN = Pattern.compile("^\\s*(.*?):(\\d+)\\s*$");
@Override
public InetSocketAddress convert(String hostAddress) {
Matcher m = HOST_AND_PORT_PATTERN.matcher(hostAddress);
if (m.matches()) {
String host = m.group(1);
int port = Integer.parseInt(m.group(2));
return new InetSocketAddress(host, port);
}
throw new IllegalArgumentException(String.format("%s is not a valid [host]:[port] value.", hostAddress));
}
}
}

49
common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/JsonPdxFunctions.java
View File

@@ -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<String, PdxInstance> jsonToPdx() {
return JSONFormatter::fromJSON;
}
public static Function<PdxInstance, String> pdxToJson() {
return obj -> {
if (obj == null) {
return null;
}
if (obj instanceof PdxInstance) {
String json = JSONFormatter.toJSON(obj);
// de-pretty
return json.replaceAll("\\r\\n\\s*", "").replaceAll("\\n\\s*", "")
.replaceAll("\\s*:\\s*", ":").trim();
}
return obj.toString();
};
}
}

17
common/metadata-store-common/pom.xml
View File

@@ -51,7 +51,6 @@
<scope>test</scope>
</dependency>
<!--Gemfire-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-logging</artifactId>
@@ -63,22 +62,6 @@
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-gemfire</artifactId>
<optional>true</optional>
<exclusions>
<exclusion>
<groupId>org.springframework.data</groupId>
<artifactId>spring-data-gemfire</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.springframework.data</groupId>
<artifactId>spring-data-geode</artifactId>
<optional>true</optional>
</dependency>
<!--JDBC-->
<dependency>

44
common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/ClientCacheAutoConfiguration.java
View File

@@ -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
* <p>
* 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 {
}

35
common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreAutoConfiguration.java
View File

@@ -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<MetadataStoreListener> metadataStoreListenerObjectProvider) {
@SuppressWarnings("unchecked")
GemfireMetadataStore gemfireMetadataStore = new GemfireMetadataStore((Region<String, String>) region);
metadataStoreListenerObjectProvider.ifAvailable(gemfireMetadataStore::addListener);
return gemfireMetadataStore;
}
}
@ConditionalOnProperty(prefix = "metadata.store", name = "type", havingValue = "hazelcast")
static class Hazelcast {

26
common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreProperties.java
View File

@@ -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 {
/**

3
common/metadata-store-common/src/test/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreAutoConfigurationTests.java
View File

@@ -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,

1
common/pom.xml
View File

@@ -16,7 +16,6 @@
<module>file-common</module>
<module>ftp-common</module>
<module>function-test-support</module>
<module>geode-common</module>
<module>metadata-store-common</module>
<module>mqtt-common</module>
<module>redis-common</module>

48
consumer/geode-consumer/README.adoc
View File

@@ -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<Message<?>>`.
If necessary, can use the bean name `geodeCqSupplier` as a qualifier.
Once injected, you can invoke the `accept` method of the `Consumer`.
## Configuration Options
Required properties:
* `geode.region.region-name` - The name of the existing remote region.
* `geode.pool.host-addresses` - A comma delimited list of `host:port` pairs. By default these are locator addresses but are cache server addresses if you set `geode.pool.connect-type=server`.
For more information on the various options available, please see:
* link:src/main/java/org/springframework/cloud/fn/consumer/geode/cq/GeodeConsumerProperties.java[GeodeConsumerProperties.java] (`geode.consumer`)
Many of the options, common to functions that use Apache Geode, are configured by several `@ConfigurationProperties` classes which are included as needed:
* link:../../common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeRegionProperties.java[GeodeRegionProperties.java] (`geode.region`)
* link:../../common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodePoolProperties.java[GeodePoolProperties.java] (`geode.pool`)
* link:../../common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeSecurityProperties.java[GeodeSecurityProperties.java] (`geode.security`)
* link:../../common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeSslProperties.java[GeodeSslProperties.java] (`geode.security.ssl`)
## Examples
See this link:src/test/java/org/springframework/cloud/fn/consumer/geode/GeodeConsumerApplicationTests.java[test suite] for examples of how this consumer is used.
## Other usage
See this link:../../../applications/sink/geode-sink/README.adoc[README] where this consumer is used to create a Spring Cloud Stream application to write to a Geode region.

44
consumer/geode-consumer/pom.xml
View File

@@ -1,44 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>spring-functions-parent</artifactId>
<version>4.0.0-SNAPSHOT</version>
<relativePath>../../spring-functions-parent/pom.xml</relativePath>
</parent>
<artifactId>geode-consumer</artifactId>
<name>geode-consumer</name>
<description>Geode consumer</description>
<dependencies>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-gemfire</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>geode-common</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.geode</groupId>
<artifactId>spring-geode-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.geode</groupId>
<artifactId>spring-geode-starter-logging</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</project>

68
consumer/geode-consumer/src/main/java/org/springframework/cloud/fn/consumer/geode/GeodeConsumerConfiguration.java
View File

@@ -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<Message<?>> geodeConsumer(Function<Message<?>, Message<?>> geodeConsumerHandler,
CacheWritingMessageHandler cacheWriter) {
return message -> cacheWriter.handleMessage(geodeConsumerHandler.apply(message));
}
@Bean
Function<Message<?>, Message<?>> geodeConsumerHandler(GeodeConsumerProperties properties) {
return new GeodeConsumerHandler(properties.isJson());
}
@Bean
CacheWritingMessageHandler cacheWriter(Region<?, ?> region, GeodeConsumerProperties properties,
@Nullable ComponentCustomizer<CacheWritingMessageHandler> cacheWritingMessageHandlerCustomizer) {
CacheWritingMessageHandler messageHandler = new CacheWritingMessageHandler(region);
messageHandler.setCacheEntries(Collections.singletonMap(properties.getKeyExpression(), "payload"));
if (cacheWritingMessageHandlerCustomizer != null) {
cacheWritingMessageHandlerCustomizer.customize(messageHandler);
}
return messageHandler;
}
}

68
consumer/geode-consumer/src/main/java/org/springframework/cloud/fn/consumer/geode/GeodeConsumerHandler.java
View File

@@ -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<?>, Message<?>> {
private final Boolean convertToJson;
private final Function<String, PdxInstance> transformer = JsonPdxFunctions.jsonToPdx();
GeodeConsumerHandler(Boolean convertToJson) {
this.convertToJson = convertToJson;
}
@Override
public Message<?> apply(Message<?> message) {
Message<?> transformedMessage = message;
Object transformedPayload = message.getPayload();
if (convertToJson) {
Object payload = message.getPayload();
if (payload instanceof byte[]) {
transformedPayload = transformer.apply(new String((byte[]) payload));
}
else if (payload instanceof String) {
transformedPayload = transformer.apply((String) payload);
}
else {
throw new MessageConversionException(String.format(
"Cannot convert object of type %s", payload.getClass()
.getName()));
}
}
return MessageBuilder
.fromMessage(message)
.withPayload(transformedPayload)
.build();
}
}

57
consumer/geode-consumer/src/main/java/org/springframework/cloud/fn/consumer/geode/GeodeConsumerProperties.java
View File

@@ -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;
}
}

118
consumer/geode-consumer/src/test/java/org/springframework/cloud/fn/consumer/geode/GeodeConsumerApplicationTests.java
View File

@@ -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<Message<?>> geodeConsumer = context.getBean("geodeConsumer", Consumer.class);
String json = objectMapper.writeValueAsString(new Stock("XXX", 100.00));
geodeConsumer.accept(new GenericMessage<>(json));
Region<String, PdxInstance> region = context.getBean(Region.class);
PdxInstance instance = region.get("XXX");
assertThat(instance.getField("price")).isEqualTo(100.00);
region.close();
});
}
@Test
void consumeWithoutJsonEnabled() {
applicationContextRunner
.withPropertyValues(
"geode.region.regionName=Stocks",
"geode.consumer.key-expression='key'",
"geode.pool.connectType=server",
"geode.pool.hostAddresses=" + "localhost:" + System.getProperty("spring.data.gemfire.cache.server.port"))
.run(context -> {
Consumer<Message<?>> geodeConsumer = context.getBean("geodeConsumer", Consumer.class);
geodeConsumer.accept(new GenericMessage<>("value"));
Region<String, String> region = context.getBean(Region.class);
String value = region.get("key");
assertThat(value).isEqualTo("value");
region.close();
});
}
@Data
@AllArgsConstructor
@NoArgsConstructor
static class Stock {
private String symbol;
private double price;
}
@SpringBootApplication
static class GeodeConsumerTestApplication {
}
}

45
consumer/geode-consumer/src/test/java/org/springframework/cloud/fn/consumer/geodeserver/GeodeServerTestConfiguration.java
View File

@@ -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<Object, Object> stocksRegion(GemFireCache gemfireCache) {
ReplicatedRegionFactoryBean<Object, Object> regionFactoryBean = new ReplicatedRegionFactoryBean<>();
regionFactoryBean.setCache(gemfireCache);
return regionFactoryBean;
}
}

16
consumer/geode-consumer/src/test/resources/logback-include.xml
View File

@@ -1,16 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<included>
<appender name="console" class="ch.qos.logback.core.ConsoleAppender">
<encoder>
<pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
</encoder>
</appender>
<appender name="delegate" class="org.springframework.geode.logging.slf4j.logback.DelegatingAppender"/>
<logger name="com.gemstone.gemfire" level="${spring.boot.data.gemfire.log.level:-WARN}"/>
<logger name="org.apache.geode" level="${spring.boot.data.gemfire.log.level:-WARN}"/>
<logger name="org.jgroups" level="${spring.boot.data.gemfire.jgroups.log.level:-ERROR}"/>
</included>

1
consumer/pom.xml
View File

@@ -17,7 +17,6 @@
<module>analytics-consumer</module>
<module>file-consumer</module>
<module>ftp-consumer</module>
<module>geode-consumer</module>
<module>jdbc-consumer</module>
<module>log-consumer</module>
<module>mongodb-consumer</module>

5
consumer/redis-consumer/src/test/java/org/springframework/cloud/fn/consumer/redis/RedisConsumerTopicTests.java
View File

@@ -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 {

9
consumer/sftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerConfiguration.java
View File

@@ -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<ChannelSftp.LsEntry> ftpSessionFactory,
SessionFactory<SftpClient.DirEntry> ftpSessionFactory,
@Nullable ComponentCustomizer<SftpMessageHandlerSpec> sftpMessageHandlerSpecCustomizer) {
IntegrationFlowBuilder integrationFlowBuilder =

5
consumer/sftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerProperties.java
View File

@@ -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;

7
consumer/sftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerSessionFactoryConfiguration.java
View File

@@ -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<ChannelSftp.LsEntry> sftpSessionFactory(SftpConsumerProperties properties, BeanFactory beanFactory) {
public SessionFactory<SftpClient.DirEntry> 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<ChannelSftp.LsEntry> csf = new CachingSessionFactory<>(sftpSessionFactory);
CachingSessionFactory<SftpClient.DirEntry> csf = new CachingSessionFactory<>(sftpSessionFactory);
return csf;
}
else {

9
function/aggregator-function/pom.xml
View File

@@ -40,15 +40,6 @@
<artifactId>spring-boot-starter-data-redis</artifactId>
<scope>runtime</scope>
</dependency>
<!--Gemfire-->
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-gemfire</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.geode</groupId>
<artifactId>spring-geode-starter</artifactId>
</dependency>
<!--JDBC-->
<dependency>
<groupId>org.springframework.integration</groupId>

29
function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/AggregatorFunctionConfiguration.java
View File

@@ -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<Message<?>>, Flux<Message<?>>> aggregatorFunction(FluxMessageChannel inputChannel,
FluxMessageChannel outputChannel) {
public Function<Flux<Message<?>>, Flux<Message<?>>> 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<AggregatorFactoryBean> aggregatorCustomizer) {
@Nullable CorrelationStrategy correlationStrategy,
@Nullable ReleaseStrategy releaseStrategy,
@Nullable MessageGroupProcessor messageGroupProcessor,
@Nullable MessageGroupStore messageStore,
@Qualifier("outputChannel") MessageChannel outputChannel,
@Nullable ComponentCustomizer<AggregatorFactoryBean> 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 {
}

13
function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/ExcludeStoresAutoConfigurationEnvironmentPostProcessor.java
View File

@@ -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));
}
}

33
function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/MessageStoreConfiguration.java
View File

@@ -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<Object, Object> region) {
return new GemfireMessageStore(region);
}
}
@ConditionalOnClass(JdbcMessageStore.class)
@ConditionalOnProperty(prefix = AggregatorFunctionProperties.PREFIX,
name = "message-store-type",

6
function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/AbstractAggregatorFunctionTests.java
View File

@@ -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 {

3
function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/DefaultAggregatorTests.java
View File

@@ -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();
}
}

69
function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/GroupTimeOutAndGemfireMessageStoreAggregatorTests.java
View File

@@ -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<Message<?>> input =
Flux.just(MessageBuilder.withPayload("1")
.setHeader(IntegrationMessageHeaderAccessor.CORRELATION_ID, "my_correlation")
.build());
Flux<Message<?>> 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();
}
}

4
function/aggregator-function/src/test/java/org/springframework/cloud/fn/aggregator/JdbcMessageStoreAggregatorTests.java
View File

@@ -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();
}
}

68
supplier/geode-supplier/README.adoc
View File

@@ -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<Flux<?>>`.
Users have to subscribe to the returned `Flux` to receive the data.
## PDX Serialization
The supplier works with PDX serialized cache objects of type https://geode.apache.org/releases/latest/javadoc/org/apache/geode/pdx/PdxInstance.html[PdxInstance], which Geode uses to store objects that can be represented as JSON.
If the target region uses PDX serialization and you set `geode.client.pdx-read-serialized` to `true`, PdxInstance objects will be returned as JSON strings.
## Beans for injection
You can import the `GeodeSupplierConfiguration` configuration in a Spring Boot application and then inject the `geodeSupplier` bean as type `Supplier<Flux<T>>`, where `T` is the expected return type.
If necessary, can use the bean name `geodeSupplier` as a qualifier.
Once injected, you can invoke the `get` method of the `Supplier` and then subscribe to the returned `Flux` to initiate the stream.
## Configuration Options
Required properties:
* `geode.region.region-name` - The name of the existing remote region.
* `geode.pool.host-addresses` - A comma delimited list of `host:port` pairs. By default these are locator addresses but are cache server addresses if you set `geode.pool.connect-type=server`.
For more information on the various options available, please see:
* link:src/main/java/org/springframework/cloud/fn/supplier/geode/GeodeSupplierProperties.java[GeodeSupplierProperties.java] (`geode.supplier`)
Many of the options, common to functions that use Apache Geode, are configured by several `@ConfigurationProperties` classes which are included as needed:
* link:../../common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeClientCacheProperties.java[GeodeClientCacheProperties.java] (`geode.client`)
* link:../../common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeRegionProperties.java[GeodeRegionProperties.java] (`geode.region`)
* link:../../common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodePoolProperties.java[GeodePoolProperties.java] (`geode.pool`)
* link:../../common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeSecurityProperties.java[GeodeSecurityProperties.java] (`geode.security`)
* link:../../common/geode-common/src/main/java/org/springframework/cloud/fn/common/geode/GeodeSslProperties.java[GeodeSslProperties.java] (`geode.security.ssl`)
## Examples
See this link:src/test/java/org/springframework/cloud/fn/supplier/geode/GeodeSupplierApplicationTests.java[test suite] for examples of how this supplier is used.
## Other usage
See this link:../../../applications/source/geode-source/README.adoc[README] where this supplier is used to create a Spring Cloud Stream application to emit entry event data.

44
supplier/geode-supplier/pom.xml
View File

@@ -1,44 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>spring-functions-parent</artifactId>
<version>4.0.0-SNAPSHOT</version>
<relativePath>../../spring-functions-parent/pom.xml</relativePath>
</parent>
<artifactId>geode-supplier</artifactId>
<name>geode-supplier</name>
<description>geode supplier</description>
<dependencies>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-gemfire</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>geode-common</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.geode</groupId>
<artifactId>spring-geode-starter-logging</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.geode</groupId>
<artifactId>spring-geode-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</project>

194
supplier/geode-supplier/src/main/java/org/springframework/cloud/fn/supplier/geode/GeodeSupplierConfiguration.java
View File

@@ -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<Message<?>> cacheEvents = EmitterProcessor.create();
@Bean
public Supplier<Flux<?>> geodeSupplier() {
return () -> cacheEvents.map(Message::getPayload);
}
@Bean
FluxMessageChannel fluxChannel() {
FluxMessageChannel fluxChannel = new FluxMessageChannel();
fluxChannel.subscribe(cacheEvents);
return fluxChannel;
}
@Bean
MessageChannel convertToStringChannel() {
return new DirectChannel();
}
@Bean
MessageChannel routerChannel() {
return new DirectChannel();
}
@Bean
PayloadTypeRouter payloadTypeRouter(FluxMessageChannel fluxChannel) {
PayloadTypeRouter router = new PayloadTypeRouter();
router.setDefaultOutputChannel(fluxChannel);
router.setChannelMapping(PdxInstance.class.getName(), "convertToStringChannel");
return router;
}
@ConditionalOnMissingBean(Interest.class)
@Bean
public Interest<?> allKeysInterest() {
return new Interest<>(Interest.ALL_KEYS);
}
@Bean
IntegrationFlow startFlow(MessageChannel routerChannel, PayloadTypeRouter payloadTypeRouter) {
return IntegrationFlows.from(routerChannel)
.route(payloadTypeRouter)
.get();
}
@Bean
Function<PdxInstance, String> pdxToJson() {
return JsonPdxFunctions.pdxToJson();
}
@Bean
IntegrationFlow convertToString(MessageChannel convertToStringChannel, FluxMessageChannel fluxChannel,
Function<PdxInstance, String> pdxToJson) {
return IntegrationFlows.from(convertToStringChannel)
.transform(pdxToJson)
.channel(fluxChannel)
.get();
}
@Bean
@Conditional({ CQNotEnabled.class }) // ConditionalOnExpression causing SpEL errors.
public MessageProducer cacheListeningMessageProducer(
MessageChannel routerChannel,
GeodeSupplierProperties properties, Region<?, ?> region) {
CacheListeningMessageProducer cacheListeningMessageProducer = new CacheListeningMessageProducer(region);
cacheListeningMessageProducer.setOutputChannel(routerChannel);
cacheListeningMessageProducer.setPayloadExpression(
properties.getEventExpression());
return cacheListeningMessageProducer;
}
@ConditionalOnProperty(prefix = "geode.supplier", name = "query")
static class CQConfiguration {
@Bean
MessageProducer continuousQueryListener(
MessageChannel routerChannel,
ContinuousQueryListenerContainer continuousQueryListenerContainer,
GeodeSupplierProperties properties) {
ContinuousQueryMessageProducer continuousQueryMessageProducer = new ContinuousQueryMessageProducer(
continuousQueryListenerContainer,
properties.getQuery());
continuousQueryMessageProducer.setPayloadExpression(properties.getEventExpression());
continuousQueryMessageProducer.setOutputChannel(routerChannel);
return continuousQueryMessageProducer;
}
@Bean
ContinuousQueryListenerContainer continuousQueryListenerContainer(ClientCache clientCache) {
ContinuousQueryListenerContainer continuousQueryListenerContainer = new ContinuousQueryListenerContainer();
try {
continuousQueryListenerContainer.setCache(clientCache);
}
catch (Exception e) {
throw new BeanCreationException(e.getLocalizedMessage(), e);
}
return continuousQueryListenerContainer;
}
}
static class CQNotEnabled implements Condition {
@Override
public boolean matches(ConditionContext context, AnnotatedTypeMetadata metadata) {
return StringUtils.isEmpty(context.getEnvironment().getProperty("geode.supplier.query"));
}
}
}

61
supplier/geode-supplier/src/main/java/org/springframework/cloud/fn/supplier/geode/GeodeSupplierProperties.java
View File

@@ -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;
}
}

4
supplier/geode-supplier/src/main/resources/geode-client.properties
View File

@@ -1,4 +0,0 @@
#
# Geode client pool must be configured with subscriptions enabled for this component.
#
geode.pool.subscriptionEnabled=true

176
supplier/geode-supplier/src/test/java/org/springframework/cloud/fn/supplier/geode/GeodeSupplierApplicationTests.java
View File

@@ -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<Flux<EntryEvent>> geodeSupplier = context.getBean("geodeSupplier", Supplier.class);
StepVerifier.create(geodeSupplier.get()).assertNext(cacheEvent -> {
assertThat(cacheEvent.getOperation().isCreate()).isTrue();
assertThat(cacheEvent.getKey()).isEqualTo("hello");
assertThat(cacheEvent.getNewValue()).isEqualTo("world");
}).assertNext(cacheEvent -> {
assertThat(cacheEvent.getOperation().isCreate()).isTrue();
assertThat(cacheEvent.getKey()).isEqualTo("foo");
assertThat(cacheEvent.getNewValue()).isEqualTo("bar");
}).assertNext(cacheEvent -> {
assertThat(cacheEvent.getOperation().isUpdate()).isTrue();
assertThat(cacheEvent.getKey()).isEqualTo("hello");
assertThat(cacheEvent.getNewValue()).isEqualTo("dave");
}).thenCancel().verify(Duration.ofSeconds(10));
});
}
@Test
void pdxReadSerialized() {
applicationContextRunner
.withPropertyValues(
"geode.region.regionName=myRegion",
"geode.client.pdx-read-serialized=true",
"geode.pool.connectType=server",
"geode.pool.hostAddresses=" + "localhost:" + System.getProperty("spring.data.gemfire.cache.server.port"))
.run(context -> {
Supplier<Flux<String>> geodeSupplier = context.getBean("geodeSupplier", Supplier.class);
// Using local region here
Region<String, PdxInstance> region = context.getBean(Region.class);
Stock stock = new Stock("XXX", 140.00);
ObjectMapper objectMapper = new ObjectMapper();
String json = objectMapper.writeValueAsString(stock);
region.put(stock.getSymbol(), JsonPdxFunctions.jsonToPdx().apply(json));
StepVerifier.create(geodeSupplier.get()).assertNext(val -> {
try {
assertThat(objectMapper.readValue(val, Stock.class)).isEqualTo(stock);
}
catch (JsonProcessingException e) {
fail(e.getMessage());
}
}).thenCancel().verify(Duration.ofSeconds(10));
});
}
@Test
void 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<Flux<String>> geodeCqSupplier = context.getBean("geodeSupplier", Supplier.class);
// Using local region here
Region<String, PdxInstance> region = context.getBean(Region.class);
putStockEvent(region, new Stock("XXX", 140.00));
putStockEvent(region, new Stock("XXX", 140.20));
putStockEvent(region, new Stock("YYY", 110.00));
putStockEvent(region, new Stock("YYY", 110.01));
putStockEvent(region, new Stock("XXX", 139.80));
StepVerifier.create(geodeCqSupplier.get()).assertNext(val -> {
try {
assertThat(objectMapper.readValue(val, Stock.class)).isEqualTo(new Stock("XXX", 140.20));
}
catch (JsonProcessingException e) {
fail(e.getMessage());
}
}).thenCancel().verify(Duration.ofSeconds(10));
});
}
private void putStockEvent(Region<String, PdxInstance> region, Stock stock) throws JsonProcessingException {
String json = objectMapper.writeValueAsString(stock);
region.put(stock.getSymbol(), JsonPdxFunctions.jsonToPdx().apply(json));
}
@Data
@AllArgsConstructor
@NoArgsConstructor
static class Stock {
private String symbol;
private double price;
}
@SpringBootApplication
static class GeodeSupplierTestApplication {
}
}

47
supplier/geode-supplier/src/test/java/org/springframework/cloud/fn/supplier/geodeserver/GeodeServerTestConfiguration.java
View File

@@ -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<Object, Object> myRegion(GemFireCache gemfireCache) {
ReplicatedRegionFactoryBean<Object, Object> regionFactoryBean = new ReplicatedRegionFactoryBean<>();
regionFactoryBean.setScope(Scope.DISTRIBUTED_ACK);
regionFactoryBean.setCache(gemfireCache);
return regionFactoryBean;
}
}

16
supplier/geode-supplier/src/test/resources/logback-include.xml
View File

@@ -1,16 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<included>
<appender name="console" class="ch.qos.logback.core.ConsoleAppender">
<encoder>
<pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
</encoder>
</appender>
<appender name="delegate" class="org.springframework.geode.logging.slf4j.logback.DelegatingAppender"/>
<logger name="com.gemstone.gemfire" level="${spring.boot.data.gemfire.log.level:-WARN}"/>
<logger name="org.apache.geode" level="${spring.boot.data.gemfire.log.level:-WARN}"/>
<logger name="org.jgroups" level="${spring.boot.data.gemfire.jgroups.log.level:-ERROR}"/>
</included>

1
supplier/pom.xml
View File

@@ -13,7 +13,6 @@
<modules>
<module>file-supplier</module>
<module>ftp-supplier</module>
<module>geode-supplier</module>
<module>http-supplier</module>
<module>jdbc-supplier</module>
<module>jms-supplier</module>

29
supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierConfiguration.java
View File

@@ -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<LsEntry> chainFilter(SftpSupplierProperties sftpSupplierProperties,
ConcurrentMetadataStore metadataStore) {
public FileListFilter<SftpClient.DirEntry> chainFilter(
SftpSupplierProperties sftpSupplierProperties,
ConcurrentMetadataStore metadataStore
) {
ChainFileListFilter<LsEntry> chainFilter = new ChainFileListFilter<>();
ChainFileListFilter<SftpClient.DirEntry> 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<LsEntry> fileListFilter) {
FileListFilter<SftpClient.DirEntry> 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<LsEntry> fileListFilter) {
FileListFilter<SftpClient.DirEntry> fileListFilter) {
return Sftp
.inboundAdapter(delegatingFactoryWrapper.getFactory())
@@ -377,13 +380,13 @@ public class SftpSupplierConfiguration {
public MessageProcessor<Message<?>> 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<LsEntry> sessionFactory;
private final SessionFactory<SftpClient.DirEntry> sessionFactory;
private final String remoteFileSeparator;
private final SftpSupplierProperties.SortSpec sort;
SftpListingMessageProducer(SessionFactory<LsEntry> sessionFactory, String remoteDirectory,
SftpListingMessageProducer(SessionFactory<SftpClient.DirEntry> sessionFactory, String remoteDirectory,
String remoteFileSeparator, SftpSupplierProperties.SortSpec sort) {
this.sessionFactory = sessionFactory;
@@ -431,10 +434,10 @@ public class SftpSupplierConfiguration {
}
public void listNames() {
Stream<LsEntry> stream;
Stream<SftpClient.DirEntry> 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());

17
supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierFactoryConfiguration.java
View File

@@ -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<LsEntry> sftpSessionFactory(SftpSupplierProperties properties, BeanFactory beanFactory) {
public SessionFactory<SftpClient.DirEntry> sftpSessionFactory(SftpSupplierProperties properties, BeanFactory beanFactory) {
return buildFactory(beanFactory, properties.getFactory());
}
@Bean
public DelegatingFactoryWrapper delegatingFactoryWrapper(SftpSupplierProperties properties,
SessionFactory<LsEntry> defaultFactory, BeanFactory beanFactory) {
SessionFactory<SftpClient.DirEntry> defaultFactory, BeanFactory beanFactory) {
return new DelegatingFactoryWrapper(properties, defaultFactory, beanFactory);
}
@@ -73,7 +74,7 @@ public class SftpSupplierFactoryConfiguration {
: null;
}
static SessionFactory<LsEntry> buildFactory(BeanFactory beanFactory, SftpSupplierProperties.Factory factory) {
static SessionFactory<SftpClient.DirEntry> 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<LsEntry> delegatingSessionFactory;
private final DelegatingSessionFactory<SftpClient.DirEntry> delegatingSessionFactory;
private final Map<Object, SessionFactory<LsEntry>> factories = new HashMap<>();
private final Map<Object, SessionFactory<SftpClient.DirEntry>> factories = new HashMap<>();
DelegatingFactoryWrapper(SftpSupplierProperties properties, SessionFactory<LsEntry> defaultFactory,
DelegatingFactoryWrapper(SftpSupplierProperties properties, SessionFactory<SftpClient.DirEntry> 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<LsEntry> getFactory() {
public DelegatingSessionFactory<SftpClient.DirEntry> getFactory() {
return this.delegatingSessionFactory;
}

15
supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierProperties.java
View File

@@ -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<ChannelSftp.LsEntry> getAttributeComparator() {
private Comparator<SftpClient.DirEntry> 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<ChannelSftp.LsEntry> comparator() {
Comparator<ChannelSftp.LsEntry> comparator = getAttributeComparator();
public Comparator<SftpClient.DirEntry> comparator() {
Comparator<SftpClient.DirEntry> comparator = getAttributeComparator();
return dir == Dir.ASC ? comparator : comparator.reversed();
}
}