Skip to content
This repository was archived by the owner on Jan 17, 2025. It is now read-only.

Support new openwhisk notification for action completion #21

Merged
merged 1 commit into from
Nov 1, 2017
Merged

Support new openwhisk notification for action completion #21

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 11 additions & 8 deletions conductor.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -117,7 +117,10 @@ const main = (() => {
const session = params.$sessionId || process.env.__OW_ACTIVATION_ID

// initialize openwhisk instance
if (!wsk) wsk = openwhisk({ ignore_certs: true })
if (!wsk) {
wsk = openwhisk({ ignore_certs: true })
if (!notify) wsk.actions.qs_options.invoke = ['blocking', 'notify', 'cause']
}

// redis keys
const apiKey = process.env.__OW_API_KEY.substring(0, process.env.__OW_API_KEY.indexOf(':'))
Expand Down Expand Up @@ -298,14 +301,14 @@ const main = (() => {
break
case 'Task':
if (typeof json.Action === 'string') { // invoke user action
const invocation = notify ? { name: json.Action, params, blocking: true } : { name: json.Action, params, notify: process.env.__OW_ACTION_NAME, cause: session }
return persist(fsm, state, stack)
.then(() => wsk.actions.invoke({ name: json.Action, params, blocking: notify })
.catch(error => error.error && error.error.response ? error.error : badRequest(`Failed to invoke action ${json.Action}: ${encodeError(error).error}`)) // catch error reponses
.then(activation => db.rpushxAsync(sessionTraceKey, activation.activationId)
.then(() => activation.response ? activation : new Promise(resolve => poll(activation.activationId, resolve))) // poll if timeout
.then(activation => {
if (notify) return wsk.actions.invoke({ name: process.env.__OW_ACTION_NAME, params: { $activationId: activation.activationId, $sessionId: session, $result: activation.response.result } })
}).then(() => blocking ? getSessionResult() : { $session: session })))
.then(() => wsk.actions.invoke(invocation)
.catch(error => error.error && error.error.response ? error.error : badRequest(`Failed to invoke action ${json.Action}: ${encodeError(error).error}`))) // catch error reponses
.then(activation => db.rpushxAsync(sessionTraceKey, activation.activationId)
.then(() => activation.response || !notify ? activation : new Promise(resolve => poll(activation.activationId, resolve)))) // poll if timeout
.then(activation => notify && wsk.actions.invoke({ name: process.env.__OW_ACTION_NAME, params: { $activationId: activation.activationId, $sessionId: session, $result: activation.response.result } }))
.then(() => blocking ? getSessionResult() : { $session: session })
} else if (typeof json.Value !== 'undefined') { // value
params = JSON.parse(JSON.stringify(json.Value))
inspect()
Expand Down