@cumulus/message

Utilities for building and parsing Cumulus workflow messages.

Usage

  npm install @cumulus/message

API

Modules

Build: Utility functions for building Cumulus messages
Executions: Utility functions for generating execution information or parsing execution information from a Cumulus message
Granules: Utility functions for parsing granule information from a Cumulus message
Queue: Utility functions for parsing queue information from a Cumulus message

Functions

getMessageAsyncOperationId(message) ⇒ undefined | string ⏏

Get the async operation ID from a workflow message, if any.

constructCollectionId(name, version) ⇒ string ⏏

Returns the collection ID.

deconstructCollectionId(collectionId) ⇒ Object ⏏

Returns the name and version of a collection based on the collectionId used in elasticsearch indexing

getCollectionNameAndVersionFromMessage(message) ⇒ CollectionInfo | undefined ⏏

Get collection name and version from execution message.

getCollectionIdFromMessage(message) ⇒ string | undefined ⏏

Get collection ID from execution message.

unwrapDeadLetterCumulusMessage(messageBody) ⇒ Object

Unwrap dead letter Cumulus message, which may be wrapped in a States cloudwatch event, which is wrapped in an SQS message.

getMessagePdr(message) ⇒ undefined | Object ⏏

Get the PDR object from a workflow message, if any.

messageHasPdr(message) ⇒ boolean ⏏

Determine if message has a PDR.

getMessagePdrPANSent(message) ⇒ boolean ⏏

Get the PAN sent status from a workflow message, if any.

getMessagePdrPANMessage(message) ⇒ string ⏏

Get the PAN message status from a workflow message, if any.

getMessagePdrName(message) ⇒ string ⏏

Get the PDR name from a workflow message, if any.

getMessagePdrRunningExecutions(message) ⇒ number ⏏

Get the number of running executions for a PDR, if any.

getMessagePdrCompletedExecutions(message) ⇒ number ⏏

Get the number of completed executions for a PDR, if any.

getMessagePdrFailedExecutions(message) ⇒ number ⏏

Get the number of failed executions for a PDR, if any.

getMessagePdrStats(message) ⇒ PdrStats ⏏

Get the PDR stats from a workflow message, if any.

getPdrPercentCompletion(stats) ⇒ number ⏏

Get the percent completion of PDR executions

generatePdrApiRecordFromMessage(message, [updatedAt]) ⇒ ApiPdr ⏏

Generate a PDR record for the API from the message.

getMessageProvider(message) ⇒ MessageProvider | string ⏏

Get the provider from a workflow message, if any.

getMessageProviderId(message) ⇒ undefined | string ⏏

Get the provider ID from a workflow message, if any.

pullStepFunctionEvent(event) ⇒ Promise.<Object> ⏏

Given a Step Function event, replace specified key in event with contents of S3 remote message

parseStepMessage(stepMessage, stepName) ⇒ Promise.<Object> ⏏

Parse step message with CMA keys and replace specified key in event with contents of S3 remote message

getFailedStepName(events, failedStepEvent) ⇒ string

Searches the Execution step History for the TaskStateEntered pertaining to the failed task Id. HistoryEvent ids are numbered sequentially, starting at one.

lastFailedEventStep(events) ⇒ HistoryEventList | undefined

Finds all failed execution events and returns the last one in the list.

getFailedExecutionMessage(inputCumulusMessage, getExecutionHistoryFunction) ⇒ Object

Get message to use for publishing failed execution notifications.

Try to get the input to the last failed step in the execution so we can update the status of any granules/PDRs that don't exist in the initial execution input.

Falls back to overall execution input.

isFileExtensionMatched(granuleFile, extension) ⇒ boolean

Check if the file has the extension

parseException(exception) ⇒ string

Ensures that the exception is returned as an object

getMetaStatus(message) ⇒ Message.WorkflowStatus | undefined ⏏

Get the status of a workflow message, if any.

getMessageWorkflowTasks(message) ⇒ Object | undefined ⏏

Get the workflow tasks in a workflow message, if any.

getMessageWorkflowStartTime(message) ⇒ number | undefined ⏏

