TMConsole.BrokerTask

3.0.2

Classes/Public/TMBroker.ps1

#region Classes

class TMBroker : System.IDisposable {

    #region Non-Static Properties

    # The settings that define how the broker will operate
    [TMBrokerSetting]$Settings = [TMBrokerSetting]::new()

    # The Task in TM that created this class instance
    [TMTask]$Task = [TMTask]::new()

    # The Task in TM that holds the Action for initializing data for the broker
    [TMBrokerSubject]$Init

    # The borrowed TMSession that the broker will use as its connection to TM.
    # New-TMBroker assigns this reference; constructing a placeholder TMSession here
    # would create token-refresh resources that are orphaned when it is overwritten.
    [TMSession]$TMSession

    # Data about the event and its Tasks
    [TMBrokerEventData]$EventData = [TMBrokerEventData]::new()

    # Object holding various values to indicate the status of the broker
    [TMBrokerStatus]$Status = [TMBrokerStatus]::new()

    # A cache that can be passed between Actions that are invoked by the broker
    [Object]$Cache

    # A list of the Tasks that the broker will automate
    [Collections.Generic.List[TMBrokerSubject]]$Subjects = [Collections.Generic.List[TMBrokerSubject]]::new()
    
    # A Handle to the TMConsole Runtime ActionRequest Queue to use for Parallel brokering
    [Object]$TMConsoleActionRequestQueue

    # Tracks whether this broker has already released its runtime references
    hidden [Boolean]$Disposed = $false

    #endregion Non-Static Properties

    #region Constructors

    TMBroker() {}

    TMBroker([TMBrokerExecutionMode]$mode, [String]$taskProperty, [String[]]$matchingCriteria) {
        $this.Settings = [TMBrokerSetting]::new($mode, $taskProperty, $matchingCriteria)
    }

    TMBroker([TMBrokerExecutionMode]$mode, [ScriptBlock]$matchExpression) {
        $this.Settings = [TMBrokerSetting]::new($mode, $matchExpression)
    }

    TMBroker([TMBrokerExecutionMode]$mode, [TMBrokerTaskFilter]$taskFilter) {
        $this.Settings = [TMBrokerSetting]::new($mode, $taskFilter)
    }

    TMBroker(
        [TMBrokerExecutionMode]$mode,
        [String]$taskProperty,
        [String[]]$matchingCriteria,
        [Int32]$timeout,
        [Int32]$pauseSeconds
    ) {
        $this.Settings = [TMBrokerSetting]::new($mode, $taskProperty, $matchingCriteria, $timeout, $pauseSeconds)
        $this.Status = [TMBrokerStatus]::new($timeout)
    }

    TMBroker(
        [TMBrokerExecutionMode]$mode,
        [ScriptBlock]$matchExpression,
        [Int32]$timeout,
        [Int32]$pauseSeconds
    ) {
        $this.Settings = [TMBrokerSetting]::new($mode, $matchExpression, $timeout, $pauseSeconds)
        $this.Status = [TMBrokerStatus]::new($timeout)
    }

    TMBroker(
        [TMBrokerExecutionMode]$mode,
        [TMBrokerTaskFilter]$taskFilter,
        [Int32]$timeout,
        [Int32]$pauseSeconds
    ) {
        $this.Settings = [TMBrokerSetting]::new($mode, $taskFilter, $timeout, $pauseSeconds)
        $this.Status = [TMBrokerStatus]::new($timeout)
    }

    TMBroker(
        [TMBrokerExecutionMode]$mode,
        [String]$taskProperty,
        [String[]]$matchingCriteria,
        [Int32]$timeout,
        [Int32]$pauseSeconds,
        [Boolean]$parallel,
        [Int32]$throttle,
        [Int32]$taskLimitBeforeRefresh,
        [Int32]$taskCacheExpirationSeconds,
        [Boolean]$showProgress
    ) {
        $this.Settings = [TMBrokerSetting]::new($mode, $taskProperty, $matchingCriteria, $timeout, $pauseSeconds, $parallel, $throttle, $taskLimitBeforeRefresh, $taskCacheExpirationSeconds, $showProgress)
        $this.Status = [TMBrokerStatus]::new($timeout, $throttle)
    }
        
    TMBroker(
        [TMBrokerExecutionMode]$mode,
        [ScriptBlock]$matchExpression,
        [Int32]$timeout,
        [Int32]$pauseSeconds,
        [Boolean]$parallel,
        [Int32]$throttle,
        [Int32]$taskLimitBeforeRefresh,
        [Int32]$taskCacheExpirationSeconds,
        [Boolean]$showProgress
    ) {
        $this.Settings = [TMBrokerSetting]::new($mode, $matchExpression, $timeout, $pauseSeconds, $parallel, $throttle, $taskLimitBeforeRefresh, $taskCacheExpirationSeconds, $showProgress)
        $this.Status = [TMBrokerStatus]::new($timeout, $throttle)
    }

    TMBroker(
        [TMBrokerExecutionMode]$mode,
        [TMBrokerTaskFilter]$taskFilter,
        [Int32]$timeout,
        [Int32]$pauseSeconds,
        [Boolean]$parallel,
        [Int32]$throttle,
        [Int32]$taskLimitBeforeRefresh,
        [Int32]$taskCacheExpirationSeconds,
        [Boolean]$showProgress
    ) {
        $this.Settings = [TMBrokerSetting]::new($mode, $taskFilter, $timeout, $pauseSeconds, $parallel, $throttle, $taskLimitBeforeRefresh, $taskCacheExpirationSeconds, $showProgress)
        $this.Status = [TMBrokerStatus]::new($timeout, $throttle)
    }

    #endregion Constructors

    #region Non-Static Methods

