MegaBites-AI/Windows-powershell
0372
1// Copyright (c) Microsoft Corporation.2// Licensed under the MIT License.3 4namespace System.Management.Automation.Runspaces5{6 using System;7 using System.Collections;8 using System.Collections.Generic;9 using System.Collections.ObjectModel;10 using System.Threading;11 using Dbg = System.Management.Automation.Diagnostics;12 using System.Management.Automation.Internal;13 14#pragma warning disable 1634, 1691 // Stops compiler from warning about unknown warnings15 16 /// <summary>17 /// This class has common base implementation for Pipeline class.18 /// LocalPipeline and RemotePipeline classes derives from it.19 /// </summary>20 internal abstract class PipelineBase : Pipeline21 {22 #region constructors23 24 /// <summary>25 /// Create a pipeline initialized with a command string.26 /// </summary>27 /// <param name="runspace">The associated Runspace/></param>28 /// <param name="command">Command string.</param>29 /// <param name="addToHistory">If true, add pipeline to history.</param>30 /// <param name="isNested">True for nested pipeline.</param>31 /// <exception cref="ArgumentNullException">32 /// Command is null and add to history is true33 /// </exception>34 protected PipelineBase(Runspace runspace, string command, bool addToHistory, bool isNested)35 : base(runspace)36 {37 Initialize(runspace, command, addToHistory, isNested);38 39 // Initialize streams40 InputStream = new ObjectStream();41 OutputStream = new ObjectStream();42 ErrorStream = new ObjectStream();43 }44 45 /// <summary>46 /// Create a Pipeline with an existing command string.47 /// Caller should validate all the parameters.48 /// </summary>49 /// <param name="runspace">50 /// The LocalRunspace to associate with this pipeline.51 /// </param>52 /// <param name="command">53 /// The command to invoke.54 /// </param>55 /// <param name="addToHistory">56 /// If true, add the command to history.57 /// </param>58 /// <param name="isNested">59 /// If true, mark this pipeline as a nested pipeline.60 /// </param>61 /// <param name="inputStream">62 /// Stream to use for reading input objects.63 /// </param>64 /// <param name="errorStream">65 /// Stream to use for writing error objects.66 /// </param>67 /// <param name="outputStream">68 /// Stream to use for writing output objects.69 /// </param>70 /// <param name="infoBuffers">71 /// Buffers used to write progress, verbose, debug, warning, information72 /// information of an invocation.73 /// </param>74 /// <exception cref="ArgumentNullException">75 /// Command is null and add to history is true76 /// </exception>77 /// <exception cref="ArgumentNullException">78 /// 1. InformationalBuffers is null79 /// </exception>80 protected PipelineBase(Runspace runspace,81 CommandCollection command,82 bool addToHistory,83 bool isNested,84 ObjectStreamBase inputStream,85 ObjectStreamBase outputStream,86 ObjectStreamBase errorStream,87 PSInformationalBuffers infoBuffers)88 : base(runspace, command)89 {90 Dbg.Assert(inputStream != null, "Caller Should validate inputstream parameter");91 Dbg.Assert(outputStream != null, "Caller Should validate outputStream parameter");92 Dbg.Assert(errorStream != null, "Caller Should validate errorStream parameter");93 Dbg.Assert(infoBuffers != null, "Caller Should validate informationalBuffers parameter");94 Dbg.Assert(command != null, "Command cannot be null");95 96 // Since we are constructing this pipeline using a commandcollection we dont need97 // to add cmd to CommandCollection again (Initialize does this).. because of this98 // I am handling history here..99 Initialize(runspace, null, false, isNested);100 if (addToHistory)101 {102 // get command text for history..103 string cmdText = command.GetCommandStringForHistory();104 HistoryString = cmdText;105 AddToHistory = addToHistory;106 }107 108 // Initialize streams109 InputStream = inputStream;110 OutputStream = outputStream;111 ErrorStream = errorStream;112 InformationalBuffers = infoBuffers;113 }114 115 /// <summary>116 /// Copy constructor to support cloning.117 /// </summary>118 /// <param name="pipeline">The source pipeline.</param>119 /// <remarks>120 /// The copy constructor's intent is to support the scenario121 /// where a host needs to run the same set of commands multiple122 /// times. This is accomplished via creating a master pipeline123 /// then cloning it and executing the cloned copy.124 /// </remarks>125 protected PipelineBase(PipelineBase pipeline)126 : this(pipeline.Runspace, null, false, pipeline.IsNested)127 {128 // NTRAID#Windows Out Of Band Releases-915851-2005/09/13129 if (pipeline == null)130 {131 throw PSTraceSource.NewArgumentNullException(nameof(pipeline));132 }133 134 if (pipeline._disposed)135 {136 throw PSTraceSource.NewObjectDisposedException("pipeline");137 }138 139 AddToHistory = pipeline.AddToHistory;140 HistoryString = pipeline.HistoryString;141 foreach (Command command in pipeline.Commands)142 {143 Command clone = command.Clone();144 145 // Attach the cloned Command to this pipeline.146 Commands.Add(clone);147 }148 }149 150 #endregion constructors151 152 #region properties153 154 private Runspace _runspace;155 156 /// <summary>157 /// Access the runspace this pipeline is created on.158 /// </summary>159 public override Runspace Runspace160 {161 get162 {163 return _runspace;164 }165 }166 167 /// <summary>168 /// This internal method doesn't do the _disposed check.169 /// </summary>170 /// <returns></returns>171 internal Runspace GetRunspace()172 {173 return _runspace;174 }175 176 private bool _isNested;177 178 /// <summary>179 /// Is this pipeline nested.180 /// </summary>181 public override bool IsNested182 {183 get184 {185 return _isNested;186 }187 }188 189 /// <summary>190 /// Is this a pulse pipeline (created by the EventManager)191 /// </summary>192 internal bool IsPulsePipeline { get; set; }193 194 private PipelineStateInfo _pipelineStateInfo = new PipelineStateInfo(PipelineState.NotStarted);195 196 /// <summary>197 /// Info about current state of the pipeline.198 /// </summary>199 /// <remarks>200 /// This value indicates the state of the pipeline after the change.201 /// </remarks>202 public override PipelineStateInfo PipelineStateInfo203 {204 get205 {206 lock (SyncRoot)207 {208 // Note:We do not return internal state.209 return _pipelineStateInfo.Clone();210 }211 }212 }213 214 // 913921-2005/07/08 ObjectWriter can be retrieved on a closed stream215 /// <summary>216 /// Access the input writer for this pipeline.217 /// </summary>218 public override PipelineWriter Input219 {220 get221 {222 return InputStream.ObjectWriter;223 }224 }225 226 /// <summary>227 /// Access the output reader for this pipeline.228 /// </summary>229 public override PipelineReader<PSObject> Output230 {231 get232 {233 return OutputStream.PSObjectReader;234 }235 }236 237 /// <summary>238 /// Access the error output reader for this pipeline.239 /// </summary>240 /// <remarks>241 /// This is the non-terminating error stream from the command.242 /// In this release, the objects read from this PipelineReader243 /// are PSObjects wrapping ErrorRecords.244 /// </remarks>245 public override PipelineReader<object> Error246 {247 get248 {249 return _errorStream.ObjectReader;250 }251 }252 253 /// <summary>254 /// Is this pipeline a child pipeline?255 ///256 /// IsChild flag makes it possible for the pipeline to differentiate between257 /// a true v1 nested pipeline and the cmdlets calling cmdlets case. See bug258 /// 211462.259 /// </summary>260 internal override bool IsChild { get; set; }261 262 #endregion properties263 264 #region stop265 266 /// <summary>267 /// Synchronous call to stop the running pipeline.268 /// </summary>269 public override void Stop()270 {271 CoreStop(true);272 }273 274 /// <summary>275 /// Asynchronous call to stop the running pipeline.276 /// </summary>277 public override void StopAsync()278 {279 CoreStop(false);280 }281 282 /// <summary>283 /// Stop the running pipeline.284 /// </summary>285 /// <param name="syncCall">If true pipeline is stopped synchronously286 /// else asynchronously.</param>287 private void CoreStop(bool syncCall)288 {289 // Is pipeline already in stopping state.290 bool alreadyStopping = false;291 lock (SyncRoot)292 {293 switch (PipelineState)294 {295 case PipelineState.NotStarted:296 SetPipelineState(PipelineState.Stopping);297 SetPipelineState(PipelineState.Stopped);298 break;299 300 // If pipeline execution has failed or completed or301 // stopped, return silently.302 case PipelineState.Stopped:303 case PipelineState.Completed:304 case PipelineState.Failed:305 return;306 // If pipeline is in Stopping state, ignore the second307 // stop.308 case PipelineState.Stopping:309 alreadyStopping = true;310 break;311 312 case PipelineState.Running:313 SetPipelineState(PipelineState.Stopping);314 break;315 }316 }317 318 // If pipeline is already in stopping state. Wait for pipeline319 // to finish. We do need to raise any events here as no320 // change of state has occurred.321 if (alreadyStopping)322 {323 if (syncCall)324 {325 PipelineFinishedEvent.WaitOne();326 }327 328 return;329 }330 331 // Raise the event outside the lock332 RaisePipelineStateEvents();333 334 // A pipeline can be stopped before it is started. See NotStarted335 // case in above switch statement. This is done to allow stoping a pipeline336 // in another thread before it has been started.337 lock (SyncRoot)338 {339 if (PipelineState == PipelineState.Stopped)340 {341 // Note:if we have reached here, Stopped state was set342 // in PipelineState.NotStarted case above. Only other343 // way Stopped can be set when this method calls344 // StopHelper below345 return;346 }347 }348 349 // Start stop operation in derived class350 ImplementStop(syncCall);351 }352 353 /// <summary>354 /// Stop execution of pipeline.355 /// </summary>356 /// <param name="syncCall">If false, call is asynchronous.</param>357 protected abstract void ImplementStop(bool syncCall);358 359 #endregion stop360 361 #region invoke362 363 /// <summary>364 /// Invoke the pipeline, synchronously, returning the results as an365 /// array of objects.366 /// </summary>367 /// <param name="input">an array of input objects to pass to the pipeline.368 /// Array may be empty but may not be null</param>369 /// <returns>An array of zero or more result objects.</returns>370 /// <remarks>Caller of synchronous exectute should not close371 /// input objectWriter. Synchronous invoke will always close the input372 /// objectWriter.373 ///374 /// On Synchronous Invoke if output is throttled and no one is reading from375 /// output pipe, Execution will block after buffer is full.376 /// </remarks>377 public override Collection<PSObject> Invoke(IEnumerable input)378 {379 // NTRAID#Windows Out Of Band Releases-915851-2005/09/13380 if (_disposed)381 {382 throw PSTraceSource.NewObjectDisposedException("pipeline");383 }384 385 CoreInvoke(input, true);386 387 // Wait for pipeline to finish execution388 PipelineFinishedEvent.WaitOne();389 390 if (SyncInvokeCall)391 {392 // Raise the pipeline completion events. These events are set in393 // pipeline execution thread. However for Synchronous execution394 // we raise the event in the main thread.395 RaisePipelineStateEvents();396 }397 398 if (PipelineStateInfo.State == PipelineState.Stopped)399 {400 return new Collection<PSObject>();401 }402 else if (PipelineStateInfo.State == PipelineState.Failed && PipelineStateInfo.Reason != null)403 {404 // If this is an error pipe for a hosting applicationand we are logging,405 // then log the error.406 if (this.Runspace.GetExecutionContext.EngineHostInterface.UI.IsTranscribing)407 {408 this.Runspace.ExecutionContext.InternalHost.UI.TranscribeResult(this.Runspace, PipelineStateInfo.Reason.Message);409 }410 411 throw PipelineStateInfo.Reason;412 }413 414 // Execution completed successfully415 // 2004/06/30-JonN was ReadAll() which was non-blocking416 return Output.NonBlockingRead(Int32.MaxValue);417 }418 419 /// <summary>420 /// Invoke the pipeline asynchronously.421 /// </summary>422 /// <remarks>423 /// Results are returned through the <see cref="Pipeline.Output"/> reader.424 /// </remarks>425 public override void InvokeAsync()426 {427 CoreInvoke(null, false);428 }429 430 /// <summary>431 /// This parameter is true if Invoke is called.432 /// It is false if InvokeAsync is called.433 /// </summary>434 protected bool SyncInvokeCall { get; private set; }435 436 /// <summary>437 /// Invoke the pipeline asynchronously with input.438 /// </summary>439 /// <param name="input">input to provide to pipeline. Input is440 /// used only for synchronous execution</param>441 /// <param name="syncCall">True if this method is called from442 /// synchronous invoke else false</param>443 /// <remarks>444 /// Results are returned through the <see cref="Pipeline.Output"/> reader.445 /// </remarks>446 /// <exception cref="InvalidOperationException">447 /// No command is added to pipeline448 /// </exception>449 /// <exception cref="InvalidPipelineStateException">450 /// PipelineState is not NotStarted.451 /// </exception>452 /// <exception cref="InvalidOperationException">453 /// 1) A pipeline is already executing. Pipeline cannot execute454 /// concurrently.455 /// 2) InvokeAsync is called on nested pipeline. Nested pipeline456 /// cannot be executed Asynchronously.457 /// 3) Attempt is made to invoke a nested pipeline directly. Nested458 /// pipeline must be invoked from a running pipeline.459 /// </exception>460 /// <exception cref="InvalidRunspaceStateException">461 /// RunspaceState is not Open462 /// </exception>463 /// <exception cref="ObjectDisposedException">464 /// Pipeline already disposed465 /// </exception>466 private void CoreInvoke(IEnumerable input, bool syncCall)467 {468 lock (SyncRoot)469 {470 // NTRAID#Windows Out Of Band Releases-915851-2005/09/13471 if (_disposed)472 {473 throw PSTraceSource.NewObjectDisposedException("pipeline");474 }475 476 if (Commands == null || Commands.Count == 0)477 {478 throw PSTraceSource.NewInvalidOperationException(479 RunspaceStrings.NoCommandInPipeline);480 }481 482 if (PipelineState != PipelineState.NotStarted)483 {484 InvalidPipelineStateException e =485 new InvalidPipelineStateException486 (487 StringUtil.Format(RunspaceStrings.PipelineReInvokeNotAllowed),488 PipelineState,489 PipelineState.NotStarted490 );491 throw e;492 }493 494 if (syncCall495 && InputStream is not PSDataCollectionStream<PSObject>496 && InputStream is not PSDataCollectionStream<object>)497 {498 // Method is called from synchronous invoke.499 if (input != null)500 {501 // TO-DO-Add a test make sure that ObjectDisposed502 // exception is thrown503 // Write input data in to inputStream and close the input504 // pipe. If Input stream is already closed an505 // ObjectDisposed exception will be thrown506 foreach (object temp in input)507 {508 InputStream.Write(temp);509 }510 }511 512 InputStream.Close();513 }514 515 SyncInvokeCall = syncCall;516 517 // Create event which will be signalled when pipeline execution518 // is completed/failed/stopped.519 // Note:Runspace.Close waits for all the running pipeline520 // to finish. This Event must be created before pipeline is521 // added to list of running pipelines. This avoids the race condition522 // where Close is called after pipeline is added to list of523 // running pipeline but before event is created.524 PipelineFinishedEvent = new ManualResetEvent(false);525 526 // 1) Do the check to ensure that pipeline no other527 // pipeline is running.528 // 2) Runspace object maintains a list of pipelines in529 // execution. Add this pipeline to the list.530 RunspaceBase.DoConcurrentCheckAndAddToRunningPipelines(this, syncCall);531 532 // Note: Set PipelineState to Running only after adding pipeline to list533 // of pipelines in execution. AddForExecution checks that runspace is in534 // state where pipeline can be run.535 // StartPipelineExecution raises this event. See Windows Bug 1160481 for536 // more details.537 SetPipelineState(PipelineState.Running);538 }539 540 try541 {542 // Let the derived class start the pipeline execution.543 StartPipelineExecution();544 }545 catch (Exception exception)546 {547 // If we fail in any of the above three steps, set the correct states.548 RunspaceBase.RemoveFromRunningPipelineList(this);549 SetPipelineState(PipelineState.Failed, exception);550 551 // Note: we are not raising the events in this case. However this is552 // fine as user is getting the exception.553 throw;554 }555 }556 557 /// <summary>558 /// Invokes a remote command and immediately disconnects if559 /// transport layer supports it.560 /// </summary>561 internal override void InvokeAsyncAndDisconnect()562 {563 throw new NotSupportedException();564 }565 566 /// <summary>567 /// Starts execution of pipeline.568 /// </summary>569 protected abstract void StartPipelineExecution();570 571 #region concurrent pipeline check572 573 private bool _performNestedCheck = true;574 575 /// <summary>576 /// For nested pipeline, system checks that Execute is called from577 /// currently executing pipeline.578 /// If PerformNestedCheck is false, this check is bypassed. This579 /// is set to true by remote provider. In remote provider case all580 /// the checks are done by the client proxy.581 /// </summary>582 internal bool PerformNestedCheck583 {584 set585 {586 _performNestedCheck = value;587 }588 }589 590 /// <summary>591 /// This is the thread on which NestedPipeline can be executed.592 /// In case of LocalPipeline, this is the thread of execution593 /// of LocalPipeline. In case of RemotePipeline, this is thread594 /// on which EnterNestedPrompt is called.595 /// RemotePipeline proxy should set it on at the beginning of596 /// EnterNestedPrompt and clear it on return.597 /// </summary>598 internal Thread NestedPipelineExecutionThread { get; set; }599 600 /// <summary>601 /// Check if anyother pipeline is executing.602 /// In case of nested pipeline, checks that it is called603 /// from currently executing pipeline's thread.604 /// </summary>605 /// <param name="syncCall">True if method is called from Invoke, false606 /// if called from InvokeAsync</param>607 /// <param name="syncObject">The sync object on which the lock is acquired.</param>608 /// <param name="isInLock">True if the method is invoked in a critical section.</param>609 /// <exception cref="InvalidOperationException">610 /// 1) A pipeline is already executing. Pipeline cannot execute611 /// concurrently.612 /// 2) InvokeAsync is called on nested pipeline. Nested pipeline613 /// cannot be executed Asynchronously.614 /// 3) Attempt is made to invoke a nested pipeline directly. Nested615 /// pipeline must be invoked from a running pipeline.616 /// </exception>617 internal void DoConcurrentCheck(bool syncCall, object syncObject, bool isInLock)618 {619 PipelineBase currentPipeline = (PipelineBase)RunspaceBase.GetCurrentlyRunningPipeline();620 621 if (!IsNested)622 {623 if (currentPipeline == null)624 {625 return;626 }627 else628 {629 // Detect if we're running a pulse pipeline, or we're running a nested pipeline630 // in a pulse pipeline631 if (currentPipeline == RunspaceBase.PulsePipeline ||632 (currentPipeline.IsNested && RunspaceBase.PulsePipeline != null))633 {634 // If so, wait and try again635 if (isInLock)636 {637 // If the method is invoked in the lock statement, release the638 // lock before wait on the pulse pipeline639 Monitor.Exit(syncObject);640 }641 642 try643 {644 RunspaceBase.WaitForFinishofPipelines();645 }646 finally647 {648 if (isInLock)649 {650 // If the method is invoked in the lock statement, acquire the651 // lock before we carry on with the rest operations652 Monitor.Enter(syncObject);653 }654 }655 656 DoConcurrentCheck(syncCall, syncObject, isInLock);657 return;658 }659 660 throw PSTraceSource.NewInvalidOperationException(661 RunspaceStrings.ConcurrentInvokeNotAllowed);662 }663 }664 else665 {666 if (_performNestedCheck)667 {668 if (!syncCall)669 {670 throw PSTraceSource.NewInvalidOperationException(671 RunspaceStrings.NestedPipelineInvokeAsync);672 }673 674 if (currentPipeline == null)675 {676 if (this.IsChild)677 {678 // OK it's not really a nested pipeline but a call with RunspaceMode=UseCurrentRunspace679 // This shouldn't fail so we'll clear the IsNested and IsChild flags and then return680 // That way executions proceeds but everything gets clean up at the end when the pipeline completes681 this.IsChild = false;682 _isNested = false;683 return;684 }685 686 throw PSTraceSource.NewInvalidOperationException(687 RunspaceStrings.NestedPipelineNoParentPipeline);688 }689 690 Dbg.Assert(currentPipeline.NestedPipelineExecutionThread != null, "Current pipeline should always have NestedPipelineExecutionThread set");691 Thread th = Thread.CurrentThread;692 693 if (!currentPipeline.NestedPipelineExecutionThread.Equals(th))694 {695 throw PSTraceSource.NewInvalidOperationException(696 RunspaceStrings.NestedPipelineNoParentPipeline);697 }698 }699 }700 }701 702 #endregion concurrent pipeline check703 704 #endregion invoke705 706 #region Connect707 708 /// <summary>709 /// Connects synchronously to a running command on a remote server.710 /// The pipeline object must be in the disconnected state.711 /// </summary>712 /// <returns>A collection of result objects.</returns>713 public override Collection<PSObject> Connect()714 {715 // Connect semantics not supported on local (non-remoting) pipelines.716 throw PSTraceSource.NewNotSupportedException(PipelineStrings.ConnectNotSupported);717 }718 719 /// <summary>720 /// Connects asynchronously to a running command on a remote server.721 /// </summary>722 public override void ConnectAsync()723 {724 // Connect semantics not supported on local (non-remoting) pipelines.725 throw PSTraceSource.NewNotSupportedException(PipelineStrings.ConnectNotSupported);726 }727 728 #endregion729 730 #region state change event731 732 /// <summary>733 /// Event raised when Pipeline's state changes.734 /// </summary>735 public override event EventHandler<PipelineStateEventArgs> StateChanged = null;736 737 /// <summary>738 /// Current state of the pipeline.739 /// </summary>740 /// <remarks>741 /// This value indicates the state of the pipeline after the change.742 /// </remarks>743 protected PipelineState PipelineState744 {745 get746 {747 return _pipelineStateInfo.State;748 }749 }750 751 /// <summary>752 /// This returns true if pipeline state is Completed, Failed or Stopped.753 /// </summary>754 /// <returns></returns>755 protected bool IsPipelineFinished()756 {757 return (PipelineState == PipelineState.Completed ||758 PipelineState == PipelineState.Failed ||759 PipelineState == PipelineState.Stopped);760 }761 762 /// <summary>763 /// This is queue of all the state change event which have occurred for764 /// this pipeline. RaisePipelineStateEvents raises event for each765 /// item in this queue. We don't raise the event with in SetPipelineState766 /// because often SetPipelineState is called with in a lock.767 /// Raising event in lock introduces chances of deadlock in GUI applications.768 /// </summary>769 private Queue<ExecutionEventQueueItem> _executionEventQueue = new Queue<ExecutionEventQueueItem>();770 771 private sealed class ExecutionEventQueueItem772 {773 public ExecutionEventQueueItem(PipelineStateInfo pipelineStateInfo, RunspaceAvailability currentAvailability, RunspaceAvailability newAvailability)774 {775 this.PipelineStateInfo = pipelineStateInfo;776 this.CurrentRunspaceAvailability = currentAvailability;777 this.NewRunspaceAvailability = newAvailability;778 }779 780 public PipelineStateInfo PipelineStateInfo;781 public RunspaceAvailability CurrentRunspaceAvailability;782 public RunspaceAvailability NewRunspaceAvailability;783 }784 785 /// <summary>786 /// Sets the new execution state.787 /// </summary>788 /// <param name="state">The new state.</param>789 /// <param name="reason">790 /// An exception indicating that state change is the result of an error,791 /// otherwise; null.792 /// </param>793 /// <remarks>794 /// Sets the internal execution state information member variable. It795 /// also adds PipelineStateInfo to a queue. RaisePipelineStateEvents796 /// raises event for each item in this queue.797 /// </remarks>798 protected void SetPipelineState(PipelineState state, Exception reason)799 {800 lock (SyncRoot)801 {802 if (state != PipelineState)803 {804 _pipelineStateInfo = new PipelineStateInfo(state, reason);805 806 // Add _pipelineStateInfo to _executionEventQueue.807 // RaisePipelineStateEvents will raise event for each item808 // in this queue.809 // Note:We are doing clone here instead of passing the member810 // _pipelineStateInfo because we donot want outside811 // to change pipeline state.812 RunspaceAvailability previousAvailability = _runspace.RunspaceAvailability;813 814 _runspace.UpdateRunspaceAvailability(_pipelineStateInfo.State, false);815 816 _executionEventQueue.Enqueue(817 new ExecutionEventQueueItem(818 _pipelineStateInfo.Clone(),819 previousAvailability,820 _runspace.RunspaceAvailability));821 }822 }823 }824 825 /// <summary>826 /// Set the new execution state.827 /// </summary>828 /// <param name="state">The new state.</param>829 protected void SetPipelineState(PipelineState state)830 {831 SetPipelineState(state, null);832 }833 834 /// <summary>835 /// Raises events for changes in execution state.836 /// </summary>837 protected void RaisePipelineStateEvents()838 {839 Queue<ExecutionEventQueueItem> tempEventQueue = null;840 EventHandler<PipelineStateEventArgs> stateChanged = null;841 bool runspaceHasAvailabilityChangedSubscribers = false;842 843 lock (SyncRoot)844 {845 stateChanged = this.StateChanged;846 runspaceHasAvailabilityChangedSubscribers = _runspace.HasAvailabilityChangedSubscribers;847 848 if (stateChanged != null || runspaceHasAvailabilityChangedSubscribers)849 {850 tempEventQueue = _executionEventQueue;851 _executionEventQueue = new Queue<ExecutionEventQueueItem>();852 }853 else854 {855 // Clear the events if there are no EventHandlers. This856 // ensures that events do not get called for state857 // changes prior to their registration.858 _executionEventQueue.Clear();859 }860 }861 862 if (tempEventQueue != null)863 {864 while (tempEventQueue.Count > 0)865 {866 ExecutionEventQueueItem queueItem = tempEventQueue.Dequeue();867 868 if (runspaceHasAvailabilityChangedSubscribers && queueItem.NewRunspaceAvailability != queueItem.CurrentRunspaceAvailability)869 {870 _runspace.RaiseAvailabilityChangedEvent(queueItem.NewRunspaceAvailability);871 }872 873 // this is shipped as part of V1. So disabling the warning here.874#pragma warning disable 56500875 // Exception raised in the eventhandler are not error in pipeline.876 // silently ignore them.877 if (stateChanged != null)878 {879 try880 {881 stateChanged(this, new PipelineStateEventArgs(queueItem.PipelineStateInfo));882 }883 catch (Exception)884 {885 }886 }887#pragma warning restore 56500888 }889 }890 }891 892 /// <summary>893 /// ManualResetEvent which is signaled when pipeline execution is894 /// completed/failed/stopped.895 /// </summary>896 internal ManualResetEvent PipelineFinishedEvent { get; private set; }897 898 #endregion899 900 #region streams901 902 /// <summary>903 /// OutputStream from PipelineProcessor. Host will read on904 /// ObjectReader of this stream. PipelineProcessor will write to905 /// ObjectWriter of this stream.906 /// </summary>907 protected ObjectStreamBase OutputStream { get; }908 909 private ObjectStreamBase _errorStream;910 /// <summary>911 /// ErrorStream from PipelineProcessor. Host will read on912 /// ObjectReader of this stream. PipelineProcessor will write to913 /// ObjectWriter of this stream.914 /// </summary>915 protected ObjectStreamBase ErrorStream916 {917 get918 {919 return _errorStream;920 }921 922 private set923 {924 Dbg.Assert(value != null, "ErrorStream cannot be null");925 _errorStream = value;926 _errorStream.DataReady += OnErrorStreamDataReady;927 }928 }929 930 // Winblue: 26115. This handler is used to populate Pipeline.HadErrors.931 private void OnErrorStreamDataReady(object sender, EventArgs e)932 {933 if (_errorStream.Count > 0)934 {935 // unsubscribe from further event notifications as936 // this notification is suffice to say there is an937 // error.938 _errorStream.DataReady -= OnErrorStreamDataReady;939 SetHadErrors(true);940 }941 }942 943 /// <summary>944 /// Informational Buffers that represent verbose, debug, progress,945 /// warning emanating from the command execution.946 /// </summary>947 /// <remarks>948 /// Informational buffers are introduced after 1.0. This can be949 /// null if executing command as part of 1.0 hosting interfaces.950 /// </remarks>951 protected PSInformationalBuffers InformationalBuffers { get; }952 953 /// <summary>954 /// Stream for providing input to PipelineProcessor. Host will write on955 /// ObjectWriter of this stream. PipelineProcessor will read from956 /// ObjectReader of this stream.957 /// </summary>958 protected ObjectStreamBase InputStream { get; }959 960 #endregion streams961 962 #region history963 964 // History information is internal so that Pipeline serialization code965 // can access it.966 967 /// <summary>968 /// If true, this pipeline is added in history.969 /// </summary>970 internal bool AddToHistory { get; set; }971 972 /// <summary>973 /// String which is added in the history.974 /// </summary>975 /// <remarks>This needs to be internal so that it can be replaced976 /// by invoke-cmd to place correct string in history.</remarks>977 internal string HistoryString { get; set; }978 979 #endregion history980 981 #region misc982 983 /// <summary>984 /// Initialized the current pipeline instance with the supplied data.985 /// </summary>986 /// <param name="runspace"></param>987 /// <param name="command"></param>988 /// <param name="addToHistory"></param>989 /// <param name="isNested"></param>990 /// <exception cref="ArgumentNullException">991 /// 1. addToHistory is true and command is null.992 /// </exception>993 private void Initialize(Runspace runspace, string command, bool addToHistory, bool isNested)994 {995 Dbg.Assert(runspace != null, "caller should validate the parameter");996 _runspace = runspace;997 998 _isNested = isNested;999 1000 if (addToHistory && command == null)1001 {1002 throw PSTraceSource.NewArgumentNullException(nameof(command));1003 }1004 1005 if (command != null)1006 {1007 Commands.Add(new Command(command, true, false));1008 }1009 1010 AddToHistory = addToHistory;1011 if (AddToHistory)1012 {1013 HistoryString = command;1014 }1015 }1016 1017 private RunspaceBase RunspaceBase1018 {1019 get1020 {1021 return (RunspaceBase)Runspace;1022 }1023 }1024 1025 /// <summary>1026 /// Object used for synchronization.1027 /// </summary>1028 protected internal object SyncRoot { get; } = new object();1029 1030 #endregion misc1031 1032 #region IDisposable Members1033 1034 /// <summary>1035 /// Set to true when object is disposed.1036 /// </summary>1037 private bool _disposed;1038 1039 /// <summary>1040 /// Protected dispose which can be overridden by derived classes.1041 /// </summary>1042 /// <param name="disposing"></param>1043 protected override1044 void1045 Dispose(bool disposing)1046 {1047 try1048 {1049 if (!_disposed)1050 {1051 _disposed = true;1052 if (disposing)1053 {1054 InputStream.Close();1055 OutputStream.Close();1056 1057 _errorStream.DataReady -= OnErrorStreamDataReady;1058 _errorStream.Close();1059 1060 _executionEventQueue.Clear();1061 }1062 }1063 }1064 finally1065 {1066 base.Dispose(disposing);1067 }1068 }1069 1070 #endregion IDisposable Members1071 }1072}1073