Get the workflow start time, if any.

getMessageWorkflowStopTime(message) ⇒ number | undefined ⏏

Get the workflow stop time, if any.

getMessageWorkflowName(message) ⇒ string | undefined ⏏

Get the workflow name, if any.

getWorkflowDuration(startTime, [stopTime]) ⇒ number ⏏

Get the workflow duration.

Build

Utility functions for building Cumulus messages

Example

const Build = require('@cumulus/message/Build');

buildQueueMessageFromTemplate(params) ⇒ `Message.CumulusMessage` ⏏

Build an SQS message from a workflow template for queueing executions.

Kind: Exported function
Returns: Message.CumulusMessage - A Cumulus message object

Param	Type	Description
params	`Object`
params.provider	`Object`	A provider object
params.collection	`Object`	A collection object
params.parentExecutionArn	`string`	ARN for parent execution
params.messageTemplate	`Object`	Message template for the workflow
params.payload	`Object`	Payload for the workflow
params.workflow	`Object`	workflow name & arn object
[params.asyncOperationId]	`string`	Async operation ID
[params.customCumulusMeta]	`Object`	Custom data for message.cumulus_meta
[params.customMeta]	`Object`	Custom data for message.meta
[params.executionNamePrefix]	`string`	Prefix to apply to the name of the enqueued execution

Executions

Utility functions for generating execution information or parsing execution information from a Cumulus message

Example

const Executions = require('@cumulus/message/Executions');

Executions
- buildExecutionArn(stateMachineArn, executionName) ⇒ string ⏏
- getExecutionUrlFromArn(executionArn) ⇒ string ⏏
- getStateMachineArnFromExecutionArn(executionArn) ⇒ string ⏏
- getMessageExecutionName(message) ⇒ string ⏏
- getMessageStateMachineArn(message) ⇒ string ⏏
- getMessageExecutionArn(message) ⇒ null | string ⏏
- getMessageExecutionParentArn(message) ⇒ undefined | string ⏏
- getMessageCumulusVersion(message) ⇒ undefined | string ⏏
- getMessageExecutionOriginalPayload(message) ⇒ unknown | undefined ⏏
- getMessageExecutionFinalPayload(message) ⇒ unknown | undefined ⏏
- generateExecutionApiRecordFromMessage(message, [updatedAt]) ⇒ ApiExecution ⏏
- global
  - generatePdrApiRecordFromMessage(message, [updatedAt]) ⇒ ApiPdr ⏏

buildExecutionArn(stateMachineArn, executionName) ⇒ `string` ⏏

Build execution ARN from a state machine ARN and execution name

Kind: Exported function
Returns: string - an execution ARN

Param	Type	Description
stateMachineArn	`string`	state machine ARN
executionName	`string`	state machine's execution name

getExecutionUrlFromArn(executionArn) ⇒ `string` ⏏

Returns execution URL from an execution ARN.

Kind: Exported function
Returns: string - returns AWS console URL for the execution

Param	Type	Description
executionArn	`string`	an execution ARN

getStateMachineArnFromExecutionArn(executionArn) ⇒ `string` ⏏

Get state machine ARN from an execution ARN

Kind: Exported function
Returns: string - a state machine ARN

Param	Type	Description
executionArn	`string`	an execution ARN

getMessageExecutionName(message) ⇒ `string` ⏏

Get the execution name from a workflow message.

Kind: Exported function
Returns: string - An execution name
Throws:

Error if there is no execution name

Param	Type	Description
message	`Message.CumulusMessage`	A workflow message object

getMessageStateMachineArn(message) ⇒ `string` ⏏

Get the state machine ARN from a workflow message.

Kind: Exported function
Returns: string - A state machine ARN
Throws:

Error if there is not state machine ARN

Param	Type	Description
message	`Message.CumulusMessage`	A workflow message object

getMessageExecutionArn(message) ⇒ `null` | `string` ⏏

Get the execution ARN from a workflow message.

Kind: Exported function
Returns: null | string - A state machine execution ARN