    <#
        Summary:
            Loads the EventData property using this object's TMSession
        Params:
            None
        Outputs:
            None
    #>
    [void]GetEventData() {
        if (-not $this.TMSession) {
            [TMBrokerOutput]::Throw('A TM Session is required to invoke this method')
        }

        if ($this.Settings.SubjectScope.FilterType -eq [TMBrokerSubjectScopeFilterType]::TaskFilter) {
            [TMBrokerOutput]::Verbose('Getting Event data with Task filter')
            $this.EventData.GetEventData(
                $this.TMSession.UserContext.Project.Id,
                ($this.EventData.Event.Name ?? $this.TMSession.UserContext.Event.Name),
                $this.TMSession.Name,
                $this.Settings.SubjectScope.TaskFilter
            )
        }
        else {
            [TMBrokerOutput]::Verbose('Getting Event data')
            $this.EventData.GetEventData(
                $this.TMSession.UserContext.Project.Id,
                ($this.EventData.Event.Name ?? $this.TMSession.UserContext.Event.Name),
                $this.TMSession.Name
            )
        }
    }

    <#
        Summary:
            Loads all of the broker-related tasks
        Params:
            TaskId - The Id of broker task
        Outputs:
            None
    #>
    [void]GetTaskData($TaskId) {
        if (-not $this.EventData) {
            [TMBrokerOutput]::Throw('Event data must be loaded before invoking this method')
        }

        # Store this Broker Task's data
        [TMBrokerOutput]::Verbose("Loading Broker Task data for Task Id: $TaskId")
        $this.Task = ($this.EventData.Tasks | Where-Object { $_.Id -eq $TaskId })

        if (-not $this.Task) {
            $this.Task = Get-TMTask -Id $TaskId -TMSession $this.TMSession.Name
        }

        # Determine if there is an init cache Task
        $this.GetInitTask()

        # Get all of the subject tasks
        $this.GetSubjectTasks()
    }

    <#
        Summary:
            Gets the init task, if present, from the Event's task data
        Params:
            None
        Outputs:
            None
    #>
    [void]GetInitTask() {
        if ($this.Settings.ExecutionMode -eq [TMBrokerExecutionMode]::Inline) {
            if (-not $this.Task) {
                [TMBrokerOutput]::Throw('Broker Task data must be loaded before invoking this method')
            }
            $InitTask = (
                $this.EventData.Tasks |
                Where-Object { $_.Id -in $this.Task.Successors.TaskId } |
                Where-Object -FilterScript $this.Settings.SubjectScope.MatchExpression
            )
            if ($InitTask) {
                [TMBrokerOutput]::Verbose('Creating Init Task object')
                $this.Init = [TMBrokerSubject]::new($InitTask)
            }
        }
    }

    <#
        Summary:
            Gets all of the subject tasks that will be managed by the broker from the Event's task data
        Params:
            None
        Outputs:
            None
    #>
    [void]GetSubjectTasks() {
        switch ($this.Settings.ExecutionMode) {
            ([TMBrokerExecutionMode]::Inline) {
                [TMBrokerOutput]::Verbose("Gathering Inline Subject Tasks")

                if (-not $this.Task -and -not $this.Init) {
                    [TMBrokerOutput]::Throw('Broker Task and Init Task data must be loaded before invoking this method')
                }

                # Initialize the subjects list
                $this.Subjects = [Collections.Generic.List[Collections.Generic.List[TMBrokerSubject]]]::new()

                foreach ($TaskId in ($this.Init.Task.Successors.TaskId ?? $this.Task.Successors.TaskId)) {

                    # Initialize a list to hold all of the subject tasks for a specific asset
                    $Workflow = [Collections.Generic.List[TMBrokerSubject]]::new()

                    # Find the first/direct successor subject task
                    $SubjectTask = $this.EventData.Tasks | Where-Object { $_.Id -eq $TaskId }

                    $i = 0
                    while ($SubjectTask) {
                        $i++

                        # Add the subject task data to the workflow
                        $Workflow.Add([TMBrokerSubject]::new($SubjectTask, $i))

                        # Look for the next subject task in the workflow
                        $SubjectTask = $this.EventData.Tasks | Where-Object { $_.Id -eq $SubjectTask.Successors.TaskId }
                    }

                    # Record how many tasks are in each asset's workflow
                    $this.Status.WorkflowTaskCount = $i

                    # Add this workflow to the list of subjects
                    $this.Subjects.Add($Workflow)
                }
            }

            ([TMBrokerExecutionMode]::Service) {
                [TMBrokerOutput]::Verbose("Gathering Service Subject Tasks")

                # Initialize the subjects list
                $this.Subjects = [Collections.Generic.List[TMBrokerSubject]]::new()

                # Filter all Tasks down to the specified scope
                $ServiceSubjectTasks = $this.EventData.Tasks | Where-Object {
                    ($_.id -ne $Broker.task.id ) -and
                    ($_.Action.Id -ne 0) -and
                    ($_.Action.name -notlike '*broker*') -and
                    -not ($_.Action.MethodParams | Where-Object { $_.ParamName -match 'get_' }) -and
                    -not ($_.Action.MethodParams | Where-Object { $_.prompt -and $_.prompt.type.ToLower() -ne 'blank' })
                }

                # Apply the match expression to filter tasks further
                if ($this.Settings.SubjectScope.FilterType -eq 'MatchExpression') {
                    $ServiceSubjectTasks = $ServiceSubjectTasks | Where-Object -FilterScript $this.Settings.SubjectScope.MatchExpression
                }

                # Add the filtered Tasks to the list of subject tasks
                foreach ($Task in $ServiceSubjectTasks) {
                    $this.Subjects.Add([TMBrokerSubject]::new($Task))
                }
            }

            default { }
        }

        # Subjects retain the TMTask instances that they need. The complete Event task
        # collection is construction-time data and is no longer needed after this point.
        $this.EventData.Tasks = @()
    }

    <#
        Summary:
            Invokes the Init Task's Action to fill the cache
        Params:
            None
        Outputs:
            None
    #>
    [void]PopulateCache() {
        if (-not $this.Init) {
            [TMBrokerOutput]::Throw('Init Task data must be loaded before invoking this method')
        }

        $this.Init.Invoke($this.TMSession)
    }

