See: Description
Interface | Description |
---|---|
ErrorProvider |
Provides a method to set an error record in vault.
|
Message |
Provides methods to update the content of a Spark message.
|
MessageAttributeValueType<T> |
Type of data supported by the Vault Java SDK.
|
MessageContext |
Contains contextual information about a Message.
|
MessageDeliveryError |
Provides methods to retrieve the failure type and error message of the failed Spark message delivery.
|
MessageDeliveryEventHandler |
Represents a queue message delivery event handler.
|
MessageDeliveryEventHandlerContext |
Retrieves the context and delivery error of the failed Spark message delivery.
|
MessageDeliveryEventHandlerSendContext |
Retrieves the context and sets tokens for the outbound Spark message.
|
MessageProcessor |
Invoked to process a message received by an inbound queue.
|
PutMessageResponse |
The response returned by
QueueService.putMessage(Message) . |
PutMessageResult |
Represents the result for a specific connection during
QueueService.putMessage(Message) . |
QueueService |
Provides methods to create and put messages in a queue.
|
Enum | Description |
---|---|
MessageDeliveryErrorType |
Error types that may occur when sending a Spark message.
|
Annotation Type | Description |
---|---|
MessageDeliveryEventHandlerInfo |
Indicates a class is a
MessageDeliveryEventHandler . |
MessageProcessorInfo |
Indicates a class is a
MessageProcessor . |
To use Spark messages, you must first set up a Connection and Queue in Vault Admin for each vault that needs to send or receive a message. Connections can be set up for the local vault, with another vault, or with an external application. Learn more about Spark Messaging in the Developer Portal.
@MessageProcessorInfo()
public class ProcessStudy implements MessageProcessor { public void execute(MessageContext context) { String docId = context.getMessage().getAttribute("docId",MessageAttributeValueType.STRING); String docName = context.getMessage().getAttribute("docName",MessageAttributeValueType.STRING); String remoteVaultId = context.getRemoteVaultId(); HttpService httpService = ServiceLocator.locate(HttpService.class); // When a message is received, create a CrossLink to the document in the source vault. HttpRequest httpRequest = httpService.newLocalHttpRequest() .setMethod(HttpMethod.POST) .setBodyParam("source_document_id__v",docId) .setBodyParam("source_vault_id__v", remoteVaultId) .setBodyParam("source_binding_rule__v", "Latest version") .setBodyParam("name__v",docName) .setBodyParam("type__v", "TypeX") .setBodyParam("lifecycle__v", "General Lifecycle") .appendPath("/api/v18.3/objects/documents"); httpService.send(httpRequest,HttpResponseBodyValueType.JSONDATA) .onSuccess(response -> { // After CrossLink is created, start the Approval workflow JsonData body = response.getResponse(); String crossLinkDocId = body.getJsonObject().getValue("id", JsonValueType.NUMBER).toString(); HttpRequest startWorkflowRequest = httpService.newLocalHttpRequest() .setMethod(HttpMethod.PUT) .setBodyParam("Approver", RequestContext.get().getCurrentUserId()) .setBodyParam("dueDate", LocalDate.now().plusDays(30).toString()) .appendPath("/api/v18.3/objects/documents/" + crossLinkDocId + "/versions/0/1/lifecycle_actions/startApproval"); httpService.send(startWorkflowRequest,HttpResponseBodyValueType.STRING) .onError(starkWorkflowResponse -> { int startWorkflowResponseCode = starkWorkflowResponse.getHttpResponse().getHttpStatusCode(); ServiceLocator.locate(LogService.class).info("RESPONSE: " + responseCode); }) .execute(); }) .onError(response -> { int responseCode = response.getHttpResponse().getHttpStatusCode(); ServiceLocator.locate(LogService.class).info("RESPONSE: " + responseCode); }) .execute(); } }
RollbackException
when the operation fails.
@MessageProcessorInfo()
public class ProcessProduct implements MessageProcessor { {@literal @}Override public void execute(MessageContext context) { RecordService recordService = ServiceLocator.locate(RecordService.class); Record record = recordService.newRecord("product__v"); record.setValue("name__v", "productName"); List<BatchOperationError> errors = new ArrayList<>(); recordService.batchSaveRecords(VaultCollections.asList(record)). onErrors(le -> errors.addAll(le)). execute(); if (! errors.isEmpty()) { // Create a new error record and set to the ErrorProvider Record errRecord = recordService.newRecord("exception_message__sys"); // Below are 5 required fields errRecord.setValue("name__v", "productCreationError"); errRecord.setValue("error_message__sys", "Product batch operation error"); errRecord.setValue("integration__sys", "V2T000000001002"); errRecord.setValue("integration_point__sys", "V2U000000001001"); errRecord.setValue("error_type__sys", Collections.singletonList("item_processing_error__sys")); messageContext.getErrorProvider().setError(errRecord); // Rollback the entire transaction throw new RollbackException("TRACKED_AND_PROPAGATED_ERRORS", "Product batch operation error"); } } }
NotificationParameters
.
@MessageDeliveryEventHandlerInfo()
public class MessageDeliveryHandler implements MessageDeliveryEventHandler { public void onError(MessageDeliveryEventHandlerContext messageDeliveryEventHandlerContext) { // Instantiate an instance of the QueueService which is used to put messages back on the queue QueueService queueService = ServiceLocator.locate(QueueService.class); // Resend Message // Get Message Context MessageContext messageContext = messageDeliveryEventHandlerContext.getMessageContext(); // Get Queue Name and original Message String queueName = messageContext.getRemoteQueueName(); Message originalMessage = messageContext.getMessage(); PutMessageResponse response = queueService.putMessage(originalMessage); // Instantiate RecordService RecordService recordService = ServiceLocator.locate(RecordService.class); MessageDeliveryError messageDeliveryError = messageDeliveryEventHandlerContext.getMessageDeliveryError(); MessageDeliveryErrorType messageDeliveryErrorType = messageDeliveryError.getType(); if(messageDeliveryErrorType == MessageDeliveryErrorType.CONFIG_ERROR) { // Create User Exception Record Record record = recordService.newRecord("exception_message__sys"); // Below are 5 required fields record.setValue("name__v", "Spark Integration Configuration Error"); record.setValue("error_message__sys", "Integration Config error"); record.setValue("integration__sys", "V1P000000001001"); record.setValue("integration_point__sys", "V1Q000000001001"); record.setValue("error_type__sys", VaultCollections.asList("item_processing_error__sys")); recordService.batchSaveRecords(VaultCollections.asList(record)).rollbackOnErrors().execute(); } else if (messageDeliveryErrorType == MessageDeliveryErrorType.CONNECTION_FAILED) { // Notify Sysadmins GroupService groupService = ServiceLocator.locate(GroupService.class); List groupNames = VaultCollections.newList(); groupNames.add("system_administrators__v"); GetGroupsResponse groupResponse = groupService.getGroupsByNames(groupNames); Group group = groupResponse.getGroupByName("system_administrators__v"); List groups = VaultCollections.newList(); groups.add(group); //NotificationService NotificationService notificationService = ServiceLocator.locate(NotificationService.class); //NotificationParameters NotificationParameters notificationParameters = notificationService.newNotificationParameters(); notificationParameters.setRecipientsByGroups(groups); //NotificationTemplate NotificationTemplate template = notificationService.newNotificationTemplate() .setTemplateName("spark_integration_error_notification_template__c") .setTokenValue("errorType", "Connection Error"); //Send Notification with a NotificationTemplate notificationService.send(notificationParameters, template); } } }
Copyright © Veeva Systems 2017–2022. All rights reserved.