Spring/spring-data-geode
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

DATAGEODE-295 - Functions return results from all servers.

Browse Source
This commit is contained in:
Patrick Johnson
2020-02-10 13:07:03 -08:00
committed by John Blum
parent bf8f610b6b
commit ae69760bf6
6 changed files with 292 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
spring-data-geode/src/main/java/org/springframework/data/gemfire/function/execution/AbstractFunctionExecution.java
View File

@@ -34,6 +34,7 @@ import org.springframework.util.Assert;
*
* @author David Turanski
* @author John Blum
* @author Patrick Johnson
*/
@SuppressWarnings("unused")
abstract class AbstractFunctionExecution {
@@ -137,7 +138,11 @@ abstract class AbstractFunctionExecution {
}
}
else {
results = (Iterable<T>) resultCollector.getResult();
if(resultCollector.getResult() instanceof Iterable) {
results = (Iterable<T>) resultCollector.getResult();
} else {
results = (Iterable<T>) Collections.singleton(resultCollector.getResult());
}
}
return replaceSingletonNullCollectionWithEmptyList(results);

12
spring-data-geode/src/main/java/org/springframework/data/gemfire/function/execution/GemfireFunctionProxyFactoryBean.java
View File

@@ -18,6 +18,7 @@ import java.util.stream.StreamSupport;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.data.gemfire.function.annotation.OnServers;
import org.springframework.data.gemfire.support.AbstractFactoryBeanSupport;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
@@ -25,6 +26,10 @@ import org.springframework.util.ClassUtils;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
<<<<<<<HEAD
=======
>>>>>>>DATAGEODE-295-Functions return results from all servers.
/**
* A Proxy {@link FactoryBean} for all non-Region Function Execution interfaces.
*
@@ -109,8 +114,11 @@ public class GemfireFunctionProxyFactoryBean extends AbstractFactoryBeanSupport<
protected Object invokeFunction(Method method, Object[] args) {
return getGemfireFunctionOperations()
.executeAndExtract(getFunctionExecutionMethodMetadata().getMethodMetadata(method).getFunctionId(), args);
GemfireFunctionOperations template = getGemfireFunctionOperations();
return method.getDeclaringClass().isAnnotationPresent(OnServers.class)
? template.execute(getFunctionExecutionMethodMetadata().getMethodMetadata(method).getFunctionId(), args)
: template.executeAndExtract(getFunctionExecutionMethodMetadata().getMethodMetadata(method).getFunctionId(), args);
}
protected Object resolveResult(MethodInvocation invocation, Object result) {

32
spring-data-geode/src/test/java/org/springframework/data/gemfire/function/execution/onservers/AllServersAdminFunctions.java Normal file
View File

@@ -0,0 +1,32 @@
/*
* 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.data.gemfire.function.execution.onservers;
import org.springframework.data.gemfire.function.annotation.FunctionId;
import org.springframework.data.gemfire.function.annotation.OnServers;
import java.util.List;
/**
* @author Patrick Johnson
*/
@OnServers
public interface AllServersAdminFunctions {
@FunctionId("GetAllMetricsFunction")
List<List<Metric>> getAllMetrics();
}

172
spring-data-geode/src/test/java/org/springframework/data/gemfire/function/execution/onservers/FunctionsReturnResultsFromAllServersIntegrationTests.java Normal file
View File