    <#
        Summary:
            Updates each Task's status and Action settings using fresh data from TM
        Params:
            None
        Outputs:
            None
    #>
    [void]GetActiveSubjectRunspaceStatus() {
        # Gather all Task Ids
        $SubjectTaskIds = $this.Status.ActiveSubjects.ToArray()
        foreach ($SubjectTaskId in $SubjectTaskIds) {
            $TMTaskId = "TMTaskId_$SubjectTaskId"
            if ([TMCRunspaceDirectory]::ActiveRunspaces.ContainsKey($TMTaskId)) {
                switch ([TMCRunspaceDirectory]::ActiveRunspaces[$TMTaskId]) {
                    'Terminated' {
                        [void]($this.Status.ActiveSubjects.Remove($SubjectTaskId))
                    }
                    'Completed' {
                        [void]($this.Status.ActiveSubjects.Remove($SubjectTaskId))
                    }
                    'Running' {}
                    Default {}
                }
            }
        }
    }

    <#
        Summary:
            Updates each Task's status and Action settings using fresh data from TM
        Params:
            None
        Outputs:
            None
    #>
    [void]RefreshTaskData() {
        # Gather all Task Ids
        $TaskIds = [Array]@(
            $this.Subjects.Task.Id
            $this.Task.Id
            $this.Init.Task.Id
        ) | Where-Object { $_ -gt 0 }


        # Query TM for Task statuses and Action params
        [TMBrokerOutput]::Verbose("Requesting fresh Task data from TransitionManager")
        $Statement = "find Task by 'id' inList([$($TaskIds -join ', ')]) fetch 'id', 'status', 'lastUpdated', 'apiAction.methodParams'"
        $ServerTaskData = Invoke-TMQLStatement -TMSession $this.TMSession.Name -Statement $Statement
        [TMBrokerOutput]::Verbose("Received data for $($ServerTaskData.Count) Task(s)")
        
        ## Re-Record all of the info into a dicti
        $TaskData = @{}
        foreach ($Task in $ServerTaskData) {
            $TaskData["TMTaskId_$($Task.Id)"] = $Task
        }

        # Update the broker and the init Task statuses
        $this.Task.Status = $TaskData["TMTaskId_$($this.Task.Id)"].Status
        
        if ($this.Init.Task.Id) {
            $this.Init.Task.Status = $TaskData["TMTaskId_$($this.Init.Id)"].Status
        }

        switch ($this.Settings.ExecutionMode) {
            ([TMBrokerExecutionMode]::Inline) {
                [TMBrokerOutput]::Verbose("Updating Inline Task Data")

                # Update the status of each Task in each workflow
                foreach ($Workflow in $this.Subjects) {
                    foreach ($Subject in $Workflow) {
                        $Subject.Task.Status = $TaskData["TMTaskId_$($Subject.Task.Id)"].Status
                    }
                }
            }

            ([TMBrokerExecutionMode]::Service) {
                [TMBrokerOutput]::Verbose("Updating Service Task data")
                foreach ($Subject in $this.Subjects) {
                    $TaskFromTM = $TaskData["TMTaskId_$($Subject.Task.Id)"]
                    $Subject.Task.Status = $TaskFromTM.Status
                    $Subject.Task.LastUpdated = $TaskFromTM.LastUpdated

                    # Update the Subject's Action settings if needed
                    $Subject.UpdateActionSettings($TaskFromTM.'apiAction.methodParams')

                    # Review Task states to update throttling settings
                    if ($this.Settings.Parallel) {
                        # Handle updating Subject Task data based on the status of the task
                        switch ($Subject.Task.Status) {
                            'Started' {
                                # Mark the Action as Started so the Broker ignores it for next time
                                $Subject.Action.ExecutionStatus = 'Started'

                                # Check if a timeout was defined in the Action params
                                if ($Subject.Action.ShouldTimeout) {
                                    # Attempt to place this Subject Task on hold with a timeout comment
                                    try {
                                        [TMBrokerOutput]::Info("Resetting Task #: $($Subject.Task.TaskNumber). Task has run longer than: $($Subject.Action.Settings.Timeout.Minutes) minutes", 'DarkYellow')
                                        $Subject.Timeout($this.TMSession.Name)
                                    }
                                    catch {
                                        [TMBrokerOutput]::Warning("Could not timeout Task # $($Subject.Task.TaskNumber): $($_.Exception.Message)")
                                    }
                                }
                            }

                            'Completed' {
                                # Check to ensure the broker does not believe it's running completed Tasks
                                if ($this.Status.ActiveSubjects -contains $Subject.Task.Id) {
                                    $this.Status.ActiveSubjects.Remove($Subject.Task.Id)
                                }
                                $Subject.Action.ExecutionStatus = 'Successful'
                            }

                            'Hold' {
                                # If the Task's status was changed either manually or due to failure,
                                # reset the execution status so that it can be re-run
                                if ($this.Status.ActiveSubjects -contains $Subject.Task.Id) {
                                    $this.Status.ActiveSubjects.Remove($Subject.Task.Id)
                                }
                                $Subject.Action.ExecutionStatus = 'Failed'

                                # Check if a retry was defined in the Action params
                                if ($Subject.Action.ShouldRetry) {
                                    # Attempt to reset the Task Action and Reset the Task to Ready
                                    try {
                                        [TMBrokerOutput]::Info("Resetting Task #: $($Subject.Task.TaskNumber). Retries left: $($Subject.Action.Settings.Retry.RemainingRetries)", 'DarkYellow')
                                        $Subject.QueueRetry($this.TMSession.Name)
                                        $this.Status.LastWebRequest = Get-Date
                                    }
                                    catch {
                                        [TMBrokerOutput]::Warning("Could not retry Task # $($Subject.Task.TaskNumber): $($_.Exception.Message)")
                                    }
                                }
                            }

                            { $_ -in 'Pending', 'Ready' } {
                                # If the Task's status was changed either manually or due to failure,
                                # reset the execution status so that it can be re-run
                                if ($this.Status.ActiveSubjects -contains $Subject.Task.Id) {
                                    $this.Status.ActiveSubjects.Remove($Subject.Task.Id)
                                }
                                $Subject.Action.ExecutionStatus = 'Pending'
                            }
                        }
                    }
                }
            }
        }
        $this.EventData.CacheExpiration = (Get-Date).AddSeconds($This.Settings.Timing.TaskCacheExpirationSeconds)
    }

