MegaBites-AI/Windows-powershell
0372
1// Copyright (c) Microsoft Corporation.2// Licensed under the MIT License.3 4using System.Collections.Generic;5using System.Collections.ObjectModel;6using System.Management.Automation.Internal;7using System.Management.Automation.Remoting;8using System.Management.Automation.Runspaces;9using System.Management.Automation.Runspaces.Internal;10using System.Threading;11 12using Dbg = System.Management.Automation.Diagnostics;13 14#pragma warning disable 1634, 1691 // Stops compiler from warning about unknown warnings15 16namespace System.Management.Automation17{18 internal class RemotePipeline : Pipeline19 {20 #region Private Members21 22 private PowerShell _powershell;23 private readonly bool _addToHistory;24 private bool _isNested;25 private bool _isSteppable;26 private readonly Runspace _runspace;27 private readonly object _syncRoot = new object();28 private bool _disposed = false;29 private string _historyString;30 private PipelineStateInfo _pipelineStateInfo = new PipelineStateInfo(PipelineState.NotStarted);31 private readonly CommandCollection _commands = new CommandCollection();32 private readonly string _computerName;33 private readonly Guid _runspaceId;34 private readonly ConnectCommandInfo _connectCmdInfo = null;35 36 /// <summary>37 /// This is queue of all the state change event which have occurred for38 /// this pipeline. RaisePipelineStateEvents raises event for each39 /// item in this queue. We don't raise the event with in SetPipelineState40 /// because often SetPipelineState is called with in a lock.41 /// Raising event in lock introduces chances of deadlock in GUI applications.42 /// </summary>43 private Queue<ExecutionEventQueueItem> _executionEventQueue = new Queue<ExecutionEventQueueItem>();44 45 private sealed class ExecutionEventQueueItem46 {47 public ExecutionEventQueueItem(PipelineStateInfo pipelineStateInfo, RunspaceAvailability currentAvailability, RunspaceAvailability newAvailability)48 {49 this.PipelineStateInfo = pipelineStateInfo;50 this.CurrentRunspaceAvailability = currentAvailability;51 this.NewRunspaceAvailability = newAvailability;52 }53 54 public PipelineStateInfo PipelineStateInfo;55 public RunspaceAvailability CurrentRunspaceAvailability;56 public RunspaceAvailability NewRunspaceAvailability;57 }58 59 private readonly bool _performNestedCheck = true;60 61 #endregion Private Members62 63 #region Constructors64 65 /// <summary>66 /// Private constructor that does most of the work constructing a remote pipeline object.67 /// </summary>68 /// <param name="runspace">RemoteRunspace object.</param>69 /// <param name="addToHistory">AddToHistory.</param>70 /// <param name="isNested">IsNested.</param>71 private RemotePipeline(RemoteRunspace runspace, bool addToHistory, bool isNested)72 : base(runspace)73 {74 _addToHistory = addToHistory;75 _isNested = isNested;76 _isSteppable = false;77 _runspace = runspace;78 _computerName = ((RemoteRunspace)_runspace).ConnectionInfo.ComputerName;79 _runspaceId = _runspace.InstanceId;80 81 // Initialize streams82 _inputCollection = new PSDataCollection<object>();83 _inputCollection.ReleaseOnEnumeration = true;84 85 _inputStream = new PSDataCollectionStream<object>(Guid.Empty, _inputCollection);86 _outputCollection = new PSDataCollection<PSObject>();87 _outputStream = new PSDataCollectionStream<PSObject>(Guid.Empty, _outputCollection);88 _errorCollection = new PSDataCollection<ErrorRecord>();89 _errorStream = new PSDataCollectionStream<ErrorRecord>(Guid.Empty, _errorCollection);90 91 // Create object stream for method executor objects.92 MethodExecutorStream = new ObjectStream();93 IsMethodExecutorStreamEnabled = false;94 95 SetCommandCollection(_commands);96 97 // Create event which will be signalled when pipeline execution98 // is completed/failed/stopped.99 // Note:Runspace.Close waits for all the running pipeline100 // to finish. This Event must be created before pipeline is101 // added to list of running pipelines. This avoids the race condition102 // where Close is called after pipeline is added to list of103 // running pipeline but before event is created.104 PipelineFinishedEvent = new ManualResetEvent(false);105 }106 107 /// <summary>108 /// Constructs a remote pipeline for the specified runspace and109 /// specified command.110 /// </summary>111 /// <param name="runspace">Runspace in which to create the pipeline.</param>112 /// <param name="command">Command as a string, to be used in pipeline creation.</param>113 /// <param name="addToHistory">Whether to add the command to the runspaces history.</param>114 /// <param name="isNested">Whether this pipeline is nested.</param>115 internal RemotePipeline(RemoteRunspace runspace, string command, bool addToHistory, bool isNested)116 : this(runspace, addToHistory, isNested)117 {118 if (command != null)119 {120 _commands.Add(new Command(command, true));121 }122 123 // initialize the underlying powershell object124 _powershell = new PowerShell(_inputStream, _outputStream, _errorStream,125 ((RemoteRunspace)_runspace).RunspacePool);126 127 _powershell.SetIsNested(isNested);128 129 _powershell.InvocationStateChanged += HandleInvocationStateChanged;130 }131 132 /// <summary>133 /// Constructs a remote pipeline object associated with a remote running134 /// command but in a disconnected state.135 /// </summary>136 /// <param name="runspace">Remote runspace associated with running command.</param>137 internal RemotePipeline(RemoteRunspace runspace)138 : this(runspace, false, false)139 {140 if (runspace.RemoteCommand == null)141 {142 throw new InvalidOperationException(PipelineStrings.InvalidRemoteCommand);143 }144 145 _connectCmdInfo = runspace.RemoteCommand;146 _commands.Add(_connectCmdInfo.Command);147 148 // Beginning state will be disconnected.149 SetPipelineState(PipelineState.Disconnected, null);150 151 // Create the underlying powershell object.152 _powershell = new PowerShell(_connectCmdInfo, _inputStream, _outputStream, _errorStream,153 ((RemoteRunspace)_runspace).RunspacePool);154 155 _powershell.InvocationStateChanged += HandleInvocationStateChanged;156 }157 158 /// <summary>159 /// Creates a cloned pipeline from the specified one.160 /// </summary>161 /// <param name="pipeline">Pipeline to clone from.</param>162 /// <remarks>This constructor is private because this will163 /// only be called from the copy method</remarks>164 private RemotePipeline(RemotePipeline pipeline)165 : this(166 (RemoteRunspace)pipeline.Runspace,167 command: null,168 addToHistory: false,169 pipeline.IsNested)170 {171 _isSteppable = pipeline._isSteppable;172 173 // NTRAID#Windows Out Of Band Releases-915851-2005/09/13174 // the above comment copied from RemotePipelineBase which175 // originally copied it from PipelineBase176 if (pipeline == null)177 {178 throw PSTraceSource.NewArgumentNullException(nameof(pipeline));179 }180 181 if (pipeline._disposed)182 {183 throw PSTraceSource.NewObjectDisposedException("pipeline");184 }185 186 _addToHistory = pipeline._addToHistory;187 _historyString = pipeline._historyString;188 foreach (Command command in pipeline.Commands)189 {190 Command clone = command.Clone();191 192 // Attach the cloned Command to this pipeline.193 Commands.Add(clone);194 }195 }196 197 /// <summary>198 /// Override for creating a copy of pipeline.199 /// </summary>200 /// <returns>201 /// Pipeline object which is copy of this pipeline202 /// </returns>203 public override Pipeline Copy()204 {205 if (_disposed)206 {207 throw PSTraceSource.NewObjectDisposedException("pipeline");208 }209 210 return (Pipeline)new RemotePipeline(this);211 }212 213 #endregion Constructors214 215 #region Properties216 217 /// <summary>218 /// Access the runspace this pipeline is created on.219 /// </summary>220 public override Runspace Runspace221 {222 get223 {224#pragma warning disable 56503225 // NTRAID#Windows Out Of Band Releases-915851-2005/09/13226 if (_disposed)227 {228 throw PSTraceSource.NewObjectDisposedException("pipeline");229 }230#pragma warning restore 56503231 232 return _runspace;233 }234 }235 236 /// <summary>237 /// This internal method doesn't do the _disposed check.238 /// </summary>239 /// <returns></returns>240 internal Runspace GetRunspace()241 {242 return _runspace;243 }244 245 /// <summary>246 /// Is this pipeline nested.247 /// </summary>248 public override bool IsNested249 {250 get251 {252 return _isNested;253 }254 }255 256 /// <summary>257 /// Internal method to set the value of IsNested. This is called258 /// by serializer.259 /// </summary>260 internal void SetIsNested(bool isNested)261 {262 _isNested = isNested;263 _powershell.SetIsNested(isNested);264 }265 266 /// <summary>267 /// Internal method to set the value of IsSteppable. This is called268 /// during DoConcurrentCheck.269 /// </summary>270 internal void SetIsSteppable(bool isSteppable)271 {272 _isSteppable = isSteppable;273 }274 275 /// <summary>276 /// Info about current state of the pipeline.277 /// </summary>278 /// <remarks>279 /// This value indicates the state of the pipeline after the change.280 /// </remarks>281 public override PipelineStateInfo PipelineStateInfo282 {283 get284 {285 lock (_syncRoot)286 {287 // Note:We do not return internal state.288 return _pipelineStateInfo.Clone();289 }290 }291 }292 293 /// <summary>294 /// Access the input writer for this pipeline.295 /// </summary>296 public override PipelineWriter Input297 {298 get299 {300 return _inputStream.ObjectWriter;301 }302 }303 304 /// <summary>305 /// Access the output reader for this pipeline.306 /// </summary>307 public override PipelineReader<PSObject> Output308 {309 get310 {311 return _outputStream.GetPSObjectReaderForPipeline(_computerName, _runspaceId);312 }313 }314 315 /// <summary>316 /// Access the error output reader for this pipeline.317 /// </summary>318 /// <remarks>319 /// This is the non-terminating error stream from the command.320 /// In this release, the objects read from this PipelineReader321 /// are PSObjects wrapping ErrorRecords.322 /// </remarks>323 public override PipelineReader<object> Error324 {325 get326 {327 return _errorStream.GetObjectReaderForPipeline(_computerName, _runspaceId);328 }329 }330 331 /// <summary>332 /// String which is added in the history.333 /// </summary>334 /// <remarks>This needs to be internal so that it can be replaced335 /// by invoke-cmd to place correct string in history.</remarks>336 internal string HistoryString337 {338 get339 {340 return _historyString;341 }342 343 set344 {345 _historyString = value;346 }347 }348 349 /// <summary>350 /// Whether the pipeline needs to be added to history of the runspace.351 /// </summary>352 public bool AddToHistory353 {354 get355 {356 return _addToHistory;357 }358 }359 360 #endregion Properties361 362 #region streams363 364 // Stream and Collection go together...a stream wraps365 // a corresponding collection to support366 // streaming behavior of the pipeline.367 private readonly PSDataCollection<PSObject> _outputCollection;368 private readonly PSDataCollectionStream<PSObject> _outputStream;369 private readonly PSDataCollection<ErrorRecord> _errorCollection;370 private readonly PSDataCollectionStream<ErrorRecord> _errorStream;371 private readonly PSDataCollection<object> _inputCollection;372 private readonly PSDataCollectionStream<object> _inputStream;373 374 /// <summary>375 /// Stream for providing input to PipelineProcessor. Host will write on376 /// ObjectWriter of this stream. PipelineProcessor will read from377 /// ObjectReader of this stream.378 /// </summary>379 protected PSDataCollectionStream<object> InputStream380 {381 get382 {383 return _inputStream;384 }385 }386 387 #endregion streams388 389 #region Invoke390 391 /// <summary>392 /// Invoke the pipeline asynchronously.393 /// </summary>394 /// <remarks>395 /// Results are returned through the <see cref="Pipeline.Output"/> reader.396 /// </remarks>397 public override void InvokeAsync()398 {399 InitPowerShell(false);400 CoreInvokeAsync();401 }402 403 /// <summary>404 /// Invokes a remote command and immediately disconnects if405 /// transport layer supports it.406 /// </summary>407 internal override void InvokeAsyncAndDisconnect()408 {409 // Initialize PowerShell invocation with "InvokeAndDisconnect" setting.410 InitPowerShell(false, true);411 CoreInvokeAsync();412 }413 414 /// <summary>415 /// Invoke the pipeline, synchronously, returning the results as an416 /// array of objects.417 /// </summary>418 /// <param name="input">an array of input objects to pass to the pipeline.419 /// Array may be empty but may not be null</param>420 /// <returns>An array of zero or more result objects.</returns>421 /// <remarks>Caller of synchronous exectute should not close422 /// input objectWriter. Synchronous invoke will always close the input423 /// objectWriter.424 ///425 /// On Synchronous Invoke if output is throttled and no one is reading from426 /// output pipe, Execution will block after buffer is full.427 /// </remarks>428 public override Collection<PSObject> Invoke(System.Collections.IEnumerable input)429 {430 if (input == null)431 {432 this.InputStream.Close();433 }434 435 InitPowerShell(true);436 437 Collection<PSObject> results;438 439 try440 {441 results = _powershell.Invoke(input);442 }443 catch (InvalidRunspacePoolStateException)444 {445 InvalidRunspaceStateException e =446 new InvalidRunspaceStateException447 (448 StringUtil.Format(RunspaceStrings.RunspaceNotOpenForPipeline, _runspace.RunspaceStateInfo.State.ToString()),449 _runspace.RunspaceStateInfo.State,450 RunspaceState.Opened451 );452 throw e;453 }454 455 return results;456 }457 458 #endregion Invoke459 460 #region Connect461 462 /// <summary>463 /// Connects synchronously to a running command on a remote server.464 /// The pipeline object must be in the disconnected state.465 /// </summary>466 /// <returns>A collection of result objects.</returns>467 public override Collection<PSObject> Connect()468 {469 InitPowerShellForConnect(true);470 471 Collection<PSObject> results;472 473 try474 {475 results = _powershell.Connect();476 }477 catch (InvalidRunspacePoolStateException)478 {479 InvalidRunspaceStateException e =480 new InvalidRunspaceStateException481 (482 StringUtil.Format(RunspaceStrings.RunspaceNotOpenForPipelineConnect, _runspace.RunspaceStateInfo.State.ToString()),483 _runspace.RunspaceStateInfo.State,484 RunspaceState.Opened485 );486 487 throw e;488 }489 490 // PowerShell object will return empty results if it was provided an alternative object to491 // collect output in. Check to see if the output was collected in a member variable.492 if (results.Count == 0)493 {494 if (_outputCollection != null && _outputCollection.Count > 0)495 {496 results = new Collection<PSObject>(_outputCollection);497 }498 }499 500 return results;501 }502 503 /// <summary>504 /// Connects asynchronously to a running command on a remote server.505 /// </summary>506 public override void ConnectAsync()507 {508 InitPowerShellForConnect(false);509 510 try511 {512 _powershell.ConnectAsync();513 }514 catch (InvalidRunspacePoolStateException)515 {516 InvalidRunspaceStateException e =517 new InvalidRunspaceStateException518 (519 StringUtil.Format(RunspaceStrings.RunspaceNotOpenForPipelineConnect, _runspace.RunspaceStateInfo.State.ToString()),520 _runspace.RunspaceStateInfo.State,521 RunspaceState.Opened522 );523 throw e;524 }525 }526 527 #endregion528 529 #region Stop530 531 /// <summary>532 /// Stop the pipeline synchronously.533 /// </summary>534 public override void Stop()535 {536 bool isAlreadyStopping = false;537 if (CanStopPipeline(out isAlreadyStopping))538 {539 // A pipeline can be stopped before it is started.so protecting against that540 if (_powershell != null)541 {542 IAsyncResult asyncresult = null;543 try544 {545 asyncresult = _powershell.BeginStop(null, null);546 }547 catch (ObjectDisposedException)548 {549 throw PSTraceSource.NewObjectDisposedException("Pipeline");550 }551 552 asyncresult.AsyncWaitHandle.WaitOne();553 }554 }555 556 // Waits until pipeline completes stop as this is a sync call.557 PipelineFinishedEvent.WaitOne();558 }559 560 /// <summary>561 /// Stop the pipeline asynchronously.562 /// This method calls the BeginStop on the underlying563 /// powershell and so any exception will be564 /// thrown on the same thread.565 /// </summary>566 public override void StopAsync()567 {568 bool isAlreadyStopping;569 if (CanStopPipeline(out isAlreadyStopping))570 {571 try572 {573 _powershell.BeginStop(null, null);574 }575 catch (ObjectDisposedException)576 {577 throw PSTraceSource.NewObjectDisposedException("Pipeline");578 }579 }580 }581 582 /// <summary>583 /// Verifies if the pipeline is in a state where it can be stopped.584 /// </summary>585 private bool CanStopPipeline(out bool isAlreadyStopping)586 {587 bool returnResult = false;588 isAlreadyStopping = false;589 lock (_syncRoot)590 {591 // SetPipelineState does not raise events..592 // so locking is ok here.593 switch (_pipelineStateInfo.State)594 {595 case PipelineState.NotStarted:596 SetPipelineState(PipelineState.Stopping, null);597 SetPipelineState(PipelineState.Stopped, null);598 returnResult = false;599 break;600 601 // If pipeline execution has failed or completed or602 // stopped, return silently.603 case PipelineState.Stopped:604 case PipelineState.Completed:605 case PipelineState.Failed:606 return false;607 608 // If pipeline is in Stopping state, ignore the second609 // stop.610 case PipelineState.Stopping:611 isAlreadyStopping = true;612 return false;613 614 case PipelineState.Running:615 case PipelineState.Disconnected:616 SetPipelineState(PipelineState.Stopping, null);617 returnResult = true;618 break;619 }620 }621 622 RaisePipelineStateEvents();623 624 return returnResult;625 }626 627 #endregion Stop628 629 #region Events630 631 /// <summary>632 /// Event raised when Pipeline's state changes.633 /// </summary>634 public override event EventHandler<PipelineStateEventArgs> StateChanged = null;635 636 #endregion Events637 638 #region Dispose639 640 /// <summary>641 /// Disposes the pipeline.642 /// </summary>643 /// <param name="disposing">True, when called on Dispose().</param>644 protected override void Dispose(bool disposing)645 {646 try647 {648 if (_disposed)649 {650 return;651 }652 653 lock (_syncRoot)654 {655 if (_disposed)656 {657 return;658 }659 660 _disposed = true;661 }662 663 if (disposing)664 {665 // wait for the pipeline to stop..this will block666 // if the pipeline is already stopping.667 Stop();668 // _pipelineFinishedEvent.Close();669 670 if (_powershell != null)671 {672 _powershell.Dispose();673 _powershell = null;674 }675 676 _inputCollection.Dispose();677 _inputStream.Dispose();678 _outputCollection.Dispose();679 _outputStream.Dispose();680 _errorCollection.Dispose();681 _errorStream.Dispose();682 MethodExecutorStream.Dispose();683 PipelineFinishedEvent.Dispose();684 }685 }686 finally687 {688 base.Dispose(disposing);689 }690 }691 692 #endregion Dispose693 694 #region Private Methods695 696 private void CoreInvokeAsync()697 {698 try699 {700 _powershell.BeginInvoke();701 }702 catch (InvalidRunspacePoolStateException)703 {704 InvalidRunspaceStateException e =705 new InvalidRunspaceStateException706 (707 StringUtil.Format(RunspaceStrings.RunspaceNotOpenForPipeline, _runspace.RunspaceStateInfo.State.ToString()),708 _runspace.RunspaceStateInfo.State,709 RunspaceState.Opened710 );711 throw e;712 }713 }714 715 private void HandleInvocationStateChanged(object sender, PSInvocationStateChangedEventArgs e)716 {717 SetPipelineState((PipelineState)e.InvocationStateInfo.State, e.InvocationStateInfo.Reason);718 719 RaisePipelineStateEvents();720 }721 722 /// <summary>723 /// Sets the new execution state.724 /// </summary>725 /// <param name="state">The new state.</param>726 /// <param name="reason">727 /// An exception indicating that state change is the result of an error,728 /// otherwise; null.729 /// </param>730 /// <remarks>731 /// Sets the internal execution state information member variable. It732 /// also adds PipelineStateInfo to a queue. RaisePipelineStateEvents733 /// raises event for each item in this queue.734 /// </remarks>735 private void SetPipelineState(PipelineState state, Exception reason)736 {737 PipelineState copyState = state;738 PipelineStateInfo copyStateInfo = null;739 740 lock (_syncRoot)741 {742 switch (_pipelineStateInfo.State)743 {744 case PipelineState.Completed:745 case PipelineState.Failed:746 case PipelineState.Stopped:747 return;748 749 case PipelineState.Running:750 {751 if (state == PipelineState.Running)752 {753 return;754 }755 }756 757 break;758 case PipelineState.Stopping:759 {760 if (state == PipelineState.Running || state == PipelineState.Stopping)761 {762 return;763 }764 else765 {766 copyState = PipelineState.Stopped;767 }768 }769 770 break;771 }772 773 _pipelineStateInfo = new PipelineStateInfo(copyState, reason);774 copyStateInfo = _pipelineStateInfo;775 776 // Add _pipelineStateInfo to _executionEventQueue.777 // RaisePipelineStateEvents will raise event for each item778 // in this queue.779 // Note:We are doing clone here instead of passing the member780 // _pipelineStateInfo because we donot want outside781 // to change pipeline state.782 RunspaceAvailability previousAvailability = _runspace.RunspaceAvailability;783 784 Guid? cmdInstanceId = (_powershell != null) ? _powershell.InstanceId : (Guid?)null;785 _runspace.UpdateRunspaceAvailability(_pipelineStateInfo.State, false, cmdInstanceId);786 787 _executionEventQueue.Enqueue(788 new ExecutionEventQueueItem(789 _pipelineStateInfo.Clone(),790 previousAvailability,791 _runspace.RunspaceAvailability));792 }793 794 // using the copyStateInfo here as this piece of code is795 // outside of lock and _pipelineStateInfo might get changed796 // by two threads running concurrently..so its value is797 // not guaranteed to be the same for this entire method call.798 // copyStateInfo is a local variable.799 if (copyStateInfo.State == PipelineState.Completed ||800 copyStateInfo.State == PipelineState.Failed ||801 copyStateInfo.State == PipelineState.Stopped)802 {803 Cleanup();804 }805 }806 807 /// <summary>808 /// Raises events for changes in execution state.809 /// </summary>810 protected void RaisePipelineStateEvents()811 {812 Queue<ExecutionEventQueueItem> tempEventQueue = null;813 EventHandler<PipelineStateEventArgs> stateChanged = null;814 bool runspaceHasAvailabilityChangedSubscribers = false;815 816 lock (_syncRoot)817 {818 stateChanged = this.StateChanged;819 runspaceHasAvailabilityChangedSubscribers = _runspace.HasAvailabilityChangedSubscribers;820 821 if (stateChanged != null || runspaceHasAvailabilityChangedSubscribers)822 {823 tempEventQueue = _executionEventQueue;824 _executionEventQueue = new Queue<ExecutionEventQueueItem>();825 }826 else827 {828 // Clear the events if there are no EventHandlers. This829 // ensures that events do not get called for state830 // changes prior to their registration.831 _executionEventQueue.Clear();832 }833 }834 835 if (tempEventQueue != null)836 {837 while (tempEventQueue.Count > 0)838 {839 ExecutionEventQueueItem queueItem = tempEventQueue.Dequeue();840 841 if (runspaceHasAvailabilityChangedSubscribers && queueItem.NewRunspaceAvailability != queueItem.CurrentRunspaceAvailability)842 {843 _runspace.RaiseAvailabilityChangedEvent(queueItem.NewRunspaceAvailability);844 }845 846 // Exception raised in the eventhandler are not error in pipeline.847 // silently ignore them.848 if (stateChanged != null)849 {850 try851 {852 stateChanged(this, new PipelineStateEventArgs(queueItem.PipelineStateInfo));853 }854 catch (Exception)855 {856 }857 }858 }859 }860 }861 862 /// <summary>863 /// Initializes the underlying PowerShell object after verifying864 /// if the pipeline is in a state where it can be invoked.865 /// If invokeAndDisconnect is true then the remote PowerShell866 /// command will be immediately disconnected after it begins867 /// running.868 /// </summary>869 /// <param name="syncCall">True if called from a sync call.</param>870 /// <param name="invokeAndDisconnect">Invoke and Disconnect.</param>871 private void InitPowerShell(bool syncCall, bool invokeAndDisconnect = false)872 {873 if (_commands == null || _commands.Count == 0)874 {875 throw PSTraceSource.NewInvalidOperationException(876 RunspaceStrings.NoCommandInPipeline);877 }878 879 if (_pipelineStateInfo.State != PipelineState.NotStarted)880 {881 InvalidPipelineStateException e =882 new InvalidPipelineStateException883 (884 StringUtil.Format(RunspaceStrings.PipelineReInvokeNotAllowed),885 _pipelineStateInfo.State,886 PipelineState.NotStarted887 );888 throw e;889 }890 891 ((RemoteRunspace)_runspace).DoConcurrentCheckAndAddToRunningPipelines(this, syncCall);892 893 PSInvocationSettings settings = new PSInvocationSettings();894 settings.AddToHistory = _addToHistory;895 settings.InvokeAndDisconnect = invokeAndDisconnect;896 897 _powershell.InitForRemotePipeline(_commands, _inputStream, _outputStream, _errorStream, settings, RedirectShellErrorOutputPipe);898 899 _powershell.RemotePowerShell.HostCallReceived += HandleHostCallReceived;900 }901 902 /// <summary>903 /// Initializes the underlying PowerShell object after verifying that it is904 /// in a state where it can connect to the remote command.905 /// </summary>906 /// <param name="syncCall"></param>907 private void InitPowerShellForConnect(bool syncCall)908 {909 if (_pipelineStateInfo.State != PipelineState.Disconnected)910 {911 throw new InvalidPipelineStateException(StringUtil.Format(PipelineStrings.PipelineNotDisconnected),912 _pipelineStateInfo.State,913 PipelineState.Disconnected);914 }915 916 // The connect may be from the same Pipeline that disconnected and in this case917 // the Pipeline state already exists. Or this could be a new Pipeline object918 // (connect reconstruction case) and new state is created.919 920 // Check to see if this pipeline already exists in the runspace.921 RemotePipeline currentPipeline = (RemotePipeline)((RemoteRunspace)_runspace).GetCurrentlyRunningPipeline();922 if (!ReferenceEquals(currentPipeline, this))923 {924 ((RemoteRunspace)_runspace).DoConcurrentCheckAndAddToRunningPipelines(this, syncCall);925 }926 927 // Initialize the PowerShell object if it hasn't been initialized before.928 if ((_powershell.RemotePowerShell) == null || !_powershell.RemotePowerShell.Initialized)929 {930 PSInvocationSettings settings = new PSInvocationSettings();931 settings.AddToHistory = _addToHistory;932 933 _powershell.InitForRemotePipelineConnect(_inputStream, _outputStream, _errorStream, settings, RedirectShellErrorOutputPipe);934 935 _powershell.RemotePowerShell.HostCallReceived += HandleHostCallReceived;936 }937 }938 939 /// <summary>940 /// Handle host call received.941 /// </summary>942 /// <param name="sender">Sender of this event, unused.</param>943 /// <param name="eventArgs">Arguments describing the host call to invoke.</param>944 private void HandleHostCallReceived(object sender, RemoteDataEventArgs<RemoteHostCall> eventArgs)945 {946 ClientMethodExecutor.Dispatch(947 _powershell.RemotePowerShell.DataStructureHandler.TransportManager,948 ((RemoteRunspace)_runspace).RunspacePool.RemoteRunspacePoolInternal.Host,949 _errorStream,950 MethodExecutorStream,951 IsMethodExecutorStreamEnabled,952 ((RemoteRunspace)_runspace).RunspacePool.RemoteRunspacePoolInternal,953 _powershell.InstanceId,954 eventArgs.Data);955 }956 957 /// <summary>958 /// Does the cleanup necessary on pipeline completion.959 /// </summary>960 private void Cleanup()961 {962 // Close the output stream if it is not closed.963 if (_outputStream.IsOpen)964 {965 try966 {967 _outputCollection.Complete();968 _outputStream.Close();969 }970 catch (ObjectDisposedException)971 {972 }973 }974 975 // Close the error stream if it is not closed.976 if (_errorStream.IsOpen)977 {978 try979 {980 _errorCollection.Complete();981 _errorStream.Close();982 }983 catch (ObjectDisposedException)984 {985 }986 }987 988 // Close the input stream if it is not closed.989 if (_inputStream.IsOpen)990 {991 try992 {993 _inputCollection.Complete();994 _inputStream.Close();995 }996 catch (ObjectDisposedException)997 {998 }999 }1000 1001 try1002 {1003 // Runspace object maintains a list of pipelines in execution.1004 // Remove this pipeline from the list. This method also calls the1005 // pipeline finished event.1006 ((RemoteRunspace)_runspace).RemoveFromRunningPipelineList(this);1007 1008 PipelineFinishedEvent.Set();1009 }1010 catch (ObjectDisposedException)1011 {1012 }1013 }1014 1015 #endregion Private Methods1016 1017 #region Internal Methods/Properties1018 1019 /// <summary>1020 /// ManualResetEvent which is signaled when pipeline execution is1021 /// completed/failed/stopped.1022 /// </summary>1023 internal ManualResetEvent PipelineFinishedEvent { get; }1024 1025 /// <summary>1026 /// Is method executor stream enabled.1027 /// </summary>1028 internal bool IsMethodExecutorStreamEnabled { get; set; }1029 1030 /// <summary>1031 /// Method executor stream.1032 /// </summary>1033 internal ObjectStream MethodExecutorStream { get; }1034 1035 /// <summary>1036 /// Check if anyother pipeline is executing.1037 /// In case of nested pipeline, checks that it is called1038 /// from currently executing pipeline's thread.1039 /// </summary>1040 /// <param name="syncCall">True if method is called from Invoke, false1041 /// if called from InvokeAsync</param>1042 /// <exception cref="InvalidOperationException">1043 /// 1) A pipeline is already executing. Pipeline cannot execute1044 /// concurrently.1045 /// 2) InvokeAsync is called on nested pipeline. Nested pipeline1046 /// cannot be executed Asynchronously.1047 /// 3) Attempt is made to invoke a nested pipeline directly. Nested1048 /// pipeline must be invoked from a running pipeline.1049 /// </exception>1050 internal void DoConcurrentCheck(bool syncCall)1051 {1052 RemotePipeline currentPipeline =1053 (RemotePipeline)((RemoteRunspace)_runspace).GetCurrentlyRunningPipeline();1054 1055 if (!_isNested)1056 {1057 if (currentPipeline == null &&1058 ((RemoteRunspace)_runspace).RunspaceAvailability != RunspaceAvailability.Busy &&1059 ((RemoteRunspace)_runspace).RunspaceAvailability != RunspaceAvailability.RemoteDebug)1060 {1061 // We can add a new pipeline to the runspace only if it is1062 // available (not busy).1063 return;1064 }1065 1066 if (currentPipeline == null &&1067 ((RemoteRunspace)_runspace).RemoteCommand != null &&1068 _connectCmdInfo != null &&1069 Guid.Equals(((RemoteRunspace)_runspace).RemoteCommand.CommandId, _connectCmdInfo.CommandId))1070 {1071 // Connect case. We can add a pipeline to a busy runspace when1072 // that pipeline represents the same command as is currently1073 // running.1074 return;1075 }1076 1077 if (currentPipeline != null &&1078 ReferenceEquals(currentPipeline, this))1079 {1080 // Reconnect case. We can add a pipeline to a busy runspace when the1081 // pipeline is the same (reconnecting).1082 return;1083 }1084 1085 if (!_isSteppable)1086 {1087 throw PSTraceSource.NewInvalidOperationException(1088 RunspaceStrings.ConcurrentInvokeNotAllowed);1089 }1090 }1091 else1092 {1093 if (_performNestedCheck)1094 {1095 if (_isSteppable)1096 {1097 return;1098 }1099 1100 if (!syncCall)1101 {1102 throw PSTraceSource.NewInvalidOperationException(1103 RunspaceStrings.NestedPipelineInvokeAsync);1104 }1105 1106 if (currentPipeline == null)1107 {1108 if (!_isSteppable)1109 {1110 throw PSTraceSource.NewInvalidOperationException(1111 RunspaceStrings.NestedPipelineNoParentPipeline);1112 }1113 }1114 }1115 }1116 }1117 1118 /// <summary>1119 /// The underlying powershell object on which this remote pipeline1120 /// is created.1121 /// </summary>1122 internal PowerShell PowerShell1123 {1124 get1125 {1126 return _powershell;1127 }1128 }1129 1130 /// <summary>1131 /// Sets the history string to the specified string.1132 /// </summary>1133 /// <param name="historyString">New history string to set to.</param>1134 internal override void SetHistoryString(string historyString)1135 {1136 _powershell.HistoryString = historyString;1137 }1138 1139 #endregion Internal Methods/Properties1140 1141 #region Remote data drain/block methods1142 1143 /// <summary>1144 /// Blocks data arriving from remote session.1145 /// </summary>1146 internal override void SuspendIncomingData()1147 {1148 _powershell.SuspendIncomingData();1149 }1150 1151 /// <summary>1152 /// Resumes data arrive from remote session.1153 /// </summary>1154 internal override void ResumeIncomingData()1155 {1156 _powershell.ResumeIncomingData();1157 }1158 1159 /// <summary>1160 /// Blocking call that waits until the current remote data1161 /// queue is empty.1162 /// </summary>1163 internal override void DrainIncomingData()1164 {1165 _powershell.WaitForServicingComplete();1166 }1167 1168 #endregion1169 }1170}1171