import { IExecutionsCurrentSummaryExtended , IPushData , IPushDataConsoleMessage , IPushDataExecutionFinished , IPushDataExecutionStarted , IPushDataNodeExecuteAfter , IPushDataNodeExecuteBefore , IPushDataTestWebhook , } from '../../Interface' ; import { externalHooks } from '@/components/mixins/externalHooks' ; import { nodeHelpers } from '@/components/mixins/nodeHelpers' ; import { showMessage } from '@/components/mixins/showMessage' ; import { titleChange } from '@/components/mixins/titleChange' ; import { workflowHelpers } from '@/components/mixins/workflowHelpers' ; import mixins from 'vue-typed-mixins' ; import { WORKFLOW_SETTINGS_MODAL_KEY } from '@/constants' ; export const pushConnection = mixins ( externalHooks , nodeHelpers , showMessage , titleChange , workflowHelpers ) . extend ( { data ( ) { return { eventSource : null as EventSource | null , reconnectTimeout : null as NodeJS . Timeout | null , retryTimeout : null as NodeJS . Timeout | null , pushMessageQueue : [ ] as Array < { event : Event ; retriesLeft : number ; } > , } ; } , computed : { sessionId ( ) { return this . $store . getters . sessionId ; } , } , methods : { pushAutomaticReconnect ( ) { if ( this . reconnectTimeout !== null ) { return ; } this . reconnectTimeout = setTimeout ( ( ) => { this . pushConnect ( ) ; } , 3000 ) ; } , /**
         * Connect to server to receive data via EventSource
         */ pushConnect ( ) { this . pushDisconnect ( ) ; const connectionUrl = `${ this . $store . getters . getRestUrl }/push?sessionId=${ this . sessionId }` ; this . eventSource = new EventSource ( connectionUrl ) ; this . eventSource . addEventListener ( 'message' , this . pushMessageReceived , false ) ; this . eventSource . addEventListener ( 'open' , ( ) => { this . $store . commit ( 'setPushConnectionActive' , true ) ; if ( this . reconnectTimeout !== null ) { clearTimeout ( this . reconnectTimeout ) ; this . reconnectTimeout = null ; } } , false ) ; this . eventSource . addEventListener ( 'error' , ( ) => { this . pushDisconnect ( ) ; if ( this . reconnectTimeout !== null ) { clearTimeout ( this . reconnectTimeout ) ; this . reconnectTimeout = null ; } this . $store . commit ( 'setPushConnectionActive' , false ) ; this . pushAutomaticReconnect ( ) ; } , false ) ; } , /**
         * Close connection to server
         */ pushDisconnect ( ) { if ( this . eventSource !== null ) { this . eventSource . close ( ) ; this . eventSource = null ; this . $store . commit ( 'setPushConnectionActive' , false ) ; } } , param { Event } event param { number } retryAttempts returns queuePushMessage ( event , retryAttempts ) { this . pushMessageQueue . push ( { event , retriesLeft : retryAttempts } ) ; if ( this . retryTimeout === null ) { this . retryTimeout = setTimeout ( this . processWaitingPushMessages , 20 ) ; } } , /**
         * Process the push messages which are waiting in the queue
         */ processWaitingPushMessages ( ) { if ( this . retryTimeout !== null ) { clearTimeout ( this . retryTimeout ) ; this . retryTimeout = null ; } const queueLength = this . pushMessageQueue . length ; for ( let i = 0 ; i < queueLength ; i ++ ) { const messageData = this . pushMessageQueue . shift ( ) ; if ( this . pushMessageReceived ( messageData ! . event , true ) === false ) { messageData ! . retriesLeft -= 1 ; if ( messageData ! . retriesLeft > 0 ) { this . pushMessageQueue . unshift ( messageData ! ) ; } break ; } } if ( this . pushMessageQueue . length !== 0 && this . retryTimeout === null ) { this . retryTimeout = setTimeout ( this . processWaitingPushMessages , 25 ) ; } } , param { Event } event param { boolean } isRetry returns { boolean } pushMessageReceived ( event , isRetry ? ) { const retryAttempts = 5 ; let receivedData ; try { receivedData = JSON . parse ( event . data ) ; } catch ( error ) { return false ; } if ( receivedData . type === 'sendConsoleMessage' ) { const pushData = receivedData . data ; console . log ( pushData . source , ... pushData . messages ) ; return true ; } if ( ! [ 'testWebhookReceived' ] . includes ( receivedData . type ) && isRetry !== true && this . pushMessageQueue . length ) { this . queuePushMessage ( event , retryAttempts ) ; return false ; } if ( receivedData . type === 'nodeExecuteAfter' || receivedData . type === 'nodeExecuteBefore' ) { if ( this . $store . getters . isActionActive ( 'workflowRunning' ) === false ) { return false ; } const pushData = receivedData . data ; if ( this . $store . getters . activeExecutionId !== pushData . executionId ) { if ( isRetry !== true ) { this . queuePushMessage ( event , retryAttempts ) ; } return false ; } } if ( receivedData . type === 'executionFinished' ) { const pushData = receivedData . data ; this . $store . commit ( 'finishActiveExecution' , pushData ) ; if ( this . $store . getters . isActionActive ( 'workflowRunning' ) === false ) { return false ; } if ( this . $store . getters . activeExecutionId !== pushData . executionId ) { if ( isRetry !== true ) { this . queuePushMessage ( event , retryAttempts ) ; } return false ; } const runDataExecuted = pushData . data ; const runDataExecutedErrorMessage = this . $getExecutionError ( runDataExecuted . data . resultData . error ) ; const workflow = this . getWorkflow ( ) ; if ( runDataExecuted . waitTill !== undefined ) { const { activeExecutionId , workflowSettings , saveManualExecutions , } = this . $store . getters ; const isSavingExecutions = workflowSettings . saveManualExecutions === undefined ? saveManualExecutions : workflowSettings . saveManualExecutions ; let action ; if ( ! isSavingExecutions ) { action = '<a class="open-settings">Turn on saving manual executions</a> and run again to see what happened after this node.' ; } else { action = `<a href="/execution/${ activeExecutionId }" target="_blank">View the execution</a> to see what happened after this node.` ; } this . $titleSet ( workflow . name as string , 'IDLE' ) ; this . $showToast ( { title : 'Workflow started waiting' , message : `${ action } <a href="https://docs.n8n.io/nodes/n8n-nodes-base.wait/" target="_blank">More info</a>` , type : 'success' , duration : 0 , onLinkClick : async ( e ) => { if ( e . classList . contains ( 'open-settings' ) ) { if ( this . $store . getters . isNewWorkflow ) { await this . saveAsNewWorkflow ( ) ; } this . $store . dispatch ( 'ui/openModal' , WORKFLOW_SETTINGS_MODAL_KEY ) ; } } , } ) ; } else if ( runDataExecuted . finished !== true ) { this . $titleSet ( workflow . name as string , 'ERROR' ) ; this . $showMessage ( { title : 'Problem executing workflow' , message : runDataExecutedErrorMessage , type : 'error' , } ) ; } else { this . $titleSet ( workflow . name as string , 'IDLE' ) ; this . $showMessage ( { title : this . $locale . baseText ( 'pushConnection.showMessage.title' ) , message : this . $locale . baseText ( 'pushConnection.showMessage.message' ) , type : 'success' , } ) ; } runDataExecuted . data . resultData . runData = this . $store . getters . getWorkflowRunData ; this . $store . commit ( 'setExecutingNode' , null ) ; this . $store . commit ( 'setWorkflowExecutionData' , runDataExecuted ) ; this . $store . commit ( 'removeActiveAction' , 'workflowRunning' ) ; this . updateNodesExecutionIssues ( ) ; let itemsCount = 0 ; if ( runDataExecuted . data . resultData . lastNodeExecuted && ! runDataExecutedErrorMessage ) { itemsCount = runDataExecuted . data . resultData . runData [ runDataExecuted . data . resultData . lastNodeExecuted ] [ 0 ] . data ! . main [ 0 ] ! . length ; } this . $externalHooks ( ) . run ( 'pushConnection.executionFinished' , { itemsCount , nodeName : runDataExecuted . data . resultData . lastNodeExecuted , errorMessage : runDataExecutedErrorMessage , runDataExecutedStartData : runDataExecuted . data . startData , resultDataError : runDataExecuted . data . resultData . error , } ) ; } else if ( receivedData . type === 'executionStarted' ) { const pushData = receivedData . data ; const executionData = { id : pushData . executionId , finished : false , mode : pushData . mode , startedAt : pushData . startedAt , retryOf : pushData . retryOf , workflowId : pushData . workflowId , workflowName : pushData . workflowName , } ; this . $store . commit ( 'addActiveExecution' , executionData ) ; } else if ( receivedData . type === 'nodeExecuteAfter' ) { const pushData = receivedData . data ; this . $store . commit ( 'addNodeExecutionData' , pushData ) ; } else if ( receivedData . type === 'nodeExecuteBefore' ) { const pushData = receivedData . data ; this . $store . commit ( 'setExecutingNode' , pushData . nodeName ) ; } else if ( receivedData . type === 'testWebhookDeleted' ) { const pushData = receivedData . data ; if ( pushData . workflowId === this . $store . getters . workflowId ) { this . $store . commit ( 'setExecutionWaitingForWebhook' , false ) ; this . $store . commit ( 'removeActiveAction' , 'workflowRunning' ) ; } } else if ( receivedData . type === 'testWebhookReceived' ) { const pushData = receivedData . data ; if ( pushData . workflowId === this . $store . getters . workflowId ) { this . $store . commit ( 'setExecutionWaitingForWebhook' , false ) ; this . $store . commit ( 'setActiveExecutionId' , pushData . executionId ) ; } this . processWaitingPushMessages ( ) ; } return true ; } , } , } ) ;