forked from apache/geode
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
GEODE-3127: add function execution to new client protocol. (apache#1357)
* GEODE-3127: add function execution to new client protocol. This only supports function execution on a region so far. It assumes the default result collector, which returns a list of results, so the response message returns a list of results. * MessageExecutionContext uses InternalCache instead of Cache * Remove super.setUp calls from operation handlers because the @before annotation already ensures these are called before the @before methods of child classes.
- Loading branch information
1 parent
b22db64
commit 6011e09
Showing
22 changed files
with
730 additions
and
40 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
29 changes: 29 additions & 0 deletions
29
geode-protobuf-messages/src/main/proto/v1/function_API.proto
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,29 @@ | ||
| /* | ||
| * 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. | ||
| */ | ||
| syntax = "proto3"; | ||
| package org.apache.geode.internal.protocol.protobuf.v1; | ||
|
|
||
| import "v1/basicTypes.proto"; | ||
|
|
||
| message ExecuteFunctionOnRegionRequest { | ||
| string functionID = 1; | ||
| string region = 2; | ||
| EncodedValue arguments = 3; | ||
| repeated EncodedValue keyFilter = 4; | ||
| } | ||
|
|
||
| message ExecuteFunctionOnRegionResponse { | ||
| repeated EncodedValue results = 1; // some functions don't return arguments. | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
135 changes: 135 additions & 0 deletions
135
...ernal/protocol/protobuf/v1/operations/ExecuteFunctionOnRegionRequestOperationHandler.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,135 @@ | ||
| /* | ||
| * 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. | ||
| */ | ||
| package org.apache.geode.internal.protocol.protobuf.v1.operations; | ||
|
|
||
| import java.util.HashSet; | ||
| import java.util.List; | ||
| import java.util.Set; | ||
|
|
||
| import org.apache.logging.log4j.Logger; | ||
|
|
||
| import org.apache.geode.cache.Region; | ||
| import org.apache.geode.cache.execute.Execution; | ||
| import org.apache.geode.cache.execute.Function; | ||
| import org.apache.geode.cache.execute.FunctionException; | ||
| import org.apache.geode.cache.execute.FunctionService; | ||
| import org.apache.geode.cache.execute.ResultCollector; | ||
| import org.apache.geode.internal.exception.InvalidExecutionContextException; | ||
| import org.apache.geode.internal.logging.LogService; | ||
| import org.apache.geode.internal.protocol.operations.ProtobufOperationHandler; | ||
| import org.apache.geode.internal.protocol.protobuf.v1.BasicTypes; | ||
| import org.apache.geode.internal.protocol.protobuf.v1.ClientProtocol; | ||
| import org.apache.geode.internal.protocol.protobuf.v1.Failure; | ||
| import org.apache.geode.internal.protocol.protobuf.v1.FunctionAPI; | ||
| import org.apache.geode.internal.protocol.protobuf.v1.MessageExecutionContext; | ||
| import org.apache.geode.internal.protocol.protobuf.v1.ProtobufSerializationService; | ||
| import org.apache.geode.internal.protocol.protobuf.v1.Result; | ||
| import org.apache.geode.internal.protocol.protobuf.v1.Success; | ||
| import org.apache.geode.internal.protocol.protobuf.v1.serialization.exception.EncodingException; | ||
| import org.apache.geode.internal.protocol.protobuf.v1.state.exception.ConnectionStateException; | ||
| import org.apache.geode.internal.security.SecurityService; | ||
| import org.apache.geode.security.NotAuthorizedException; | ||
|
|
||
| public class ExecuteFunctionOnRegionRequestOperationHandler implements | ||
| ProtobufOperationHandler<FunctionAPI.ExecuteFunctionOnRegionRequest, FunctionAPI.ExecuteFunctionOnRegionResponse> { | ||
| @Override | ||
| public Result<FunctionAPI.ExecuteFunctionOnRegionResponse, ClientProtocol.ErrorResponse> process( | ||
| ProtobufSerializationService serializationService, | ||
| FunctionAPI.ExecuteFunctionOnRegionRequest request, | ||
| MessageExecutionContext messageExecutionContext) | ||
| throws InvalidExecutionContextException, ConnectionStateException { | ||
|
|
||
| final String functionID = request.getFunctionID(); | ||
| final String regionName = request.getRegion(); | ||
|
|
||
| final Function<?> function = FunctionService.getFunction(functionID); | ||
| if (function == null) { | ||
| return Failure.of(ClientProtocol.ErrorResponse.newBuilder() | ||
| .setError(BasicTypes.Error.newBuilder().setErrorCode(BasicTypes.ErrorCode.INVALID_REQUEST) | ||
| .setMessage("Function with ID \"" + functionID + "\" not found").build()) | ||
| .build()); | ||
| } | ||
|
|
||
| final Region<Object, Object> region = messageExecutionContext.getCache().getRegion(regionName); | ||
| if (region == null) { | ||
| return Failure.of(ClientProtocol.ErrorResponse.newBuilder() | ||
| .setError(BasicTypes.Error.newBuilder().setErrorCode(BasicTypes.ErrorCode.INVALID_REQUEST) | ||
| .setMessage("Region \"" + regionName + "\" not found")) | ||
| .build()); | ||
| } | ||
|
|
||
| final SecurityService securityService = messageExecutionContext.getCache().getSecurityService(); | ||
|
|
||
| try { | ||
| // check security for function. | ||
| function.getRequiredPermissions(regionName).forEach(securityService::authorize); | ||
| } catch (NotAuthorizedException ex) { | ||
| return Failure.of(ClientProtocol.ErrorResponse.newBuilder() | ||
| .setError(BasicTypes.Error.newBuilder() | ||
| .setMessage("Authorization failed for function \"" + functionID + "\"") | ||
| .setErrorCode(BasicTypes.ErrorCode.AUTHORIZATION_FAILED)) | ||
| .build()); | ||
| } | ||
|
|
||
| try { | ||
| Execution execution = FunctionService.onRegion(region); | ||
|
|
||
| final Object arguments = serializationService.decode(request.getArguments()); | ||
|
|
||
| if (arguments != null) { | ||
| execution = execution.setArguments(arguments); | ||
| } | ||
|
|
||
| execution = execution.withFilter(parseFilter(serializationService, request)); | ||
|
|
||
| final ResultCollector<Object, List<Object>> resultCollector = execution.execute(functionID); | ||
|
|
||
| if (function.hasResult()) { | ||
| List<Object> results = resultCollector.getResult(); | ||
|
|
||
| final FunctionAPI.ExecuteFunctionOnRegionResponse.Builder responseMessage = | ||
| FunctionAPI.ExecuteFunctionOnRegionResponse.newBuilder(); | ||
| for (Object result : results) { | ||
| responseMessage.addResults(serializationService.encode(result)); | ||
| } | ||
| return Success.of(responseMessage.build()); | ||
| } else { | ||
| // This is fire and forget. | ||
| return Success.of(FunctionAPI.ExecuteFunctionOnRegionResponse.newBuilder().build()); | ||
| } | ||
| } catch (FunctionException ex) { | ||
| return Failure.of(ClientProtocol.ErrorResponse.newBuilder() | ||
| .setError(BasicTypes.Error.newBuilder().setErrorCode(BasicTypes.ErrorCode.SERVER_ERROR) | ||
| .setMessage("Function execution failed: " + ex.toString())) | ||
| .build()); | ||
| } catch (EncodingException ex) { | ||
| return Failure.of(ClientProtocol.ErrorResponse.newBuilder() | ||
| .setError(BasicTypes.Error.newBuilder().setErrorCode(BasicTypes.ErrorCode.SERVER_ERROR) | ||
| .setMessage("Encoding failed: " + ex.toString())) | ||
| .build()); | ||
| } | ||
| } | ||
|
|
||
| private Set<Object> parseFilter(ProtobufSerializationService serializationService, | ||
| FunctionAPI.ExecuteFunctionOnRegionRequest request) throws EncodingException { | ||
| List<BasicTypes.EncodedValue> encodedFilter = request.getKeyFilterList(); | ||
| Set<Object> filter = new HashSet<>(); | ||
|
|
||
| for (BasicTypes.EncodedValue filterKey : encodedFilter) { | ||
| filter.add(serializationService.decode(filterKey)); | ||
| } | ||
| return filter; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.