Param	Type	Description
message	`Message.CumulusMessage`	A workflow message object

getMessageExecutionParentArn(message) ⇒ `undefined` | `string` ⏏

Get the parent execution ARN from a workflow message, if any.

Kind: Exported function
Returns: undefined | string - A state machine execution ARN

Param	Type	Description
message	`Message.CumulusMessage`	A workflow message object

getMessageCumulusVersion(message) ⇒ `undefined` | `string` ⏏

Get the Cumulus version from a workflow message, if any.

Kind: Exported function
Returns: undefined | string - The cumulus version

Param	Type	Description
message	`Message.CumulusMessage`	A workflow message object

getMessageExecutionOriginalPayload(message) ⇒ `unknown` | `undefined` ⏏

Get the workflow original payload, if any.

Kind: Exported function
Returns: unknown | undefined - The workflow original payload

Param	Type	Description
message	`MessageWithPayload`	A workflow message object

getMessageExecutionFinalPayload(message) ⇒ `unknown` | `undefined` ⏏

Get the workflow final payload, if any.

Kind: Exported function
Returns: unknown | undefined - The workflow final payload

Param	Type	Description
message	`MessageWithPayload`	A workflow message object

generateExecutionApiRecordFromMessage(message, [updatedAt]) ⇒ `ApiExecution` ⏏

Generate an execution record for the API from the message.

Kind: Exported function
Returns: ApiExecution - An execution API record

Param	Type	Description
message	`MessageWithPayload`	A workflow message object
[updatedAt]	`string`	Optional updated timestamp to apply to record

generatePdrApiRecordFromMessage(message, [updatedAt]) ⇒ `ApiPdr` ⏏

Generate a PDR record for the API from the message.

Kind: global method of Executions
Returns: ApiPdr - An PDR API record

Param	Type	Description
message	`MessageWithOptionalPayloadPdr`	A workflow message object
[updatedAt]	`string`	Optional updated timestamp to apply to record

Granules

Utility functions for parsing granule information from a Cumulus message

Example

const Granules = require('@cumulus/message/Granules');

Granules
- getMessageGranules(message) ⇒ Array.<Object> | undefined ⏏
- messageHasGranules(message) ⇒ boolean ⏏
- getGranuleStatus(workflowStatus, granule) ⇒ string ⏏
- getGranuleQueryFields(message) ⇒ unknown | undefined ⏏
- generateGranuleApiRecord(message) ⇒ Promise.<ApiGranule> ⏏

getMessageGranules(message) ⇒ `Array.<Object>` | `undefined` ⏏

Get granules from payload?.granules of a workflow message.

Kind: Exported function
Returns: Array.<Object> | undefined - An array of granule objects, or undefined if message.payload.granules is not set

Param	Type	Description
message	`MessageWithGranules`	A workflow message

messageHasGranules(message) ⇒ `boolean` ⏏

Determine if message has a granules object.

Kind: Exported function
Returns: boolean - true if message has a granules object

Param	Type	Description
message	`MessageWithOptionalGranules`	A workflow message object

getGranuleStatus(workflowStatus, granule) ⇒ `string` ⏏

Determine the status of a granule.

Kind: Exported function
Returns: string - The granule status

Param	Type	Description
workflowStatus	`string`	The workflow status
granule	`MessageGranule`	A granule record conforming to the 'api' schema

getGranuleQueryFields(message) ⇒ `unknown` | `undefined` ⏏

Get the query fields of a granule, if any

Kind: Exported function
Returns: unknown | undefined - The granule query fields, if any

Param	Type	Description
message	`MessageWithGranules`	A workflow message

generateGranuleApiRecord(message) ⇒ `Promise.<ApiGranule>` ⏏

Generate an API granule record

Kind: Exported function
Returns: Promise.<ApiGranule> - The granule API record

Param	Type	Description
message	`MessageWithGranules`	A workflow message

Queue

Utility functions for parsing queue information from a Cumulus message

Example

const Queue = require('@cumulus/message/Queue');

