1#include <Processors/Port.h>
2#include <Processors/IProcessor.h>
3
4namespace DB
5{
6
7void connect(OutputPort & output, InputPort & input)
8{
9 if (input.state || output.state)
10 throw Exception("Port is already connected", ErrorCodes::LOGICAL_ERROR);
11
12 auto out_name = output.getProcessor().getName();
13 auto in_name = input.getProcessor().getName();
14
15 assertBlocksHaveEqualStructure(input.getHeader(), output.getHeader(), " function connect between " + out_name + " and " + in_name);
16
17 input.output_port = &output;
18 output.input_port = &input;
19 input.state = std::make_shared<Port::State>();
20 output.state = input.state;
21}
22
23}
24