(self, session_id, dag_def)
| 91 | return chunks_list |
| 92 | |
| 93 | def generate_runstep_requests(self, session_id, dag_def): |
| 94 | runstep_requests = [] |
| 95 | chunks_list = self.split(dag_def) |
| 96 | # head |
| 97 | runstep_request = message_pb2.RunStepRequest( |
| 98 | head=message_pb2.RunStepRequestHead(session_id=session_id, dag_def=dag_def) |
| 99 | ) |
| 100 | runstep_requests.append(runstep_request) |
| 101 | # bodies |
| 102 | for chunks, op_key in chunks_list: |
| 103 | for i, chunk in enumerate(chunks): |
| 104 | # check the last element |
| 105 | has_next = True |
| 106 | if i + 1 == len(chunks): |
| 107 | has_next = False |
| 108 | runstep_request = message_pb2.RunStepRequest( |
| 109 | body=message_pb2.RunStepRequestBody( |
| 110 | chunk=chunk, op_key=op_key, has_next=has_next |
| 111 | ) |
| 112 | ) |
| 113 | runstep_requests.append(runstep_request) |
| 114 | # return a generator for stream request |
| 115 | for item in runstep_requests: |
| 116 | yield item |
| 117 | |
| 118 | def parse_runstep_responses(self, responses): |
| 119 | chunks = [] |
no test coverage detected