@cumulus/message

Utilities for building and parsing Cumulus workflow messages.

Usage

  npm install @cumulus/message

API

Modules

Functions

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 | Message.CumulusMessage | 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 | Message.CumulusMessage | 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 |

isCumulusMessageLike()

Bare check for CumulusMessage Shape

Kind: global function

isDLQRecordLike()

Bare check for SQS message Shape

Kind: global function

unwrapDeadLetterCumulusMessage()

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

Kind: global function

extractSQSMetadata(message) ⇒

peel out metadata from an SQS(/DLQ)record

Kind: global function
Returns: the given message without its body

| Param | Description | | --- | --- | | message | DLQ or SQS message |

hoistCumulusMessageDetails()

Reformat object with key attributes at top level.

Kind: global function

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 | failedStepEvent | Step Function's failed event. | | failedStepEvent.id | | number (long), Step Functions failed event id. |

lastFailedEventStep(events) ⇒ Array.<HistoryEvent> | undefined

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

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

| Param | Type | Description | | --- | --- | --- | | events | Array.<HistoryEvent> | 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