@@ -0,0 +1,172 @@
/*
* 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.data.gemfire.function.execution.onservers;
import org.apache.geode.StatisticDescriptor;
import org.apache.geode.Statistics;
import org.apache.geode.StatisticsType;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheFactory;
import org.apache.geode.cache.execute.Function;
import org.apache.geode.cache.execute.FunctionContext;
import org.apache.geode.cache.execute.FunctionService;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.internal.statistics.StatisticsManager;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.gemfire.config.annotation.ClientCacheApplication;
import org.springframework.data.gemfire.function.config.EnableGemfireFunctionExecutions;
import org.springframework.data.gemfire.process.ProcessWrapper;
import org.springframework.data.gemfire.test.support.ClientServerIntegrationTestsSupport;
import org.springframework.data.gemfire.transaction.config.EnableGemfireCacheTransactions;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringRunner;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Patrick Johnson
*/
@SuppressWarnings("unused")
@RunWith(SpringRunner.class)
@ContextConfiguration(classes = FunctionsReturnResultsFromAllServersIntegrationTests.Config.class)
public class FunctionsReturnResultsFromAllServersIntegrationTests extends ClientServerIntegrationTestsSupport {
private static final int PORT_1 = 40407;
private static final int PORT_2 = 40403;
private static ProcessWrapper gemfireServer1;
private static ProcessWrapper gemfireServer2;
@Autowired
private AllServersAdminFunctions allServersAdminFunctions;
@Autowired
private SingleServerAdminFunctions singleServerAdminFunctions;
@BeforeClass
public static void startGemFireServer() throws Exception {
gemfireServer1 = run(MetricsFunctionServerProcess.class,
String.format("-D%s=%d", GEMFIRE_CACHE_SERVER_PORT_PROPERTY, PORT_1));
waitForServerToStart(DEFAULT_HOSTNAME, PORT_1);
gemfireServer2 = run(MetricsFunctionServerProcess.class,
String.format("-D%s=%d", GEMFIRE_CACHE_SERVER_PORT_PROPERTY, PORT_2));
waitForServerToStart(DEFAULT_HOSTNAME, PORT_2);
}
@ClientCacheApplication(servers = {@ClientCacheApplication.Server(port = PORT_1), @ClientCacheApplication.Server(port = PORT_2)})
@Configuration
@EnableGemfireFunctionExecutions(basePackageClasses = AllServersAdminFunctions.class)
@EnableGemfireCacheTransactions
static class Config { }
@AfterClass
public static void stopGemFireServer() {
stop(gemfireServer1);
stop(gemfireServer2);
}
@Test
public void executeFunctionOnAllServers() {
List<List<Metric>> metrics = allServersAdminFunctions.getAllMetrics();
assertThat(metrics.size()).isEqualTo(2);
}
@Test
public void executeFunctionOnSingleServer() {
List<Metric> metrics = singleServerAdminFunctions.getAllMetrics();
assertThat(metrics.size()).isEqualTo(672);
}
static class MetricsFunctionServerProcess {
private static final int DEFAULT_CACHE_SERVER_PORT = 40404;
private static final String CACHE_SERVER_PORT_PROPERTY = "spring.data.gemfire.cache.server.port";
private static final String GEMFIRE_LOG_LEVEL = "error";
private static final String GEMFIRE_NAME = "MetricsServer" + getCacheServerPort();
public static void main(String[] args) throws Exception {
registerFunctions(startCacheServer(newGemFireCache()));
}
private static Cache newGemFireCache() {
return new CacheFactory()
.set("name", GEMFIRE_NAME)
.set("log-level", GEMFIRE_LOG_LEVEL)
.create();
}
private static Cache startCacheServer(Cache gemfireCache) throws IOException {
CacheServer cacheServer = gemfireCache.addCacheServer();
cacheServer.setPort(getCacheServerPort());
cacheServer.start();
return gemfireCache;
}
private static int getCacheServerPort() {
return Integer.getInteger(CACHE_SERVER_PORT_PROPERTY, DEFAULT_CACHE_SERVER_PORT);
}
private static Cache registerFunctions(Cache gemfireCache) {
FunctionService.registerFunction(new GetAllMetricsFunction());
return gemfireCache;
}
}
static class GetAllMetricsFunction implements Function<List<Metric>> {
private final InternalDistributedSystem system =
(InternalDistributedSystem) CacheFactory.getAnyInstance().getDistributedSystem();
@Override
public void execute(FunctionContext context) {
List<Metric> allMetrics = new ArrayList<>();
StatisticsManager statisticsManager = system.getStatisticsManager();
for (Statistics statistics : statisticsManager.getStatsList()) {
StatisticsType statisticsType = statistics.getType();
for (StatisticDescriptor descriptor : statisticsType.getStatistics()) {
String statName = descriptor.getName();
Metric metric = new Metric(statName, statistics.get(statName), statisticsType.getName(), statistics.getTextId());
allMetrics.add(metric);
}
}
context.getResultSender().lastResult(allMetrics);
}
@Override
public String getId() {
return getClass().getSimpleName();
}
}
}

40
spring-data-geode/src/test/java/org/springframework/data/gemfire/function/execution/onservers/Metric.java Normal file
View File

@@ -0,0 +1,40 @@
/*
* 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.data.gemfire.function.execution.onservers;
import java.io.Serializable;
/**
* @author Patrick Johnson
*/
public class Metric implements Serializable {
private String name;
private Number value;
private String category;
private String type;
public Metric(String name, Number value, String category, String type) {
this.name = name;
this.value = value;
this.category = category;
this.type = type;
}
}

32
spring-data-geode/src/test/java/org/springframework/data/gemfire/function/execution/onservers/SingleServerAdminFunctions.java Normal file
View File

@@ -0,0 +1,32 @@
/*
* 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.data.gemfire.function.execution.onservers;
import org.springframework.data.gemfire.function.annotation.FunctionId;
import org.springframework.data.gemfire.function.annotation.OnServer;
import java.util.List;
/**
* @author Patrick Johnson
*/
@OnServer
public interface SingleServerAdminFunctions {
@FunctionId("GetAllMetricsFunction")
List<Metric> getAllMetrics();
}