    <#
        Summary:
            Updates the TMBrokerProgress properties on this Status object to be used for tracking and progress bars
        Params:
            None
        Outputs:
            None
    #>
    [void]RefreshBrokerProgress() {
        [TMBrokerOutput]::Verbose("Updating Progress data")
        $this.Status.CompletedTasks.Value = ($this.Subjects | Where-Object { $_.Task.Status -eq 'Completed' -and $_.Action.ExecutionStatus -eq 'Successful' }).Count
        $this.Status.ElapsedMinutes.Value = [Math]::Ceiling($this.Settings.Timing.Timer.Elapsed.TotalMinutes)
        if ($this.Settings.Parallel) {
            $this.Status.Throttle.Value = $this.Status.ActiveSubjects.Count
        }
    }

    <#
        Summary:
            Sends a lightweight request to TM to keep the web services jsession alive
        Params:
            None
        Outputs:
            None
    #>
    [void]KeepAlive() {
        if (((Get-Date) - $this.Status.LastWebRequest).TotalMinutes -gt 10) {
            try {
                [TMBrokerOutput]::Verbose("Making Keep Alive request to TransitionManager")
                $WebRequestSplat = @{
                    Uri                  = "https://$($this.TMSession.TMServer)/tdstm/ws/progress/demo"
                    Method               = 'GET'
                    WebSession           = $this.TMSession.TMWebSession
                    SkipCertificateCheck = $this.TMSession.AllowInsecureSSL
                }
                $Response = Invoke-WebRequest @WebRequestSplat

                if ($Response.StatusCode -notin 200, 204) {
                    throw "The status code $($Response.StatusCode) does not indicate success"
                }
                [TMBrokerOutput]::Verbose("Response Status Code: $($Response.StatusCode)")
            }
            catch {
                [TMBrokerOutput]::Warning("Keep alive request failed: $($_.Exception.Message)")
            }

            $this.Status.LastWebRequest = Get-Date
        }
    }

