pipe.cpp

#include "headers.h"

// Execute cmd1 | cmd2 | ... | cmdN. All stages run concurrently, connected by
// pipes, and the shell waits for every stage before returning.
void executePipes(vector<string> args)
{
    // Collect the individual command strings (drop the "|" separators).
    vector<string> commands;
    for (const string &a : args)
        if (a != "|")
        {
            string c = a;
            size_t s = c.find_first_not_of(" \t");
            size_t e = c.find_last_not_of(" \t");
            if (s != string::npos)
                commands.push_back(c.substr(s, e - s + 1));
        }

    if (commands.empty())
        return;

    int prevRead = -1; // read end of the previous pipe
    vector<pid_t> children;

    for (size_t i = 0; i < commands.size(); ++i)
    {
        bool last = (i + 1 == commands.size());
        int fds[2] = {-1, -1};

        if (!last && pipe(fds) == -1)
        {
            perror("pipe");
            break;
        }

        const string &command = commands[i];
        bool hasRedirection =
            command.find('>') != string::npos || command.find('<') != string::npos;

        vector<string> tokens =
            hasRedirection ? tokenizeForRedirection(command) : tokenize(command);
        if (tokens.empty())
        {
            if (!last)
            {
                close(fds[0]);
                close(fds[1]);
            }
            continue;
        }

        pid_t pid = fork();
        if (pid == -1)
        {
            perror("fork");
            if (!last)
            {
                close(fds[0]);
                close(fds[1]);
            }
            break;
        }

        if (pid == 0)
        {
            signal(SIGINT, SIG_DFL);
            signal(SIGTSTP, SIG_DFL);

            if (prevRead != -1)
            {
                dup2(prevRead, STDIN_FILENO);
                close(prevRead);
            }
            if (!last)
            {
                close(fds[0]);
                dup2(fds[1], STDOUT_FILENO);
                close(fds[1]);
            }

            if (hasRedirection)
            {
                redirection(tokens);
                _exit(0);
            }

            char **argv = new char *[tokens.size() + 1];
            for (size_t k = 0; k < tokens.size(); ++k)
            {
                argv[k] = new char[tokens[k].size() + 1];
                strcpy(argv[k], tokens[k].c_str());
            }
            argv[tokens.size()] = nullptr;

            execvp(argv[0], argv);
            cerr << tokens[0] << ": command not found" << endl;
            _exit(127);
        }

        children.push_back(pid);

        if (prevRead != -1)
            close(prevRead);
        if (!last)
        {
            close(fds[1]);
            prevRead = fds[0];
        }
    }

    if (prevRead != -1)
        close(prevRead);

    for (pid_t pid : children)
    {
        int status = 0;
        while (waitpid(pid, &status, 0) == -1 && errno == EINTR)
            ;
    }
}