Method prepare

src/Processors/Sources/DelayedSource.cpp:32–86

}

IProcessor::Status DelayedSource::prepare()
{
    /// At first, wait for main output is needed and expand pipeline.
    if (inputs.empty())
    {
        auto & first_output = outputs.front();

        /// If main output port was finished before callback was called, stop execution.
        if (first_output.isFinished())
        {
            for (auto & output : outputs)
                output.finish();

            return Status::Finished;
        }

        if (!first_output.isNeeded())
            return Status::PortFull;

        /// Call creator callback to get processors.
        if (processors.empty())
            return Status::Ready;

        return Status::UpdatePipeline;
    }

    /// Process ports in order: main, totals, extremes
    auto output = outputs.begin();
    for (auto input = inputs.begin(); input != inputs.end(); ++input, ++output)
    {
        if (output->isFinished())
        {
            input->close();
            continue;
        }

        if (!output->canPush())
            return Status::PortFull;

        if (input->isFinished())
        {
            output->finish();
            continue;
        }

        input->setNeeded();
        if (!input->hasData())
            return Status::NeedData;

        output->pushData(input->pullData(true));
        return Status::PortFull;
    }

    return Status::Finished;
}

/// Fix port from returned pipe. Create source_port if created or drop if source_port is null.
static void synchronizePorts(OutputPort *& pipe_port, OutputPort * source_port, SharedHeader header, Processors & processors)