    <#
        Method: Run
        Description: Executes the scoped subject Tasks
        Parameters: None
    #>
    [void]Run() {
        try {
            [TMBrokerOutput]::Debug("Execution Mode: $($this.Settings.ExecutionMode)")
            [TMBrokerOutput]::Debug("Execution Order: $($this.Settings.ExecutionOrder)")
            [TMBrokerOutput]::Debug("Execution Sort Order: $($this.Settings.ExecutionSortOrder)")
            [TMBrokerOutput]::Debug("Parallel: $($this.Settings.Parallel)")
            [TMBrokerOutput]::Debug("Throttle: $($this.Settings.Throttle)")
            [TMBrokerOutput]::Debug("Timeout Minutes: $($this.Settings.Timing.TimeoutMinutes)")
            [TMBrokerOutput]::Debug("Pause Seconds: $($this.Settings.Timing.PauseSeconds)")
            [TMBrokerOutput]::Debug("Task Invocation Limit Before Refresh: $($this.Settings.TaskLimitBeforeRefresh)")
            [TMBrokerOutput]::Verbose("Starting Broker execution")
            
            # Initialize values
            $this.Status.CompletedTasks.MaxValue = $this.Subjects.Count
            $this.Settings.Timing.Timer.Start()
            $this.RefreshTaskData()
            $Refresh = $False
            $Sleep = $False

            $BrokerRunning = $true
            while ( $BrokerRunning ) {
                
                ## Timeout check
                if ($this.Settings.Timing.Timer.Elapsed.TotalMinutes -ge $this.Settings.Timing.TimeoutMinutes) {
                    Write-Host "The broker has run for it's allowed maximum time: $($this.Settings.Timing.TimeoutMinutes) minutes"
                    [TMBrokerOutput]::Throw("The broker has run for it's allowed maximum time: $($this.Settings.Timing.TimeoutMinutes) minutes")
                }

                ## Had a busy day. Take a little nap to refresh
                if ($Sleep) {
                    [TMBrokerOutput]::Verbose("Pausing for $($this.Settings.Timing.PauseSeconds) second(s)")
                    Start-Sleep -Seconds $this.Settings.Timing.PauseSeconds
                    $Sleep = $False
                }

                # Perform a refresh of the task data from TransitionManager if necessary
                if ($Refresh) {
                    $this.Status.TasksExecutedSinceRefresh = 0
                    $Refresh = $False
                    
                    $this.RefreshTaskData()
                    
                    # Saftey check the broker task status in TM, exit if the task status is not Started
                    if ($this.Task.Status -ne 'Started') {
                        [TMBrokerOutput]::Throw('The status of the Broker Task has changed outside of TMConsole')
                    }
                }

                # If needed, make a small request to TM to keep the session alive
                $this.KeepAlive()
                
                # If needed, make a small request to TM to keep the session alive
                $this.GetActiveSubjectRunspaceStatus()

                # Refresh the progress properties to be output to the TMC UI
                if ($this.Settings.ShowProgress) {

                    $this.RefreshBrokerProgress()
                    # Update the TMC UI's progress bars
                    $ProgressSplat = @{
                        Id              = 1
                        ParentId        = 0
                        Activity        = 'Subject Tasks'
                        Status          = "$($this.Status.CompletedTasks.Value) of $($this.Status.CompletedTasks.MaxValue) tasks completed"
                        PercentComplete = $this.Status.CompletedTasks.PercentComplete
                    }
                    Write-Progress @ProgressSplat

                    $ProgressSplat = @{
                        Id              = 2
                        ParentId        = 0
                        Activity        = 'Timeout'
                        Status          = "$([Math]::Ceiling($this.Settings.Timing.TimeoutMinutes - $this.Settings.Timing.Timer.Elapsed.TotalMinutes)) minutes left"
                        PercentComplete = $this.Status.ElapsedMinutes.PercentComplete
                    }
                    Write-Progress @ProgressSplat

                    if ($this.Settings.Parallel) {
                        $ProgressSplat = @{
                            Id              = 3
                            ParentId        = 0
                            Activity        = 'Throttle'
                            Status          = "$($this.Status.ActiveSubjects.Count) of $($this.Settings.Throttle)"
                            PercentComplete = $this.Status.Throttle.PercentComplete
                        }
                        Write-Progress @ProgressSplat
                    }
                }

                # Execute Subject Tasks
                switch ($this.Settings.ExecutionMode) {

                    # Inline Brokers run a workflow step worth of tasks at once
                    ([TMBrokerExecutionMode]::Inline) {
                        foreach ($Workflow in $this.Subjects) {
                            $NextWorkflowSubject = $Workflow |
                            Where-Object { $_.Task.Status -ne 'Completed' -and $_.Action.ExecutionStatus -eq 'Pending' -and $_.Task.Id -notin $this.Status.ActiveSubjects } |
                            Sort-Object Order | Select-Object -First 1

                            if ($NextWorkflowSubject) {
                                if ($this.Settings.Parallel) {
                                    if ($this.Status.ActiveSubjects.Count -lt $this.Settings.Throttle) {
                                        # Record the Task ID as belonging to this broker for Throttling
                                        $this.Status.ActiveSubjects.Add($NextWorkflowSubject.Task.Id)

                                        # Invoke the next workflow Subject
                                        $NextWorkflowSubject.InvokeParallel($this.TMSession, $this.Cache)
                                    }
                                }
                                else {
                                    $NextWorkflowSubject.Invoke($this.TMSession, $this.Cache)
                                }
                                $this.Status.TasksExecutedSinceRefresh++
                                $this.Status.LastWebRequest = Get-Date
                            }
                        }
                    }

                    # Service Brokers run one task at a time, when they become ready
                    ([TMBrokerExecutionMode]::Service) {
                        
                        # Get the most preferred actionable subject
                        $PreferredActionableSubject = $this.Subjects |
                        Where-Object { $_.Task.Status -eq 'Ready' -and $_.Action.ExecutionStatus -eq 'Pending' } |
                        Sort-Object { $_.Task."$($this.Settings.ExecutionOrder)" } -Descending:$($this.Settings.ExecutionSortOrder -eq 'Descending') |
                        Select-Object -First 1

                        # Invoke the Most Preferred, Actionable Subject
                        if ($PreferredActionableSubject) {
                            
                            # Run a Subject in a normal invocation runspace, but track that task so it 'consumes' one runspace
                            if ($this.Settings.Parallel) {
                                # Honor Throttling settings
                                if ($this.Status.ActiveSubjects.Count -lt $this.Settings.Throttle) {
                                    
                                    # Update the local cache so this task won't run again until another refresh from TM
                                    $PreferredActionableSubject.Task.Status = 'Started'
                                    
                                    # Record the Task ID as belonging to this broker for Throttling
                                    $this.Status.ActiveSubjects.Add($PreferredActionableSubject.Task.Id)
                                    
                                    # Invoke the Subject
                                    $PreferredActionableSubject.GetActionRequest($this.TMSession, $this.Cache)
                                    
                                    ## This should always be true, but in development it's useful in testing
                                    if ($This.TMConsoleActionRequestQueue) {
                                        
                                        ## Parallel Subject tasks might start too quickly and cause a file/credential deadlock.
                                        ## Inject a short sleep to prevent this.
                                        $PreferredActionableSubject.ActionRequest.options.apiAction.script = "Start-Sleep -Milliseconds (Get-Random -Minimum 50 -Maximum 250); `n" + $PreferredActionableSubject.ActionRequest.options.apiAction.script
                                        $This.TMConsoleActionRequestQueue.Enqueue($PreferredActionableSubject.ActionRequest)
                                        $PreferredActionableSubject.ActionRequest = $null
                                    }
                                    $this.Status.TasksExecutedSinceRefresh++
                                }
                            }
                            else {
                                # Update the local cache so this task won't run again until another refresh from TM
                                $PreferredActionableSubject.Task.Status = 'Started'
                                
                                # Invoke this ActionRequest directly, in this runspace
                                $PreferredActionableSubject.Invoke($this.TMSession, $this.Cache)
                                $this.Status.TasksExecutedSinceRefresh++
                            }
                        }
                    }
                }

                # Check for Sleep and Refresh requirements
                if ($this.Status.TasksExecutedSinceRefresh -eq $this.Settings.TaskLimitBeforeRefresh) {
                    $Refresh = $true
                }
                if ($this.Settings.Parallel) {
                    if ($this.Status.ActiveSubjects.Count -eq $this.Settings.Throttle) {
                        $Refresh = $true
                    }
                    if ($this.Subjects.Task.Status -notcontains 'Ready') {
                        $Refresh = $true
                    }
                }
                else {
                    if ($this.Subjects.Task.Status -notcontains 'Ready') {
                        $Sleep = $true
                        $Refresh = $true
                    }
                }

                ## Check for Exit conditions
                $RemainingTaskCount = 0
                $This.Subjects.Task.Status | Group-Object | ForEach-Object {
                    if ($_.Name -in @('Ready', 'Pending', 'Hold')) {
                        $RemainingTaskCount += $_.Count
                    }
                }
                
                if ($RemainingTaskCount -eq 0) {
                    Write-Host "The broker has run to completion, there are no remaining Subject tasks to invoke."
                    $BrokerRunning = $false
                }
            }
        }
        finally {
            $this.Dispose()
        }
    }

    <#
        Method: Dispose
        Description: Releases task, ActionRequest, event, cache, session, and queue
                        references owned or retained by this broker. The TMSession is
                        borrowed from $Global:TMSessions and is not disposed here.
        Parameters: None
    #>
    [void]Dispose() {
        if ($this.Disposed) {
            return
        }

        try {
            if ($null -ne $this.Settings -and $null -ne $this.Settings.Timing -and $null -ne $this.Settings.Timing.Timer) {
                $this.Settings.Timing.Timer.Stop()
            }
        }
        catch { }

        try {
            if ($null -ne $this.Init) {
                $this.Init.ActionRequest = $null
                $this.Init.Task = $null
                $this.Init.Action = $null
                $this.Init.Asset = $null
            }

            if ($null -ne $this.Subjects) {
                foreach ($Subject in $this.Subjects) {
                    if ($null -ne $Subject) {
                        $Subject.ActionRequest = $null
                        $Subject.Task = $null
                        $Subject.Action = $null
                        $Subject.Asset = $null
                    }
                }
                $this.Subjects.Clear()
            }
        }
        catch { }

        try {
            if ($null -ne $this.EventData) {
                $this.EventData.Tasks = @()
                $this.EventData.Event = $null
            }
        }
        catch { }

        try {
            if ($null -ne $this.Status -and $null -ne $this.Status.ActiveSubjects) {
                $this.Status.ActiveSubjects.Clear()
            }
        }
        catch { }

        $this.Task = $null
        $this.Init = $null
        $this.EventData = $null
        $this.Cache = $null
        $this.TMConsoleActionRequestQueue = $null
        $this.TMSession = $null
        $this.Disposed = $true
    }

