diff --git a/applications/sink/geode-sink/pom.xml b/applications/sink/geode-sink/pom.xml index 9aeabee5..d459d16e 100644 --- a/applications/sink/geode-sink/pom.xml +++ b/applications/sink/geode-sink/pom.xml @@ -35,8 +35,9 @@ - org.springframework.cloud.fn - function-test-support + org.springframework.data + spring-data-geode-test + ${spring-data-geode-test.version} test diff --git a/applications/sink/geode-sink/src/test/java/org/springframework/cloud/stream/app/sink/geode/GeodeSinkTests.java b/applications/sink/geode-sink/src/test/java/org/springframework/cloud/stream/app/sink/geode/GeodeSinkTests.java index 10ff2277..7f5435ff 100644 --- a/applications/sink/geode-sink/src/test/java/org/springframework/cloud/stream/app/sink/geode/GeodeSinkTests.java +++ b/applications/sink/geode-sink/src/test/java/org/springframework/cloud/stream/app/sink/geode/GeodeSinkTests.java @@ -16,24 +16,27 @@ package org.springframework.cloud.stream.app.sink.geode; +import java.io.IOException; + +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.testcontainers.shaded.com.fasterxml.jackson.databind.ObjectMapper; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.boot.test.context.runner.ApplicationContextRunner; import org.springframework.cloud.fn.consumer.geode.GeodeConsumerConfiguration; -import org.springframework.cloud.fn.test.support.geode.GeodeContainer; -import org.springframework.cloud.fn.test.support.geode.GeodeContainerIntializer; +import org.springframework.cloud.stream.app.sink.geodeserver.GeodeServerTestConfiguration; import org.springframework.cloud.stream.binder.test.InputDestination; import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; import org.springframework.context.annotation.Import; +import org.springframework.data.gemfire.tests.integration.ForkingClientServerIntegrationTestsSupport; import org.springframework.messaging.support.GenericMessage; import static org.assertj.core.api.Assertions.assertThat; @@ -43,22 +46,22 @@ public class GeodeSinkTests { private static ApplicationContextRunner applicationContextRunner; - private static GeodeContainer geode; - private ObjectMapper objectMapper = new ObjectMapper(); @BeforeAll - static void setup() { - GeodeContainerIntializer initializer = new GeodeContainerIntializer( - geodeContainer -> { - geodeContainer.connectAndExecGfsh("create region --name=Stocks --type=REPLICATE"); - }); + static void setup() throws IOException { + ForkingClientServerIntegrationTestsSupport.startGemFireServer( + GeodeServerTestConfiguration.class); applicationContextRunner = new ApplicationContextRunner() .withUserConfiguration( TestChannelBinderConfiguration.getCompleteConfiguration(GeodeSinkTestApplication.class)); + } - geode = initializer.geodeContainer(); + @AfterAll + static void stopServer() { + ForkingClientServerIntegrationTestsSupport.stopGemFireServer(); + ForkingClientServerIntegrationTestsSupport.clearCacheServerPortAndPoolPortProperties(); } @Test @@ -70,7 +73,7 @@ public class GeodeSinkTests { "geode.consumer.json=true", "geode.consumer.key-expression=payload.getField('symbol')", "geode.pool.connectType=server", - "geode.pool.hostAddresses=" + "localhost:" + geode.getCacheServerPort()) + "geode.pool.hostAddresses=" + "localhost:" + System.getProperty("spring.data.gemfire.cache.server.port")) .run(context -> { InputDestination inputDestination = context.getBean(InputDestination.class); @@ -92,7 +95,7 @@ public class GeodeSinkTests { "geode.region.regionName=Stocks", "geode.consumer.key-expression='key'", "geode.pool.connectType=server", - "geode.pool.hostAddresses=" + "localhost:" + geode.getCacheServerPort()) + "geode.pool.hostAddresses=" + "localhost:" + System.getProperty("spring.data.gemfire.cache.server.port")) .run(context -> { InputDestination inputDestination = context.getBean(InputDestination.class); inputDestination.send(new GenericMessage<>("value")); diff --git a/applications/sink/geode-sink/src/test/java/org/springframework/cloud/stream/app/sink/geodeserver/GeodeServerTestConfiguration.java b/applications/sink/geode-sink/src/test/java/org/springframework/cloud/stream/app/sink/geodeserver/GeodeServerTestConfiguration.java new file mode 100644 index 00000000..82664ea9 --- /dev/null +++ b/applications/sink/geode-sink/src/test/java/org/springframework/cloud/stream/app/sink/geodeserver/GeodeServerTestConfiguration.java @@ -0,0 +1,46 @@ +/* + * 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.stream.app.sink.geodeserver; + +import org.apache.geode.cache.GemFireCache; + +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.data.gemfire.ReplicatedRegionFactoryBean; +import org.springframework.data.gemfire.config.annotation.CacheServerApplication; + +@CacheServerApplication +public class GeodeServerTestConfiguration { + + public static void main(String[] args) { + + AnnotationConfigApplicationContext applicationContext = + new AnnotationConfigApplicationContext(GeodeServerTestConfiguration.class); + + applicationContext.registerShutdownHook(); + } + + @Bean("Stocks") + public ReplicatedRegionFactoryBean stocksRegion(GemFireCache gemfireCache) { + + ReplicatedRegionFactoryBean regionFactoryBean = new ReplicatedRegionFactoryBean<>(); + + regionFactoryBean.setCache(gemfireCache); + + return regionFactoryBean; + } +} diff --git a/applications/source/geode-source/pom.xml b/applications/source/geode-source/pom.xml index 24298fb1..9ad65ea9 100644 --- a/applications/source/geode-source/pom.xml +++ b/applications/source/geode-source/pom.xml @@ -35,11 +35,13 @@ - org.springframework.cloud.fn - function-test-support + org.springframework.data + spring-data-geode-test + ${spring-data-geode-test.version} test + diff --git a/applications/source/geode-source/src/test/java/org/springframework/cloud/stream/app/source/geode/GeodeSourceTests.java b/applications/source/geode-source/src/test/java/org/springframework/cloud/stream/app/source/geode/GeodeSourceTests.java index 0db023d8..df3e73fd 100644 --- a/applications/source/geode-source/src/test/java/org/springframework/cloud/stream/app/source/geode/GeodeSourceTests.java +++ b/applications/source/geode-source/src/test/java/org/springframework/cloud/stream/app/source/geode/GeodeSourceTests.java @@ -16,6 +16,7 @@ package org.springframework.cloud.stream.app.source.geode; +import java.io.IOException; import java.time.Duration; import java.util.ArrayList; import java.util.List; @@ -27,6 +28,7 @@ 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; @@ -35,11 +37,11 @@ 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.geode.GeodeSupplierConfiguration; -import org.springframework.cloud.fn.test.support.geode.GeodeContainer; -import org.springframework.cloud.fn.test.support.geode.GeodeContainerIntializer; +import org.springframework.cloud.stream.app.source.geodeserver.GeodeServerTestConfiguration; import org.springframework.cloud.stream.binder.test.OutputDestination; import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; import org.springframework.context.annotation.Import; +import org.springframework.data.gemfire.tests.integration.ForkingClientServerIntegrationTestsSupport; import org.springframework.messaging.Message; import static org.assertj.core.api.Assertions.assertThat; @@ -49,22 +51,24 @@ public class GeodeSourceTests { private static ApplicationContextRunner applicationContextRunner; - private static GeodeContainer geode; private ObjectMapper objectMapper = new ObjectMapper(); @BeforeAll - static void setup() { - GeodeContainerIntializer initializer = new GeodeContainerIntializer( - geodeContainer -> { - geodeContainer.connectAndExecGfsh("create region --name=myRegion --type=REPLICATE"); - }); + static void setup() throws IOException { + ForkingClientServerIntegrationTestsSupport.startGemFireServer( + GeodeServerTestConfiguration.class); applicationContextRunner = new ApplicationContextRunner() .withUserConfiguration( TestChannelBinderConfiguration.getCompleteConfiguration(GeodeSourceTestApplication.class)); - geode = initializer.geodeContainer(); + } + + @AfterAll + static void stopServer() { + ForkingClientServerIntegrationTestsSupport.stopGemFireServer(); + ForkingClientServerIntegrationTestsSupport.clearCacheServerPortAndPoolPortProperties(); } @Test @@ -73,7 +77,7 @@ public class GeodeSourceTests { .withPropertyValues("geode.region.regionName=myRegion", "geode.supplier.event-expression=key+':'+newValue", "geode.pool.connectType=server", - "geode.pool.hostAddresses=" + "localhost:" + geode.getCacheServerPort(), + "geode.pool.hostAddresses=" + "localhost:" + System.getProperty("spring.data.gemfire.cache.server.port"), "spring.cloud.function.definition=geodeSupplier") .run(context -> { @@ -105,7 +109,7 @@ public class GeodeSourceTests { "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:" + geode.getCacheServerPort()) + "geode.pool.hostAddresses=" + "localhost:" + System.getProperty("spring.data.gemfire.cache.server.port")) .run(context -> { OutputDestination outputDestination = context.getBean(OutputDestination.class); // Using local region here diff --git a/applications/source/geode-source/src/test/java/org/springframework/cloud/stream/app/source/geodeserver/GeodeServerTestConfiguration.java b/applications/source/geode-source/src/test/java/org/springframework/cloud/stream/app/source/geodeserver/GeodeServerTestConfiguration.java new file mode 100644 index 00000000..266868ef --- /dev/null +++ b/applications/source/geode-source/src/test/java/org/springframework/cloud/stream/app/source/geodeserver/GeodeServerTestConfiguration.java @@ -0,0 +1,47 @@ +/* + * 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.stream.app.source.geodeserver; + +import org.apache.geode.cache.GemFireCache; +import org.apache.geode.cache.Scope; + +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.data.gemfire.ReplicatedRegionFactoryBean; +import org.springframework.data.gemfire.config.annotation.CacheServerApplication; + +@CacheServerApplication +public class GeodeServerTestConfiguration { + + public static void main(String[] args) { + + AnnotationConfigApplicationContext applicationContext = + new AnnotationConfigApplicationContext(GeodeServerTestConfiguration.class); + + applicationContext.registerShutdownHook(); + } + + @Bean("myRegion") + public ReplicatedRegionFactoryBean myRegion(GemFireCache gemfireCache) { + + ReplicatedRegionFactoryBean regionFactoryBean = new ReplicatedRegionFactoryBean<>(); + regionFactoryBean.setScope(Scope.DISTRIBUTED_ACK); + regionFactoryBean.setCache(gemfireCache); + + return regionFactoryBean; + } +} diff --git a/functions/common/function-test-support/pom.xml b/functions/common/function-test-support/pom.xml index 40b9b20f..4ef1608b 100644 --- a/functions/common/function-test-support/pom.xml +++ b/functions/common/function-test-support/pom.xml @@ -16,7 +16,6 @@ 1.6.0 1.1.1 - 1.2.7.RELEASE 4.0.3 @@ -28,12 +27,6 @@ compile - - org.springframework.data - spring-data-geode - true - - org.springframework.integration spring-integration-ftp diff --git a/functions/common/function-test-support/src/main/java/org/springframework/cloud/fn/test/support/geode/GeodeContainer.java b/functions/common/function-test-support/src/main/java/org/springframework/cloud/fn/test/support/geode/GeodeContainer.java deleted file mode 100644 index b00ff406..00000000 --- a/functions/common/function-test-support/src/main/java/org/springframework/cloud/fn/test/support/geode/GeodeContainer.java +++ /dev/null @@ -1,197 +0,0 @@ -/* - * Copyright 2020-2020 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * https://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.cloud.fn.test.support.geode; - -import java.io.IOException; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.LinkedList; -import java.util.List; -import java.util.concurrent.Future; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.testcontainers.containers.GenericContainer; - -import org.springframework.lang.NonNull; -import org.springframework.util.Assert; -import org.springframework.util.StringUtils; - -/** - * A Test Container that starts a Geode Locator and Server on configured ports. This also - * provides methods for executing one or more Gfsh commands. - */ -public class GeodeContainer extends GenericContainer { - private static Logger logger = LoggerFactory.getLogger(GeodeContainer.class); - - private final int locatorPort; - - private final int cacheServerPort; - - private final boolean useLocator; - - /** - * Create a Geode container from a Docker image. - * @param dockerImageName the name of the image. - * @param locatorPort the locator port. - * @param cacheServerPort the cache server port. - * @param useLocator set to use a locator. - */ - public GeodeContainer(@NonNull String dockerImageName, int locatorPort, int cacheServerPort, boolean useLocator) { - super(dockerImageName); - this.locatorPort = locatorPort; - this.cacheServerPort = cacheServerPort; - this.useLocator = useLocator; - } - - public GeodeContainer(@NonNull String dockerImageName, int locatorPort, int cacheServerPort) { - this(dockerImageName, locatorPort, cacheServerPort, false); - } - - /** - * Create a Geode Container from a {@code Future}. Test containers provides some - * implementations as image builders, such as - * {@link org.testcontainers.images.builder.ImageFromDockerfile}. - * @param image the image builder. - * @param locatorPort the locator port. - * @param cacheServerPort the server port. - * @param useLocator set to use a locator. - */ - public GeodeContainer(@NonNull Future image, int locatorPort, int cacheServerPort, boolean useLocator) { - super(image); - this.locatorPort = locatorPort; - this.cacheServerPort = cacheServerPort; - this.useLocator = useLocator; - } - - public GeodeContainer(@NonNull Future image, int locatorPort, int cacheServerPort) { - this(image, locatorPort, cacheServerPort, false); - } - - /** - * A convenience method to connect to a locator with Gfsh. - * @return the connect command String. - */ - public String connect() { - return useLocator ? "connect --locator=" + locators() : "connect --jmx-manager=localhost[1099]"; - } - - /** - * Get the locator port. - * @return the locator port. - */ - public int getLocatorPort() { - return locatorPort; - } - - /** - * Get the cache server port. - * @return the cache server port. - */ - public int getCacheServerPort() { - return cacheServerPort; - } - - /** - * - * @return Geode locators as host[port],... - */ - public String locators() { - return "localhost[" + locatorPort + "]"; - } - - /** - * Invoke the `gfsh` shell, Connect to the locator and execute the commands. - * @param command a list of commands to execute in a single `gfsh` invocation. - * @return the {@link org.testcontainers.containers.Container.ExecResult} - */ - public ExecResult connectAndExecGfsh(String... command) { - ArrayList args = new ArrayList<>(Arrays.asList(command)); - args.add(0, connect()); - return execInContainer(Gfsh.command(args.toArray(new String[args.size()])).commandParts()); - } - - /** - * Invoke the `gfsh` shell, and execute the commands. - * @param command a list of commands to execute in a single `gfsh` invocation. - * @return the {@link org.testcontainers.containers.Container.ExecResult} - */ - public ExecResult execGfsh(String... command) { - return execInContainer(Gfsh.command(command).commandParts()); - } - - /** - * Executes a command in the container, logging stdout and stderr and wrapping checked - * exceptions. - * @see GenericContainer#execInContainer(String...) - * @param command the command to execute. - * @return the {@link org.testcontainers.containers.Container.ExecResult} - */ - @Override - public ExecResult execInContainer(String... command) { - try { - ExecResult execResult = super.execInContainer(command); - logger.debug("stdout: {}", execResult.getStdout()); - if (execResult.getExitCode() != 0) { - logger.warn("stdout: {}", execResult.getStdout()); - logger.warn("stderr: {}", execResult.getStderr()); - } - return execResult; - } - catch (IOException e) { - throw new RuntimeException(e.getMessage(), e); - } - catch (InterruptedException e) { - throw new RuntimeException(e.getMessage(), e); - } - } - - /** - * Builds a Gfsh command. - */ - public final static class Gfsh { - - public static Command command(String... gfshCommands) { - return new Command(gfshCommands); - } - - public final static class Command { - private final List commandParts = new LinkedList<>(); - - private Command(String... gfshCommands) { - Assert.notEmpty(gfshCommands, "at least one command is required"); - for (String gfshCommand : gfshCommands) { - Assert.hasText(gfshCommand, "command must contain text"); - if (commandParts.size() == 0) { - commandParts.add("gfsh"); - } - commandParts.add("-e"); - commandParts.add(gfshCommand); - } - } - - public String[] commandParts() { - return commandParts.toArray(new String[commandParts.size()]); - } - - public String toString() { - return StringUtils.collectionToDelimitedString(commandParts, ","); - } - } - } - -} diff --git a/functions/common/function-test-support/src/main/java/org/springframework/cloud/fn/test/support/geode/GeodeContainerIntializer.java b/functions/common/function-test-support/src/main/java/org/springframework/cloud/fn/test/support/geode/GeodeContainerIntializer.java deleted file mode 100644 index c01b6618..00000000 --- a/functions/common/function-test-support/src/main/java/org/springframework/cloud/fn/test/support/geode/GeodeContainerIntializer.java +++ /dev/null @@ -1,114 +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; - -import org.springframework.util.SocketUtils; - -/** - * Creates and starts a {@link GeodeContainer} using available random ports for locator - * and server. Runs a {@code Consumer} post processor if provided. - * @author David Turanski - */ -public class GeodeContainerIntializer { - - private int locatorPort; - - private int cacheServerPort; - - private GeodeContainer geode; - - private Optional> postProcessor; - - private final boolean useLocator; - - /** - * Create, start, and perform post processing on a {@link GeodeContainer}. - * @param postProcessor a {@code Consumer} to run after the container is - * started. - */ - public GeodeContainerIntializer(Consumer postProcessor) { - this(postProcessor, false); - } - - public GeodeContainerIntializer(Consumer postProcessor, boolean useLocator) { - this.useLocator = useLocator; - - cacheServerPort = SocketUtils.findAvailableTcpPort(); - - locatorPort = SocketUtils.findAvailableTcpPort(); - - this.postProcessor = Optional.ofNullable(postProcessor); - - geode = new GeodeContainer(new ImageFromDockerfile() - .withFileFromClasspath("Dockerfile", "geode/Dockerfile") - .withBuildArg("CACHE_SERVER_PORT", String.valueOf(cacheServerPort)) - .withBuildArg("LOCATOR_PORT", String.valueOf(locatorPort)), - locatorPort, cacheServerPort, useLocator); - startContainer(); - } - - /** - * Create and start a {@link GeodeContainer}. - */ - public GeodeContainerIntializer() { - this(null, false); - } - - private void startContainer() { - // There is apparently no way to initialize Geode with random port mapping. Ports - // must be the same on client and server. - Consumer cmd = e -> { - e.withHostConfig(new HostConfig().withPortBindings( - new PortBinding(Ports.Binding.bindPort(cacheServerPort), new ExposedPort(cacheServerPort)), - new PortBinding(Ports.Binding.bindPort(locatorPort), new ExposedPort(locatorPort)))); - }; - - // Wait forever - - geode.withCommand("tail", "-f", "/dev/null").withCreateContainerCmdModifier(cmd).start(); - - if (useLocator) { - geode.execGfsh("start locator --name=Locator1 --hostname-for-clients=localhost --port=" + locatorPort); - geode.execGfsh(geode.connect(), - "start server --name=Server1 --hostname-for-clients=localhost --server-port=" + cacheServerPort); - } - else { - geode.execGfsh( - "start server --name=Server1 --hostname-for-clients=localhost --server-port=" + cacheServerPort + - " --J=-Dgemfire.jmx-manager=true --J=-Dgemfire.jmx-manager-start=true"); - } - postProcessor.ifPresent(geodeContainerConsumer -> geodeContainerConsumer.accept(geode)); - } - - /** - * @return the {@link GeodeContainer} instance. - */ - public GeodeContainer geodeContainer() { - return geode; - } - -} diff --git a/functions/common/function-test-support/src/main/resources/geode/Dockerfile b/functions/common/function-test-support/src/main/resources/geode/Dockerfile deleted file mode 100644 index df5847e5..00000000 --- a/functions/common/function-test-support/src/main/resources/geode/Dockerfile +++ /dev/null @@ -1,22 +0,0 @@ -# Licensed to the Apache Software Foundation (ASF) under one -# or more contributor license agreements. See the NOTICE file -# distributed with this work for additional information -# regarding copyright ownership. The ASF licenses this file -# to you under the Apache License, Version 2.0 (the -# "License"); you may not use this file except in compliance -# with the License. You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -FROM apachegeode/geode:1.12.0 - -ARG CACHE_SERVER_PORT=40404 -ARG LOCATOR_PORT=10334 - -EXPOSE ${LOCATOR_PORT} ${CACHE_SERVER_PORT} diff --git a/functions/consumer/geode-consumer/pom.xml b/functions/consumer/geode-consumer/pom.xml index 279086c2..dd249b38 100644 --- a/functions/consumer/geode-consumer/pom.xml +++ b/functions/consumer/geode-consumer/pom.xml @@ -31,11 +31,10 @@ config-common ${project.version} - - org.springframework.cloud.fn - function-test-support - ${revision} + org.springframework.data + spring-data-geode-test + ${spring-data-geode-test.version} test diff --git a/functions/consumer/geode-consumer/src/test/java/org/springframework/cloud/fn/consumer/geode/GeodeConsumerApplicationTests.java b/functions/consumer/geode-consumer/src/test/java/org/springframework/cloud/fn/consumer/geode/GeodeConsumerApplicationTests.java index 59f0b8a1..cc551a07 100644 --- a/functions/consumer/geode-consumer/src/test/java/org/springframework/cloud/fn/consumer/geode/GeodeConsumerApplicationTests.java +++ b/functions/consumer/geode-consumer/src/test/java/org/springframework/cloud/fn/consumer/geode/GeodeConsumerApplicationTests.java @@ -16,6 +16,7 @@ package org.springframework.cloud.fn.consumer.geode; +import java.io.IOException; import java.util.function.Consumer; import com.fasterxml.jackson.databind.ObjectMapper; @@ -24,14 +25,15 @@ 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.test.support.geode.GeodeContainer; -import org.springframework.cloud.fn.test.support.geode.GeodeContainerIntializer; +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; @@ -39,24 +41,23 @@ import static org.assertj.core.api.Assertions.assertThat; @Tag("integration") public class GeodeConsumerApplicationTests { - private static ApplicationContextRunner applicationContextRunner; - private static GeodeContainer geode; - private ObjectMapper objectMapper = new ObjectMapper(); @BeforeAll - static void setup() { - GeodeContainerIntializer initializer = new GeodeContainerIntializer( - geodeContainer -> { - geodeContainer.connectAndExecGfsh("create region --name=Stocks --type=REPLICATE"); - }); + static void setup() throws IOException { + ForkingClientServerIntegrationTestsSupport.startGemFireServer( + GeodeServerTestConfiguration.class); applicationContextRunner = new ApplicationContextRunner() .withUserConfiguration(GeodeConsumerTestApplication.class); + } - geode = initializer.geodeContainer(); + @AfterAll + static void stopServer() { + ForkingClientServerIntegrationTestsSupport.stopGemFireServer(); + ForkingClientServerIntegrationTestsSupport.clearCacheServerPortAndPoolPortProperties(); } @Test @@ -67,7 +68,7 @@ public class GeodeConsumerApplicationTests { "geode.consumer.json=true", "geode.consumer.key-expression=payload.getField('symbol')", "geode.pool.connectType=server", - "geode.pool.hostAddresses=" + "localhost:" + geode.getCacheServerPort()) + "geode.pool.hostAddresses=" + "localhost:" + System.getProperty("spring.data.gemfire.cache.server.port")) .run(context -> { Consumer> geodeConsumer = context.getBean("geodeConsumer", Consumer.class); @@ -88,7 +89,7 @@ public class GeodeConsumerApplicationTests { "geode.region.regionName=Stocks", "geode.consumer.key-expression='key'", "geode.pool.connectType=server", - "geode.pool.hostAddresses=" + "localhost:" + geode.getCacheServerPort()) + "geode.pool.hostAddresses=" + "localhost:" + System.getProperty("spring.data.gemfire.cache.server.port")) .run(context -> { Consumer> geodeConsumer = context.getBean("geodeConsumer", Consumer.class); diff --git a/functions/consumer/geode-consumer/src/test/java/org/springframework/cloud/fn/consumer/geodeserver/GeodeServerTestConfiguration.java b/functions/consumer/geode-consumer/src/test/java/org/springframework/cloud/fn/consumer/geodeserver/GeodeServerTestConfiguration.java new file mode 100644 index 00000000..4ad3a629 --- /dev/null +++ b/functions/consumer/geode-consumer/src/test/java/org/springframework/cloud/fn/consumer/geodeserver/GeodeServerTestConfiguration.java @@ -0,0 +1,45 @@ +/* + * Copyright 2020-2020 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.fn.consumer.geodeserver; + +import org.apache.geode.cache.GemFireCache; + +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.data.gemfire.ReplicatedRegionFactoryBean; +import org.springframework.data.gemfire.config.annotation.CacheServerApplication; + +@CacheServerApplication +public class GeodeServerTestConfiguration { + + public static void main(String[] args) { + + AnnotationConfigApplicationContext applicationContext = + new AnnotationConfigApplicationContext(GeodeServerTestConfiguration.class); + + applicationContext.registerShutdownHook(); + } + + @Bean("Stocks") + public ReplicatedRegionFactoryBean stocksRegion(GemFireCache gemfireCache) { + + ReplicatedRegionFactoryBean regionFactoryBean = new ReplicatedRegionFactoryBean<>(); + regionFactoryBean.setCache(gemfireCache); + + return regionFactoryBean; + } +} diff --git a/functions/supplier/geode-supplier/pom.xml b/functions/supplier/geode-supplier/pom.xml index 14d7690e..88b296ff 100644 --- a/functions/supplier/geode-supplier/pom.xml +++ b/functions/supplier/geode-supplier/pom.xml @@ -55,9 +55,9 @@ - org.springframework.cloud.fn - function-test-support - ${project.version} + org.springframework.data + spring-data-geode-test + ${spring-data-geode-test.version} test diff --git a/functions/supplier/geode-supplier/src/test/java/org/springframework/cloud/fn/supplier/geode/GeodeSupplierApplicationTests.java b/functions/supplier/geode-supplier/src/test/java/org/springframework/cloud/fn/supplier/geode/GeodeSupplierApplicationTests.java index 0178cd2d..e07e0816 100644 --- a/functions/supplier/geode-supplier/src/test/java/org/springframework/cloud/fn/supplier/geode/GeodeSupplierApplicationTests.java +++ b/functions/supplier/geode-supplier/src/test/java/org/springframework/cloud/fn/supplier/geode/GeodeSupplierApplicationTests.java @@ -16,6 +16,7 @@ package org.springframework.cloud.fn.supplier.geode; +import java.io.IOException; import java.time.Duration; import java.util.function.Supplier; @@ -27,6 +28,7 @@ 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; @@ -36,8 +38,8 @@ import reactor.test.StepVerifier; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.boot.test.context.runner.ApplicationContextRunner; import org.springframework.cloud.fn.common.geode.JsonPdxFunctions; -import org.springframework.cloud.fn.test.support.geode.GeodeContainer; -import org.springframework.cloud.fn.test.support.geode.GeodeContainerIntializer; +import 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; @@ -47,21 +49,21 @@ public class GeodeSupplierApplicationTests { private static ApplicationContextRunner applicationContextRunner; - private static GeodeContainer geode; - private final ObjectMapper objectMapper = new ObjectMapper(); @BeforeAll - static void setup() { - GeodeContainerIntializer initializer = new GeodeContainerIntializer( - geodeContainer -> { - geodeContainer.connectAndExecGfsh("create region --name=myRegion --type=REPLICATE"); - }); + static void setup() throws IOException { + ForkingClientServerIntegrationTestsSupport.startGemFireServer( + GeodeServerTestConfiguration.class); applicationContextRunner = new ApplicationContextRunner() .withUserConfiguration(GeodeSupplierTestApplication.class); + } - geode = initializer.geodeContainer(); + @AfterAll + static void stopServer() { + ForkingClientServerIntegrationTestsSupport.stopGemFireServer(); + ForkingClientServerIntegrationTestsSupport.clearCacheServerPortAndPoolPortProperties(); } @Test @@ -70,7 +72,7 @@ public class GeodeSupplierApplicationTests { .withPropertyValues("geode.region.regionName=myRegion", "geode.supplier.event-expression=#root", "geode.pool.connectType=server", - "geode.pool.hostAddresses=" + "localhost:" + geode.getCacheServerPort()) + "geode.pool.hostAddresses=" + "localhost:" + System.getProperty("spring.data.gemfire.cache.server.port")) .run(context -> { Region region = context.getBean(Region.class); region.put("hello", "world"); @@ -103,7 +105,7 @@ public class GeodeSupplierApplicationTests { "geode.region.regionName=myRegion", "geode.client.pdx-read-serialized=true", "geode.pool.connectType=server", - "geode.pool.hostAddresses=" + "localhost:" + geode.getCacheServerPort()) + "geode.pool.hostAddresses=" + "localhost:" + System.getProperty("spring.data.gemfire.cache.server.port")) .run(context -> { Supplier> geodeSupplier = context.getBean("geodeSupplier", Supplier.class); // Using local region here @@ -132,7 +134,7 @@ public class GeodeSupplierApplicationTests { "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:" + geode.getCacheServerPort()) + "geode.pool.hostAddresses=" + "localhost:" + System.getProperty("spring.data.gemfire.cache.server.port")) .run(context -> { Supplier> geodeCqSupplier = context.getBean("geodeSupplier", Supplier.class); // Using local region here diff --git a/functions/supplier/geode-supplier/src/test/java/org/springframework/cloud/fn/supplier/geodeserver/GeodeServerTestConfiguration.java b/functions/supplier/geode-supplier/src/test/java/org/springframework/cloud/fn/supplier/geodeserver/GeodeServerTestConfiguration.java new file mode 100644 index 00000000..f517b42e --- /dev/null +++ b/functions/supplier/geode-supplier/src/test/java/org/springframework/cloud/fn/supplier/geodeserver/GeodeServerTestConfiguration.java @@ -0,0 +1,47 @@ +/* + * Copyright 2020-2020 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.fn.supplier.geodeserver; + +import org.apache.geode.cache.GemFireCache; +import org.apache.geode.cache.Scope; + +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.data.gemfire.ReplicatedRegionFactoryBean; +import org.springframework.data.gemfire.config.annotation.CacheServerApplication; + +@CacheServerApplication +public class GeodeServerTestConfiguration { + + public static void main(String[] args) { + + AnnotationConfigApplicationContext applicationContext = + new AnnotationConfigApplicationContext(GeodeServerTestConfiguration.class); + + applicationContext.registerShutdownHook(); + } + + @Bean("myRegion") + public ReplicatedRegionFactoryBean myRegion(GemFireCache gemfireCache) { + + ReplicatedRegionFactoryBean regionFactoryBean = new ReplicatedRegionFactoryBean<>(); + regionFactoryBean.setScope(Scope.DISTRIBUTED_ACK); + regionFactoryBean.setCache(gemfireCache); + + return regionFactoryBean; + } +} diff --git a/stream-applications-build/pom.xml b/stream-applications-build/pom.xml index e5ffa344..fd019f55 100644 --- a/stream-applications-build/pom.xml +++ b/stream-applications-build/pom.xml @@ -44,6 +44,7 @@ 2.5.11.RELEASE 2.2.15.RELEASE 3.0.14.BUILD-SNAPSHOT + 0.0.21.RELEASE 1.15.2 1.2.5 2.22.2