1) Create the operator, operation & manager classes for gRPC request.
2) Use CompletableFuture to track CDS request flow.
Issue-ID: POLICY-2384
Change-Id: I84e30131a69c2d24c1871ceebced2b69194f619c
Signed-off-by: Ram Krishna Verma <ram_krishna.verma@bell.ca>
<version>${project.version}</version>
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>org.onap.policy.models.policy-models-interactions.model-actors</groupId>
+ <artifactId>actor.aai</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
<dependency>
<groupId>org.onap.policy.models.policy-models-interactions.model-impl</groupId>
<artifactId>events</artifactId>
--- /dev/null
+/*-
+ * ============LICENSE_START=======================================================
+ * Copyright (C) 2020 Bell Canada. All rights reserved.
+ * ================================================================================
+ * 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
+ *
+ * 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.controlloop.actor.cds;
+
+import java.util.concurrent.CompletableFuture;
+import org.onap.ccsdk.cds.controllerblueprints.common.api.EventType;
+import org.onap.ccsdk.cds.controllerblueprints.processing.api.ExecutionServiceOutput;
+import org.onap.policy.cds.api.CdsProcessorListener;
+import org.onap.policy.controlloop.actorserviceprovider.OperationOutcome;
+import org.onap.policy.controlloop.policy.PolicyResult;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * CDS Actor service-manager implementation.
+ */
+public class CdsActorServiceManager implements CdsProcessorListener {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(CdsActorServiceManager.class);
+
+ private final CompletableFuture<OperationOutcome> future;
+
+ private final OperationOutcome outcome;
+
+ /**
+ * Constructs the object.
+ *
+ * @param outcome the operation outcome to populate
+ * @param future the future to complete
+ */
+ public CdsActorServiceManager(OperationOutcome outcome, CompletableFuture<OperationOutcome> future) {
+ this.outcome = outcome;
+ this.future = future;
+ }
+
+ /**
+ * {@inheritDoc}.
+ */
+ @Override
+ public void onMessage(final ExecutionServiceOutput message) {
+ LOGGER.info("Received notification from CDS: {}", message);
+ EventType eventType = message.getStatus().getEventType();
+ switch (eventType) {
+ case EVENT_COMPONENT_PROCESSING:
+ LOGGER.info("CDS is processing the message: {}", message);
+ break;
+ case EVENT_COMPONENT_EXECUTED:
+ outcome.setResult(PolicyResult.SUCCESS);
+ future.complete(outcome);
+ break;
+ default:
+ outcome.setResult(PolicyResult.FAILURE);
+ future.complete(outcome);
+ break;
+ }
+ }
+
+ /**
+ * {@inheritDoc}.
+ */
+ @Override
+ public void onError(final Throwable throwable) {
+ future.completeExceptionally(throwable);
+ }
+}
import org.slf4j.LoggerFactory;
/**
- * CDS Actor service-provider implementation. This is a deploy dark feature for El-Alto release.
+ * CDS Actor service-provider implementation. This is a deploy dark feature for El-Alto
+ * release.
*/
public class CdsActorServiceProvider extends ActorImpl {
private static final Logger LOGGER = LoggerFactory.getLogger(CdsActorServiceProvider.class);
+ /**
+ * Constructs the object.
+ */
public CdsActorServiceProvider() {
super(CdsActorConstants.CDS_ACTOR);
+
+ addOperator(new GrpcOperator(CdsActorConstants.CDS_ACTOR, GrpcOperation.NAME, GrpcOperation::new));
}
+ // TODO old code: remove lines down to **HERE**
+
/**
* {@inheritDoc}.
*/
}
/**
- * Build the CDS ExecutionServiceInput request from the policy object and the AAI enriched parameters. TO-DO: Avoid
- * leaking Exceptions to the Kie Session thread. TBD item for Frankfurt release.
+ * Build the CDS ExecutionServiceInput request from the policy object and the AAI
+ * enriched parameters. TO-DO: Avoid leaking Exceptions to the Kie Session thread. TBD
+ * item for Frankfurt release.
*
* @param onset the event that is reporting the alert for policy to perform an action.
- * @param operation the control loop operation specifying the actor, operation, target, etc.
- * @param policy the policy specified from the yaml generated by CLAMP or through Policy API.
+ * @param operation the control loop operation specifying the actor, operation,
+ * target, etc.
+ * @param policy the policy specified from the yaml generated by CLAMP or through
+ * Policy API.
* @param aaiParams Map of enriched AAI attributes in node.attribute notation.
- * @return an Optional ExecutionServiceInput instance if valid else an Optional empty object is returned.
+ * @return an Optional ExecutionServiceInput instance if valid else an Optional empty
+ * object is returned.
*/
public Optional<ExecutionServiceInput> constructRequest(VirtualControlLoopEvent onset,
- ControlLoopOperation operation, Policy policy, Map<String, String> aaiParams) {
+ ControlLoopOperation operation, Policy policy, Map<String, String> aaiParams) {
- // For the current operational TOSCA policy model (yaml) CBA name and version are embedded in the payload
- // section, with the new policy type model being proposed in Frankfurt we will be able to move it out.
+ // For the current operational TOSCA policy model (yaml) CBA name and version are
+ // embedded in the payload
+ // section, with the new policy type model being proposed in Frankfurt we will be
+ // able to move it out.
Map<String, String> payload = policy.getPayload();
if (!validateCdsMandatoryParams(policy)) {
return Optional.empty();
String cbaName = payload.get(CdsActorConstants.KEY_CBA_NAME);
String cbaVersion = payload.get(CdsActorConstants.KEY_CBA_VERSION);
- // Retain only the payload by removing CBA name and version once they are extracted
+ // Retain only the payload by removing CBA name and version once they are
+ // extracted
// to be put in CDS request header.
payload.remove(CdsActorConstants.KEY_CBA_NAME);
payload.remove(CdsActorConstants.KEY_CBA_VERSION);
- // Embed payload from policy to ConfigDeployRequest object, serialize and inject into grpc request.
+ // Embed payload from policy to ConfigDeployRequest object, serialize and inject
+ // into grpc request.
String cbaActionName = policy.getRecipe();
CdsActionRequest request = new CdsActionRequest();
request.setPolicyPayload(payload);
// Inject AAI properties into payload map. Offer flexibility to the usecase
// implementation to inject whatever AAI parameters are of interest to them.
- // E.g. For vFW usecase El-Alto inject service-instance-id, generic-vnf-id as needed by CDS.
+ // E.g. For vFW usecase El-Alto inject service-instance-id, generic-vnf-id as
+ // needed by CDS.
request.setAaiProperties(aaiParams);
// Inject any additional event parameters that may be present in the onset event
Builder struct = Struct.newBuilder();
try {
String requestStr = request.generateCdsPayload();
- Preconditions.checkState(!Strings.isNullOrEmpty(requestStr), "Unable to build "
- + "config-deploy-request from payload parameters: {}", payload);
+ Preconditions.checkState(!Strings.isNullOrEmpty(requestStr),
+ "Unable to build " + "config-deploy-request from payload parameters: {}", payload);
JsonFormat.parser().merge(requestStr, struct);
} catch (InvalidProtocolBufferException | CoderException e) {
LOGGER.error("Failed to embed CDS payload string into the input request. blueprint({}:{}) for action({})",
- cbaName, cbaVersion, cbaActionName, e);
+ cbaName, cbaVersion, cbaActionName, e);
return Optional.empty();
}
// Build CDS gRPC request common-header
- CommonHeader commonHeader = CommonHeader.newBuilder()
- .setOriginatorId(CdsActorConstants.ORIGINATOR_ID)
- .setRequestId(onset.getRequestId().toString())
- .setSubRequestId(operation.getSubRequestId())
- .build();
+ CommonHeader commonHeader = CommonHeader.newBuilder().setOriginatorId(CdsActorConstants.ORIGINATOR_ID)
+ .setRequestId(onset.getRequestId().toString()).setSubRequestId(operation.getSubRequestId())
+ .build();
// Build CDS gRPC request action-identifier
- ActionIdentifiers actionIdentifiers = ActionIdentifiers.newBuilder()
- .setBlueprintName(cbaName)
- .setBlueprintVersion(cbaVersion)
- .setActionName(cbaActionName)
- .setMode(CdsActorConstants.CDS_MODE)
- .build();
+ ActionIdentifiers actionIdentifiers =
+ ActionIdentifiers.newBuilder().setBlueprintName(cbaName).setBlueprintVersion(cbaVersion)
+ .setActionName(cbaActionName).setMode(CdsActorConstants.CDS_MODE).build();
// Finally build the ExecutionServiceInput gRPC request object.
- ExecutionServiceInput executionServiceInput = ExecutionServiceInput.newBuilder()
- .setCommonHeader(commonHeader)
- .setActionIdentifiers(actionIdentifiers)
- .setPayload(struct.build())
- .build();
+ ExecutionServiceInput executionServiceInput = ExecutionServiceInput.newBuilder().setCommonHeader(commonHeader)
+ .setActionIdentifiers(actionIdentifiers).setPayload(struct.build()).build();
return Optional.of(executionServiceInput);
}
String cbaName = payload.get(CdsActorConstants.KEY_CBA_NAME);
String cbaVersion = payload.get(CdsActorConstants.KEY_CBA_VERSION);
String cbaActionName = policy.getRecipe();
- return !Strings.isNullOrEmpty(cbaName) && !Strings.isNullOrEmpty(cbaVersion) && !Strings
- .isNullOrEmpty(cbaActionName);
+ return !Strings.isNullOrEmpty(cbaName) && !Strings.isNullOrEmpty(cbaVersion)
+ && !Strings.isNullOrEmpty(cbaActionName);
}
public class CdsActorServiceManager implements CdsProcessorListener {
* @return the cds response.
*/
public CdsResponse sendRequestToCds(CdsProcessorGrpcClient cdsClient, CdsServerProperties cdsProps,
- ExecutionServiceInput executionServiceInput) {
+ ExecutionServiceInput executionServiceInput) {
try {
LOGGER.trace("Start CdsActorServiceProvider.executeCdsBlueprintProcessor {}.", executionServiceInput);
- // TO-DO: Handle requests asynchronously once the callback support is added to actors.
+ // TO-DO: Handle requests asynchronously once the callback support is
+ // added to actors.
CountDownLatch countDownLatch = cdsClient.sendRequest(executionServiceInput);
boolean status = countDownLatch.await(cdsProps.getTimeout(), TimeUnit.SECONDS);
if (!status) {
LOGGER.info("Status of the CDS gRPC request is: {}", getCdsStatus());
CdsResponse response = new CdsResponse();
- response.setRequestId(
- executionServiceInput != null && executionServiceInput.getCommonHeader() != null
- ? executionServiceInput.getCommonHeader().getRequestId() : null);
+ response.setRequestId(executionServiceInput != null && executionServiceInput.getCommonHeader() != null
+ ? executionServiceInput.getCommonHeader().getRequestId()
+ : null);
response.setStatus(this.getCdsStatus());
return response;
}
return cdsStatus.get();
}
}
+
+ // **HERE**
}
--- /dev/null
+/*-
+ * ============LICENSE_START=======================================================
+ * Copyright (C) 2020 Bell Canada. All rights reserved.
+ * ================================================================================
+ * 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
+ *
+ * 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.controlloop.actor.cds;
+
+import java.util.concurrent.Executor;
+import java.util.concurrent.TimeUnit;
+import lombok.Getter;
+import org.onap.policy.cds.properties.CdsServerProperties;
+import org.onap.policy.controlloop.actorserviceprovider.parameters.OperatorConfig;
+
+/**
+ * Configuration for gRPC Operators.
+ */
+@Getter
+public class GrpcConfig extends OperatorConfig {
+
+ /**
+ * Default timeout, in milliseconds, if none specified in the request.
+ */
+ private final long timeoutMs;
+
+ private CdsServerProperties cdsServerProperties;
+
+ /**
+ * Constructs the object.
+ *
+ * @param blockingExecutor executor to be used for tasks that may perform blocking I/O
+ * @param params operator parameters
+ */
+ public GrpcConfig(Executor blockingExecutor, CdsServerProperties params) {
+ super(blockingExecutor);
+ cdsServerProperties = params;
+ timeoutMs = TimeUnit.MILLISECONDS.convert(params.getTimeout(), TimeUnit.SECONDS);
+ }
+}
--- /dev/null
+/*-
+ * ============LICENSE_START=======================================================
+ * Copyright (C) 2020 Bell Canada. All rights reserved.
+ * ================================================================================
+ * 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
+ *
+ * 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.controlloop.actor.cds;
+
+import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.Struct;
+import com.google.protobuf.Struct.Builder;
+import com.google.protobuf.util.JsonFormat;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import lombok.Getter;
+import org.onap.ccsdk.cds.controllerblueprints.common.api.ActionIdentifiers;
+import org.onap.ccsdk.cds.controllerblueprints.common.api.CommonHeader;
+import org.onap.ccsdk.cds.controllerblueprints.processing.api.ExecutionServiceInput;
+import org.onap.policy.aai.AaiConstants;
+import org.onap.policy.aai.AaiCqResponse;
+import org.onap.policy.cds.client.CdsProcessorGrpcClient;
+import org.onap.policy.common.utils.coder.CoderException;
+import org.onap.policy.controlloop.actor.aai.AaiCustomQueryOperation;
+import org.onap.policy.controlloop.actor.cds.constants.CdsActorConstants;
+import org.onap.policy.controlloop.actor.cds.request.CdsActionRequest;
+import org.onap.policy.controlloop.actorserviceprovider.OperationOutcome;
+import org.onap.policy.controlloop.actorserviceprovider.impl.OperationPartial;
+import org.onap.policy.controlloop.actorserviceprovider.parameters.ControlLoopOperationParams;
+
+/**
+ * Operation that uses gRPC to send request to CDS.
+ *
+ */
+@Getter
+public class GrpcOperation extends OperationPartial {
+
+ public static final String NAME = "gRPC";
+
+ private CdsProcessorGrpcClient client;
+
+ /**
+ * Configuration for this operation.
+ */
+ private final GrpcConfig config;
+
+ /**
+ * Constructs the object.
+ *
+ * @param params operation parameters
+ * @param config configuration for this operation
+ */
+ public GrpcOperation(ControlLoopOperationParams params, GrpcConfig config) {
+ super(params, config);
+ this.config = config;
+ }
+
+ /**
+ * If no timeout is specified, then it returns the operator's configured timeout.
+ */
+ @Override
+ protected long getTimeoutMs(Integer timeoutSec) {
+ return (timeoutSec == null || timeoutSec == 0 ? config.getTimeoutMs() : super.getTimeoutMs(timeoutSec));
+ }
+
+ /**
+ * Ensures that A&AI customer query has been performed.
+ */
+ @Override
+ @SuppressWarnings("unchecked")
+ protected CompletableFuture<OperationOutcome> startPreprocessorAsync() {
+ ControlLoopOperationParams cqParams = params.toBuilder().actor(AaiConstants.ACTOR_NAME)
+ .operation(AaiCustomQueryOperation.NAME).payload(null).retry(null).timeoutSec(null).build();
+
+ // run Custom Query and Guard, in parallel
+ return allOf(() -> params.getContext().obtain(AaiCqResponse.CONTEXT_KEY, cqParams), this::startGuardAsync);
+ }
+
+ /**
+ * {@inheritDoc}.
+ */
+ @Override
+ protected CompletableFuture<OperationOutcome> startOperationAsync(int attempt, OperationOutcome outcome) {
+
+ CompletableFuture<OperationOutcome> future = new CompletableFuture<>();
+ client = new CdsProcessorGrpcClient(new CdsActorServiceManager(outcome, future),
+ config.getCdsServerProperties());
+
+ ExecutionServiceInput request = constructRequest(params);
+ client.sendRequest(request);
+ return future;
+ }
+
+ /**
+ * Build the CDS ExecutionServiceInput request from the policy object and the AAI
+ * enriched parameters. TO-DO: Avoid leaking Exceptions to the Kie Session thread. TBD
+ * item for Frankfurt release.
+ *
+ * @param params the control loop parameters specifying the onset, payload, etc.
+ * @return an ExecutionServiceInput instance.
+ */
+ public ExecutionServiceInput constructRequest(ControlLoopOperationParams params) {
+
+ // For the current operational TOSCA policy model (yaml) CBA name and version are
+ // embedded in the payload
+ // section, with the new policy type model being proposed in Frankfurt we will be
+ // able to move it out.
+ if (!validateCdsMandatoryParams(params)) {
+ throw new IllegalArgumentException("missing cds mandatory params - " + params);
+ }
+ Map<String, String> payload = convertPayloadMap(params.getPayload());
+ String cbaName = payload.get(CdsActorConstants.KEY_CBA_NAME);
+ String cbaVersion = payload.get(CdsActorConstants.KEY_CBA_VERSION);
+
+ // Retain only the payload by removing CBA name and version once they are
+ // extracted
+ // to be put in CDS request header.
+ payload.remove(CdsActorConstants.KEY_CBA_NAME);
+ payload.remove(CdsActorConstants.KEY_CBA_VERSION);
+
+ // Embed payload from policy to ConfigDeployRequest object, serialize and inject
+ // into grpc request.
+ String cbaActionName = params.getOperation();
+ CdsActionRequest request = new CdsActionRequest();
+ request.setPolicyPayload(payload);
+ request.setActionName(cbaActionName);
+ request.setResolutionKey(UUID.randomUUID().toString());
+
+ // Inject AAI properties into payload map. Offer flexibility to the usecase
+ // implementation to inject whatever AAI parameters are of interest to them.
+ // E.g. For vFW usecase El-Alto inject service-instance-id, generic-vnf-id as
+ // needed by CDS.
+ request.setAaiProperties(params.getContext().getEnrichment());
+
+ // Inject any additional event parameters that may be present in the onset event
+ if (params.getContext().getEvent().getAdditionalEventParams() != null) {
+ request.setAdditionalEventParams(params.getContext().getEvent().getAdditionalEventParams());
+ }
+
+ Builder struct = Struct.newBuilder();
+ try {
+ String requestStr = request.generateCdsPayload();
+ Preconditions.checkState(!Strings.isNullOrEmpty(requestStr),
+ "Unable to build " + "config-deploy-request from payload parameters: {}", payload);
+ JsonFormat.parser().merge(requestStr, struct);
+ } catch (InvalidProtocolBufferException | CoderException e) {
+ throw new IllegalArgumentException("Failed to embed CDS payload string into the input request. blueprint({"
+ + cbaName + "}:{" + cbaVersion + "}) for action({" + cbaActionName + "})", e);
+ }
+
+ // Build CDS gRPC request common-header
+ CommonHeader commonHeader = CommonHeader.newBuilder().setOriginatorId(CdsActorConstants.ORIGINATOR_ID)
+ .setRequestId(params.getContext().getEvent().getRequestId().toString())
+ .setSubRequestId(Integer.toString(0)).build();
+
+ // Build CDS gRPC request action-identifier
+ ActionIdentifiers actionIdentifiers =
+ ActionIdentifiers.newBuilder().setBlueprintName(cbaName).setBlueprintVersion(cbaVersion)
+ .setActionName(cbaActionName).setMode(CdsActorConstants.CDS_MODE).build();
+
+ // Finally build & return the ExecutionServiceInput gRPC request object.
+ return ExecutionServiceInput.newBuilder().setCommonHeader(commonHeader).setActionIdentifiers(actionIdentifiers)
+ .setPayload(struct.build()).build();
+ }
+
+ private Map<String, String> convertPayloadMap(Map<String, Object> payload) {
+ Map<String, String> convertedPayload = new HashMap<>();
+ for (Entry<String, Object> entry : payload.entrySet()) {
+ convertedPayload.put(entry.getKey(), entry.getValue().toString());
+ }
+ return convertedPayload;
+ }
+
+ private boolean validateCdsMandatoryParams(ControlLoopOperationParams params) {
+ if (params == null || params.getPayload() == null) {
+ return false;
+ }
+ Map<String, Object> payload = params.getPayload();
+ if (payload.get(CdsActorConstants.KEY_CBA_NAME) == null
+ || payload.get(CdsActorConstants.KEY_CBA_VERSION) == null) {
+ return false;
+ }
+ return !Strings.isNullOrEmpty(payload.get(CdsActorConstants.KEY_CBA_NAME).toString())
+ && !Strings.isNullOrEmpty(payload.get(CdsActorConstants.KEY_CBA_VERSION).toString())
+ && !Strings.isNullOrEmpty(params.getOperation());
+ }
+}
--- /dev/null
+/*-
+ * ============LICENSE_START=======================================================
+ * Copyright (C) 2020 Bell Canada. All rights reserved.
+ * ================================================================================
+ * 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
+ *
+ * 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.controlloop.actor.cds;
+
+import java.util.Map;
+import lombok.Getter;
+import org.onap.policy.cds.properties.CdsServerProperties;
+import org.onap.policy.common.parameters.ValidationResult;
+import org.onap.policy.controlloop.actorserviceprovider.Operation;
+import org.onap.policy.controlloop.actorserviceprovider.Util;
+import org.onap.policy.controlloop.actorserviceprovider.impl.OperationMaker;
+import org.onap.policy.controlloop.actorserviceprovider.impl.OperatorPartial;
+import org.onap.policy.controlloop.actorserviceprovider.parameters.ControlLoopOperationParams;
+import org.onap.policy.controlloop.actorserviceprovider.parameters.ParameterValidationRuntimeException;
+
+/**
+ * Operator that uses gRPC. The operator's parameters must be a
+ * {@link CdsServerProperties}.
+ */
+@Getter
+public class GrpcOperator extends OperatorPartial {
+
+ /**
+ * Function to make an operation.
+ */
+ private final OperationMaker<GrpcConfig, GrpcOperation> operationMaker;
+
+ /**
+ * Current configuration. This is set by {@link #doConfigure(Map)}.
+ */
+ private GrpcConfig currentConfig;
+
+ /**
+ * Constructs the object.
+ *
+ * @param actorName name of the actor with which this operator is associated
+ * @param name operation name
+ */
+ public GrpcOperator(String actorName, String name) {
+ this(actorName, name, null);
+ }
+
+ /**
+ * Constructs the object.
+ *
+ * @param actorName name of the actor with which this operator is associated
+ * @param name operation name
+ * @param operationMaker function to make an operation
+ */
+ public GrpcOperator(String actorName, String name, OperationMaker<GrpcConfig, GrpcOperation> operationMaker) {
+ super(actorName, name);
+ this.operationMaker = operationMaker;
+ }
+
+ /**
+ * Translates the parameters to an {@link CdsServerProperties} and then extracts the
+ * relevant values.
+ */
+ @Override
+ protected void doConfigure(Map<String, Object> parameters) {
+ currentConfig = makeConfiguration(parameters);
+ }
+
+ /**
+ * Makes a new configuration using the specified parameters.
+ *
+ * @param parameters operator parameters
+ * @return a new configuration
+ */
+ protected GrpcConfig makeConfiguration(Map<String, Object> parameters) {
+ CdsServerProperties params = Util.translate(getFullName(), parameters, CdsServerProperties.class);
+ ValidationResult result = params.validate();
+ if (!result.isValid()) {
+ throw new ParameterValidationRuntimeException("invalid parameters", result);
+ }
+ return new GrpcConfig(getBlockingExecutor(), params);
+ }
+
+ /**
+ * {@inheritDoc}.
+ */
+ @Override
+ public Operation buildOperation(ControlLoopOperationParams params) {
+ if (operationMaker == null) {
+ throw new UnsupportedOperationException("cannot make operation for " + getFullName());
+ }
+ verifyRunning();
+ return operationMaker.apply(params, currentConfig);
+ }
+}
Code Review / policy / models.git / commitdiff