    #endregion Non-Static Methods
}


class TMBrokerEventData {

    #region Non-Static Properties

    [TMEvent]$Event
    [TMTask[]]$Tasks = [System.Collections.Generic.List[TMTask]]::new()
    [datetime]$CacheExpiration
    #endregion Non-Static Properties

    #region Constructors

    TMBrokerEventData() {}

    TMBrokerEventData([Int32]$projectId, [String]$eventName, [String]$tmSession) {
        $this.GetEventData($projectId, $eventName, $tmSession)
    }

    [void]GetEventData([Int32]$projectId, [String]$eventName, [String]$tmSession) {

        if (-not $this.Event) {
            # Get the Event object
            $this.Event = Get-TMEvent -TMSession $tmSession -ProjectId $projectId -Name $eventName
        }

        # Get all of the broker-related Tasks in the Event
        $this.Tasks = Get-TMTask -TMSession $tmSession -ProjectId $projectId -EventName $this.Event.name
    }

    [void]GetEventData([Int32]$projectId, [String]$eventName, [String]$tmSession, [TMBrokerTaskFilter]$Filter) {

        if (-not $this.Event) {
            # Get the Event object
            $this.Event = Get-TMEvent -TMSession $tmSession -ProjectId $projectId -Name $eventName
        }

        # Get all of the broker-related Tasks in the Event
        $TaskSplat = $Filter.ToHashTable()
        $this.Tasks = Get-TMTask -TMSession $tmSession -ProjectId $projectId -EventName $this.Event.name @TaskSplat
    }

    #endregion Constructors

}


class TMBrokerSubjectScope {

    #region Non-Static Properties

    # If a Task filter or match expression is not defined, this is the Task property that will be evaluated with MatchingCriteria
    [TMBrokerSubjectScopeTaskProperty]$TaskProperty = [TMBrokerSubjectScopeTaskProperty]::Title

    # If a Task filter or match expression is not defined, this list of values will be matched against TaskProperty
    [String[]]$MatchingCriteria

    # A ScriptBlock containing an expression that will be used as the -FilterScript parameter on Where-Object
    # after all of the Event Tasks have been retrieved from TM
    [ScriptBlock]$MatchExpression

    # The type of filtering that will be used to determine the Broker's Subject Tasks
    [TMBrokerSubjectScopeFilterType]$FilterType = [TMBrokerSubjectScopeFilterType]::TaskFilter

    # The TaskFilter object that will be used as a splat with Get-TMTask
    [TMBrokerTaskFilter]$TaskFilter = [TMBrokerTaskFilter]::new()

    #endregion Non-Static Properties

    #region Constructors

    TMBrokerSubjectScope() {
        $this.TaskFilter.Title.Add('\[Subject\]')
    }

    TMBrokerSubjectScope([TMBrokerSubjectScopeTaskProperty]$taskProperty, [String[]]$matchingCriteria) {
        $this.TaskProperty = $taskProperty
        $this.MatchingCriteria = $matchingCriteria
        $matchingCriteria | ForEach-Object {
            $this.TaskFilter."$taskProperty".Add($_)
        }
    }

    TMBrokerSubjectScope([ScriptBlock]$matchExpression) {
        $this.MatchExpression = $matchExpression
        $this.FilterType = [TMBrokerSubjectScopeFilterType]::MatchExpression
    }

    TMBrokerSubjectScope([TMBrokerTaskFilter]$taskFilter) {
        $this.TaskFilter = $taskFilter
    }

    #endregion Constructors

    #region Non-Static Methods

    <#
        Summary:
            Sets the MatchExpression value to a ScriptBlock that can be used with Where-Object
        Params:
            None
        Outputs:
            None
    #>
    hidden [void]GetMatchExpression() {
        $this.MatchExpression = [ScriptBlock]::Create("`$_.$($this.TaskProperty) -match '$([TMBrokerSubjectScope]::GetMatchString($this.MatchingCriteria))'")
    }

    #endregion Non-Static Methods

    #region Static Methods

    <#
        Summary:
            Converts a list of values to a regular expression that can be used with -match
        Params:
            Criteria - The list of values to be converted to a match string
        Outputs:
            A String formatted as a regular expression
    #>
    static [String]GetMatchString([String[]]$Criteria) {
        return ('(' + ($Criteria -join ')|(') + ')')
    }

    #endregion Static Methods

}


class TMBrokerSetting {

    #region Non-Static Properties

    # The values that define how the Broker decides which Tasks it will manage and execute
    [TMBrokerSubjectScope]$SubjectScope = [TMBrokerSubjectScope]::new()

    # The mode that defines the way in which the Broker will execute its Subject Tasks
    [TMBrokerExecutionMode]$ExecutionMode = [TMBrokerExecutionMode]::Service

    # Values that define the Broker's timeout and refresh intervals
    [TMBrokerTiming]$Timing = [TMBrokerTiming]::new()

    # When determining the next Subject Task to invoke, they will be sorted by this Task property
    [TMBrokerSettingExecutionOrder]$ExecutionOrder = [TMBrokerSettingExecutionOrder]::Score

    # When determining the next Subject Task to invoke, they will be sorted in this order
    [ValidateSet('Ascending', 'Descending')]
    [String]$ExecutionSortOrder = 'Descending'

    # Will the Subject Tasks be executed in parallel or one at a time?
    [Boolean]$Parallel = $true
    
