Run a simple echo server.
(self, context, reader, writer)
| 497 | writer.write_metadata(str(num_batches).encode("utf-8")) |
| 498 | |
| 499 | def exchange_echo(self, context, reader, writer): |
| 500 | """Run a simple echo server.""" |
| 501 | assert reader.stats.num_messages == 0 |
| 502 | started = False |
| 503 | for chunk in reader: |
| 504 | if not started and chunk.data: |
| 505 | writer.begin(chunk.data.schema, options=self.options) |
| 506 | started = True |
| 507 | if chunk.app_metadata and chunk.data: |
| 508 | writer.write_with_metadata(chunk.data, chunk.app_metadata) |
| 509 | elif chunk.app_metadata: |
| 510 | writer.write_metadata(chunk.app_metadata) |
| 511 | elif chunk.data: |
| 512 | assert reader.stats.num_messages != 0 |
| 513 | writer.write_batch(chunk.data) |
| 514 | else: |
| 515 | assert False, "Should not happen" |
| 516 | |
| 517 | def exchange_transform(self, context, reader, writer): |
| 518 | """Sum rows in an uploaded table.""" |
no test coverage detected