Method Communicate

util/system/shellcommand.cpp:847–1098

}

void TShellCommand::TImpl::Communicate(TProcessInfo* pi) {
    THolder<IOutputStream> outputHolder;
    IOutputStream* output = pi->Parent->Options_.OutputStream;
    if (!output) {
        outputHolder.Reset(output = new TStringOutput(pi->Parent->CollectedOutput));
    }

    THolder<IOutputStream> errorHolder;
    IOutputStream* error = pi->Parent->Options_.ErrorStream;
    if (!error) {
        errorHolder.Reset(error = new TStringOutput(pi->Parent->CollectedError));
    }

    IInputStream*& input = pi->Parent->Options_.InputStream;

#if defined(_unix_)
    // not really needed, io is done via poll
    if (pi->OutputFd.IsOpen()) {
        SetNonBlock(pi->OutputFd);
    }
    if (pi->ErrorFd.IsOpen()) {
        SetNonBlock(pi->ErrorFd);
    }
    if (pi->InputFd.IsOpen()) {
        SetNonBlock(pi->InputFd);
    }
#endif

    try {
#if defined(_win_)
        TPipePump pumps[3] = {0};
        pumps[0] = {&pi->ErrorFd, error};
        pumps[1] = {&pi->OutputFd, output};

        TVector<THolder<TThread>> streamThreads;
        streamThreads.emplace_back(new TThread(&TImpl::ReadStream, &pumps[0]));
        streamThreads.emplace_back(new TThread(&TImpl::ReadStream, &pumps[1]));

        if (input) {
            pumps[2] = {&pi->InputFd, nullptr, input, &pi->Parent->Options_.ShouldCloseInput};
            streamThreads.emplace_back(new TThread(&TImpl::WriteStream, &pumps[2]));
        }

        for (auto& threadHolder : streamThreads) {
            threadHolder->Start();
        }
#else
        TBuffer buffer(DATA_BUFFER_SIZE);
        TBuffer inputBuffer(DATA_BUFFER_SIZE);
        int bytes;
        int bytesToWrite = 0;
        char* bufPos = nullptr;
#endif
        TWaitResult waitPidResult;
        TExitStatus status = 0;

        while (true) {
            {