    # The maximum number of Subject Tasks that can be runninng in parallel
    [Int32]$Throttle = 8
    
    # The maximum number of Subject Tasks allowed to be invoked before a task refresh
    [Int32]$TaskLimitBeforeRefresh = 4
    
    # Will the Broker use Progress bars to display it's progress
    [Boolean]$ShowProgress = $false
    
    #endregion Non-Static Properties

    #region Constructors

    TMBrokerSetting() {}

    TMBrokerSetting(
        [TMBrokerExecutionMode]$mode,
        [String]$taskProperty,
        [String[]]$matchingCriteria,
        [Int32]$timeout,
        [Int32]$pauseSeconds
    ) {
        $this.ExecutionMode = $mode
        $this.SubjectScope = [TMBrokerSubjectScope]::new($taskProperty, $matchingCriteria)
        $this.Timing = [TMBrokerTiming]::new($timeout, $pauseSeconds)
    }

    TMBrokerSetting(
        [TMBrokerExecutionMode]$mode,
        [ScriptBlock]$matchExpression,
        [Int32]$timeout,
        [Int32]$pauseSeconds
    ) {
        $this.ExecutionMode = $mode
        $this.SubjectScope = [TMBrokerSubjectScope]::new($matchExpression)
        $this.Timing = [TMBrokerTiming]::new($timeout, $pauseSeconds)
    }

    TMBrokerSetting(
        [TMBrokerExecutionMode]$mode,
        [TMBrokerTaskFilter]$taskFilter,
        [Int32]$timeout,
        [Int32]$pauseSeconds
    ) {
        $this.ExecutionMode = $mode
        $this.SubjectScope = [TMBrokerSubjectScope]::new($taskFilter)
        $this.Timing = [TMBrokerTiming]::new($timeout, $pauseSeconds)
    }

    TMBrokerSetting(
        [TMBrokerExecutionMode]$mode,
        [String]$taskProperty,
        [String[]]$matchingCriteria,
        [Int32]$timeout,
        [Int32]$pauseSeconds,
        [Boolean]$parallel,
        [Int32]$throttle,
        [Int32]$taskLimitBeforeRefresh,
        [Int32]$taskCacheExpirationSeconds,
        [Boolean]$showProgress
    ) {
        $this.ExecutionMode = $mode
        $this.SubjectScope = [TMBrokerSubjectScope]::new($taskProperty, $matchingCriteria)
        $this.Timing = [TMBrokerTiming]::new($timeout, $pauseSeconds, $taskCacheExpirationSeconds)
        $this.Parallel = $parallel
        $this.Throttle = $throttle
        $this.TaskLimitBeforeRefresh = $taskLimitBeforeRefresh
        $this.ShowProgress = $showProgress
    }

    TMBrokerSetting(
        [TMBrokerExecutionMode]$mode,
        [ScriptBlock]$matchExpression,
        [Int32]$timeout,
        [Int32]$pauseSeconds,
        [Boolean]$parallel,
        [Int32]$throttle,
        [Int32]$taskLimitBeforeRefresh,
        [Int32]$taskCacheExpirationSeconds,
        [Boolean]$showProgress
    ) {
        $this.ExecutionMode = $mode
        $this.SubjectScope = [TMBrokerSubjectScope]::new($matchExpression)
        $this.Timing = [TMBrokerTiming]::new($timeout, $pauseSeconds, $taskCacheExpirationSeconds)
        $this.Parallel = $parallel
        $this.Throttle = $throttle
        $this.TaskLimitBeforeRefresh = $taskLimitBeforeRefresh
        $this.ShowProgress = $showProgress
    }

    TMBrokerSetting(
        [TMBrokerExecutionMode]$mode,
        [TMBrokerTaskFilter]$taskFilter,
        [Int32]$timeout,
        [Int32]$pauseSeconds,
        [Boolean]$parallel,
        [Int32]$throttle,
        [Int32]$taskLimitBeforeRefresh,
        [Int32]$taskCacheExpirationSeconds,
        [Boolean]$showProgress
    ) {
        $this.ExecutionMode = $mode
        $this.SubjectScope = [TMBrokerSubjectScope]::new($taskFilter)
        $this.Timing = [TMBrokerTiming]::new($timeout, $pauseSeconds, $taskCacheExpirationSeconds)
        $this.Parallel = $parallel
        $this.Throttle = $throttle
        $this.TaskLimitBeforeRefresh = $taskLimitBeforeRefresh
        $this.ShowProgress = $showProgress
    }

    TMBrokerSetting([TMBrokerExecutionMode]$mode, [String]$taskProperty, [String[]]$matchingCriteria) {
        $this.ExecutionMode = $mode
        $this.SubjectScope = [TMBrokerSubjectScope]::new($taskProperty, $matchingCriteria)
    }

    TMBrokerSetting([TMBrokerExecutionMode]$mode, [ScriptBlock]$matchExpression) {
        $this.ExecutionMode = $mode
        $this.SubjectScope = [TMBrokerSubjectScope]::new($matchExpression)
    }

    TMBrokerSetting([TMBrokerExecutionMode]$mode, [TMBrokerTaskFilter]$taskFilter) {
        $this.ExecutionMode = $mode
        $this.SubjectScope = [TMBrokerSubjectScope]::new($taskFilter)
    }
    #endregion Constructors

}


class TMBrokerTiming {

    #region Non-Static Properties

    # How many minutes the Broker will run before ending its execution
    [Int64]$TimeoutMinutes = 120
    
    # If the Broker is idle, with no Tasks to invoke, the number of seconds to wait before
    # querying TM again for Task data
    [Int64]$PauseSeconds = 15

    # The maximum number of Seconds that an pass before a refresh of tasks is required
    [Int32]$TaskCacheExpirationSeconds = 30

    # The timer that represents the Broker's execution time
    [Diagnostics.Stopwatch]$Timer = [Diagnostics.Stopwatch]::new()

    #endregion Non-Static Properties

    #region Constructors

    TMBrokerTiming () {}

