| 85 | } |
| 86 | |
| 87 | void StreamingExchangeSource::receiveHello() |
| 88 | { |
| 89 | StreamingExchangeProtocol::PacketHeader header{}; |
| 90 | size_t position = 0; |
| 91 | readFromSocket(reinterpret_cast<char *>(&header), sizeof(header), position); |
| 92 | if (position != sizeof(header)) |
| 93 | throw Poco::Net::NetException(fmt::format( |
| 94 | "Failed to receive SinkHello header from socket for exchange stream {}, expected {} bytes but received {}", |
| 95 | stream_name, sizeof(header), position)); |
| 96 | |
| 97 | if (header.packet_type != StreamingExchangeProtocol::PacketType::SinkHello) |
| 98 | throw Exception(ErrorCodes::UNEXPECTED_PACKET_FROM_CLIENT, |
| 99 | "Unexpected packet type 0x{:x} (expected SinkHello 0x{:x}) for exchange stream {}", |
| 100 | header.packet_type, static_cast<UInt64>(StreamingExchangeProtocol::PacketType::SinkHello), stream_name); |
| 101 | |
| 102 | if (header.bytes_size > StreamingExchangeProtocol::MAX_HELLO_BODY_BYTES) |
| 103 | throw Exception(ErrorCodes::UNEXPECTED_PACKET_FROM_CLIENT, |
| 104 | "SinkHello body size {} exceeds the limit {} for exchange stream {}", |
| 105 | header.bytes_size, StreamingExchangeProtocol::MAX_HELLO_BODY_BYTES, stream_name); |
| 106 | |
| 107 | if (header.bytes_size < sizeof(UInt64)) |
| 108 | throw Exception(ErrorCodes::UNEXPECTED_PACKET_FROM_CLIENT, |
| 109 | "SinkHello body size {} is too small to contain the protocol version for exchange stream {}", |
| 110 | header.bytes_size, stream_name); |
| 111 | |
| 112 | PODArray<char> body_buffer(header.bytes_size); |
| 113 | size_t body_position = 0; |
| 114 | readFromSocket(body_buffer.data(), body_buffer.size(), body_position); |
| 115 | if (body_position != body_buffer.size()) |
| 116 | throw Poco::Net::NetException(fmt::format( |
| 117 | "Failed to receive SinkHello body from socket for exchange stream {}, expected {} bytes but received {}", |
| 118 | stream_name, body_buffer.size(), body_position)); |
| 119 | |
| 120 | ReadBufferFromMemory body_in(body_buffer.data(), body_buffer.size()); |
| 121 | StreamingExchangeProtocol::SinkHelloBody sink_hello; |
| 122 | sink_hello.read(body_in); |
| 123 | |
| 124 | if (sink_hello.sink_version != StreamingExchangeProtocol::PROTOCOL_VERSION) |
| 125 | throw Exception(ErrorCodes::PROTOCOL_VERSION_MISMATCH, |
| 126 | "Streaming exchange protocol version mismatch for stream {}: this node speaks version {}, sink at {}:{} speaks version {}", |
| 127 | stream_name, StreamingExchangeProtocol::PROTOCOL_VERSION, host, port, sink_hello.sink_version); |
| 128 | } |
| 129 | |
| 130 | IProcessor::Status StreamingExchangeSource::prepare() |
| 131 | { |