Queue
- getQueueUrl(message) ⇒ string ⏏
- getMaximumExecutions(message, queueUrl) ⇒ number ⏏
- hasQueueAndExecutionLimit(message) ⇒ boolean ⏏

getQueueUrl(message) ⇒ `string` ⏏

Get the queue URL from a workflow message.

Kind: Exported function
Returns: string - A queue URL

Param	Type	Description
message	`MessageWithQueueInfo`	A workflow message object

getMaximumExecutions(message, queueUrl) ⇒ `number` ⏏

Get the maximum executions for a queue.

Kind: Exported function
Returns: number - Count of the maximum executions for the queue
Throws:

Error if no maximum executions can be found

Param	Type	Description
message	`Message.CumulusMessage`	A workflow message object
queueUrl	`string`	A queue URL

hasQueueAndExecutionLimit(message) ⇒ `boolean` ⏏

Determine if there is a queue and queue execution limit in the message.

Kind: Exported function
Returns: boolean - True if there is a queue and execution limit.

Param	Type	Description
message	`MessageWithQueueInfo`	A workflow message object

unwrapDeadLetterCumulusMessage(messageBody) ⇒ `Object`

Unwrap dead letter Cumulus message, which may be wrapped in a States cloudwatch event, which is wrapped in an SQS message.

Kind: global function
Returns: Object - the cumulus message or nearest available object

Param	Type	Description
messageBody	`Object`	received SQS message

getFailedStepName(events, failedStepEvent) ⇒ `string`

Searches the Execution step History for the TaskStateEntered pertaining to the failed task Id. HistoryEvent ids are numbered sequentially, starting at one.

Kind: global function
Returns: string - name of the current stepfunction task or 'UnknownFailedStepName'.

Param	Type	Description
events	`Array.<HistoryEvent>`	Step Function events array
failedStepEvent	`HistoryEvent`	Step Function's failed event.

lastFailedEventStep(events) ⇒ `HistoryEventList` | `undefined`

Finds all failed execution events and returns the last one in the list.

Kind: global function
Returns: HistoryEventList | undefined - - the last lambda or activity that failed in the event array, or an empty array.

Param	Type	Description
events	`Array.<HistoryEventList>`	array of AWS Stepfunction execution HistoryEvents

getFailedExecutionMessage(inputCumulusMessage, getExecutionHistoryFunction) ⇒ `Object`

Get message to use for publishing failed execution notifications.

Try to get the input to the last failed step in the execution so we can update the status of any granules/PDRs that don't exist in the initial execution input.

Falls back to overall execution input.

Kind: global function
Returns: Object - - CumulusMessage Execution step message or execution input message

Param	Type	Description
inputCumulusMessage	`Object`	Workflow execution input message
getExecutionHistoryFunction	`function`	Testing override for mock/etc of StepFunctions.getExecutionHistory

isFileExtensionMatched(granuleFile, extension) ⇒ `boolean`

Check if the file has the extension

Kind: global function
Returns: boolean - whether the file has the extension

Param	Type	Description
granuleFile	`ApiFile`	Granule file
extension	`string`	File extension to check

parseException(exception) ⇒ `string`

Ensures that the exception is returned as an object

Kind: global function
Returns: string - an stringified exception

Param	Type	Description
exception	`Object` \| `undefined`	the exception

About Cumulus

Cumulus is a cloud-based data ingest, archive, distribution and management prototype for NASA's future Earth science data streams.

Cumulus Documentation

Contributing

To make a contribution, please see our contributing guidelines.

Generated automatically using npm run build-docs

@cumulus/message

@cumulus/message

Usage

API

Modules

Functions

Build

buildQueueMessageFromTemplate(params) ⇒ Message.CumulusMessage ⏏

Executions

buildExecutionArn(stateMachineArn, executionName) ⇒ string ⏏

getExecutionUrlFromArn(executionArn) ⇒ string ⏏

getStateMachineArnFromExecutionArn(executionArn) ⇒ string ⏏

getMessageExecutionName(message) ⇒ string ⏏

getMessageStateMachineArn(message) ⇒ string ⏏

getMessageExecutionArn(message) ⇒ null | string ⏏