    TMBrokerTiming ([Int64]$timeoutMinutes, [Int64]$pauseSeconds) {
        $this.TimeoutMinutes = $timeoutMinutes
        $this.PauseSeconds = $pauseSeconds
    }
    TMBrokerTiming ([Int64]$timeoutMinutes, [Int64]$pauseSeconds, [Int32]$taskCacheExpirationSeconds) {
        $this.TimeoutMinutes = $timeoutMinutes
        $this.PauseSeconds = $pauseSeconds
        $this.TaskCacheExpirationSeconds = $taskCacheExpirationSeconds
    }
    #endregion Constructors

}


class TMBrokerStatus {

    #region Non-Static Properties

    # A list of the Subject Task Ids that are being executed in parallel
    [Collections.Generic.List[Int64]]$ActiveSubjects = [Collections.Generic.List[Int64]]::new()

    # The number of Subject Tasks that have been executed since fresh Task data has been received from TM
    [Int64]$TasksExecutedSinceRefresh = 0

    # The last date/time that a request was made to one of TM's web services endpoints.
    # Used to determine if a keep alive ping needs to be made
    [DateTime]$LastWebRequest = (Get-Date)

    # The number of Subject Tasks that have been completed successfully. Used for TMC progress bars
    [TMBrokerProgress]$CompletedTasks = [TMBrokerProgress]::new()

    # The number of minutes that have elapsed since the Broker was started. Used for TMC progress bars
    [TMBrokerProgress]$ElapsedMinutes = [TMBrokerProgress]::new()

    # The number of Subject Tasks thatare currently running. Used for TMC progress bars
    [TMBrokerProgress]$Throttle = [TMBrokerProgress]::new()

    #endregion Non-Static Properties

    #region Constructors

    TMBrokerStatus () {}

    TMBrokerStatus ([Int32]$timeoutMinutes) {
        $this.ElapsedMinutes = [TMBrokerProgress]::new($timeoutMinutes)
    }

    TMBrokerStatus ([Int32]$timeoutMinutes, [Int32]$throttle) {
        $this.ElapsedMinutes = [TMBrokerProgress]::new($timeoutMinutes)
        $this.Throttle = [TMBrokerProgress]::new($throttle)
    }

    #endregion Constructors

}


class TMBrokerProgress {

    #region Non-Static Properties

    [Int32]$Value = 0
    [Int32]$MaxValue = 1

    #endregion Non-Static Properties

    #region Constructors

    TMBrokerProgress() {
        $this.addPublicMembers()
    }

    TMBrokerProgress([Int32]$maxValue) {
        $this.addPublicMembers()
        $this.MaxValue = $maxValue
    }

    TMBrokerProgress([Int32]$currentValue, [Int32]$maxValue) {
        $this.addPublicMembers()
        $this.Value = $currentValue
        $this.MaxValue = $maxValue
    }

    #endregion Constructors

    #region Private Methods

    <#
        Summary:
            Adds members with calculated get and/or set methods
        Params:
            None
        Outputs:
            None
    #>
    hidden [void]addPublicMembers() {
        # public readonly Int32 PercentComplete
        $this.PSObject.Properties.Add(
            [PSScriptProperty]::new(
                'PercentComplete',
                { # get
                    return [Int32][Math]::Ceiling(($this.Value / $this.MaxValue) * 100)
                }
            )
        )
    }

    #endregion Private Methods

}


class TMBrokerTaskFilter {

    #region Non-Static Properties

    [Collections.Generic.List[Int32]]$TaskNumber = [Collections.Generic.List[Int32]]::new()
    [Collections.Generic.List[Int32]]$TaskSpecId = [Collections.Generic.List[Int32]]::new()
    [Collections.Generic.List[String]]$Status = [Collections.Generic.List[String]]::new()
    [Collections.Generic.List[String]]$AssetName = [Collections.Generic.List[String]]::new()
    [Collections.Generic.List[String]]$AssetType = [Collections.Generic.List[String]]::new()
    [Collections.Generic.List[String]]$AssetClass = [Collections.Generic.List[String]]::new()
    [Collections.Generic.List[String]]$ActionName = [Collections.Generic.List[String]]::new()
    [Collections.Generic.List[String]]$Category = [Collections.Generic.List[String]]::new()
    [Collections.Generic.List[String]]$Title = [Collections.Generic.List[String]]::new()
    [Collections.Generic.List[String]]$Team = [Collections.Generic.List[String]]::new()

    #endregion Non-Static Properties

    #region Constructors

    TMBrokerTaskFilter() {}

    #endregion Constructors

    #region Non-Static Methods

    [Hashtable]ToHashTable() {
        $returnHashtable = @{}

        if ($this.TaskNumber) { $returnHashtable.TaskNumber = $this.TaskNumber }
        if ($this.TaskSpecId) { $returnHashtable.TaskSpecId = $this.TaskSpecId }
        if ($this.Status) { $returnHashtable.Status = $this.Status }
        if ($this.AssetName) { $returnHashtable.AssetName = $this.AssetName }
        if ($this.AssetType) { $returnHashtable.AssetType = $this.AssetType }
        if ($this.AssetClass) { $returnHashtable.AssetClass = $this.AssetClass }
        if ($this.ActionName) { $returnHashtable.ActionName = $this.ActionName }
        if ($this.Category) { $returnHashtable.Category = $this.Category }
        if ($this.Title) { $returnHashtable.Title = $this.Title }
        if ($this.Team) { $returnHashtable.Team = $this.Team }

        return $returnHashtable
    }

    [String]ToString() {
        return "{$($this.ToHashTable().Keys -join ', ')}"
    }

    #endregion Non-Static Methods

}

#endregion Classes


#region Enumerations

enum TMBrokerExecutionMode {
    Service
    Inline
}


enum TMBrokerSettingExecutionOrder {
    TaskNumber
    Score
    TaskSpecId
}


enum TMBrokerSubjectScopeTaskProperty {
    ActionName
    AssetClass
    AssetName
    AssetType
    Category
    Status
    TaskNumber
    TaskSpecId
    Team
    Title
}


enum TMBrokerSubjectScopeFilterType {
    TaskFilter
    MatchExpression
}

#endregion Enumerations