| 179 | */ |
| 180 | |
| 181 | Block readImpl() override |
| 182 | { |
| 183 | if (all_read) |
| 184 | return received_payload.block; |
| 185 | |
| 186 | /// Run threads if this has not already been done. |
| 187 | if (!started) |
| 188 | { |
| 189 | started = true; |
| 190 | processor.process(); |
| 191 | } |
| 192 | |
| 193 | /// We will wait until the next block is ready or an exception is thrown. |
| 194 | //std::cerr << "popping\n"; |
| 195 | if (!output_queue.pop(received_payload)) |
| 196 | throw Exception(ErrorCodes::LOGICAL_ERROR, "Could not pop output_data"); |
| 197 | |
| 198 | if (received_payload.exception) |
| 199 | { |
| 200 | if (exception_callback) |
| 201 | exception_callback(); |
| 202 | std::rethrow_exception(received_payload.exception); |
| 203 | } |
| 204 | |
| 205 | if (!received_payload.block) |
| 206 | all_read = true; |
| 207 | |
| 208 | return received_payload.block; |
| 209 | } |
| 210 | |
| 211 | /// Called either after everything is read, or after cancel. |
| 212 | void readSuffix() override |