getMessageExecutionParentArn(message) ⇒ undefined | string ⏏

getMessageCumulusVersion(message) ⇒ undefined | string ⏏

getMessageExecutionOriginalPayload(message) ⇒ unknown | undefined ⏏

getMessageExecutionFinalPayload(message) ⇒ unknown | undefined ⏏

generateExecutionApiRecordFromMessage(message, [updatedAt]) ⇒ ApiExecution ⏏

generatePdrApiRecordFromMessage(message, [updatedAt]) ⇒ ApiPdr ⏏

Granules

getMessageGranules(message) ⇒ Array.<Object> | undefined ⏏

messageHasGranules(message) ⇒ boolean ⏏

getGranuleStatus(workflowStatus, granule) ⇒ string ⏏

getGranuleQueryFields(message) ⇒ unknown | undefined ⏏

generateGranuleApiRecord(message) ⇒ Promise.<ApiGranule> ⏏

Queue

getQueueUrl(message) ⇒ string ⏏

getMaximumExecutions(message, queueUrl) ⇒ number ⏏

hasQueueAndExecutionLimit(message) ⇒ boolean ⏏

unwrapDeadLetterCumulusMessage(messageBody) ⇒ Object

getFailedStepName(events, failedStepEvent) ⇒ string

lastFailedEventStep(events) ⇒ HistoryEventList | undefined

getFailedExecutionMessage(inputCumulusMessage, getExecutionHistoryFunction) ⇒ Object

isFileExtensionMatched(granuleFile, extension) ⇒ boolean

parseException(exception) ⇒ string

About Cumulus

Contributing

Readme

Keywords

Package Sidebar

Install

Repository

Homepage

DownloadsWeekly Downloads

Version

License

Unpacked Size

Total Files

Last publish

Collaborators

buildQueueMessageFromTemplate(params) ⇒ `Message.CumulusMessage` ⏏

buildExecutionArn(stateMachineArn, executionName) ⇒ `string` ⏏

getExecutionUrlFromArn(executionArn) ⇒ `string` ⏏

getStateMachineArnFromExecutionArn(executionArn) ⇒ `string` ⏏

getMessageExecutionName(message) ⇒ `string` ⏏

getMessageStateMachineArn(message) ⇒ `string` ⏏

getMessageExecutionArn(message) ⇒ `null` | `string` ⏏

getMessageExecutionParentArn(message) ⇒ `undefined` | `string` ⏏

getMessageCumulusVersion(message) ⇒ `undefined` | `string` ⏏

getMessageExecutionOriginalPayload(message) ⇒ `unknown` | `undefined` ⏏

getMessageExecutionFinalPayload(message) ⇒ `unknown` | `undefined` ⏏

generateExecutionApiRecordFromMessage(message, [updatedAt]) ⇒ `ApiExecution` ⏏

generatePdrApiRecordFromMessage(message, [updatedAt]) ⇒ `ApiPdr` ⏏

getMessageGranules(message) ⇒ `Array.<Object>` | `undefined` ⏏

messageHasGranules(message) ⇒ `boolean` ⏏

getGranuleStatus(workflowStatus, granule) ⇒ `string` ⏏

getGranuleQueryFields(message) ⇒ `unknown` | `undefined` ⏏

generateGranuleApiRecord(message) ⇒ `Promise.<ApiGranule>` ⏏

getQueueUrl(message) ⇒ `string` ⏏

getMaximumExecutions(message, queueUrl) ⇒ `number` ⏏

hasQueueAndExecutionLimit(message) ⇒ `boolean` ⏏

unwrapDeadLetterCumulusMessage(messageBody) ⇒ `Object`

getFailedStepName(events, failedStepEvent) ⇒ `string`

lastFailedEventStep(events) ⇒ `HistoryEventList` | `undefined`

getFailedExecutionMessage(inputCumulusMessage, getExecutionHistoryFunction) ⇒ `Object`

isFileExtensionMatched(granuleFile, extension) ⇒ `boolean`

parseException(exception) ⇒ `string`

Weekly Downloads