(self, data)
| 197 | self.close() |
| 198 | |
| 199 | def push(self, data): |
| 200 | if not isinstance(data, (bytes, bytearray, memoryview)): |
| 201 | raise TypeError('data argument must be byte-ish (%r)', |
| 202 | type(data)) |
| 203 | sabs = self.ac_out_buffer_size |
| 204 | if len(data) > sabs: |
| 205 | for i in range(0, len(data), sabs): |
| 206 | self.producer_fifo.append(data[i:i+sabs]) |
| 207 | else: |
| 208 | self.producer_fifo.append(data) |
| 209 | self.initiate_send() |
| 210 | |
| 211 | def push_with_producer(self, producer): |
| 212 | self.producer_fifo.append(producer) |
nothing calls this directly
no test coverage detected