Return a list of (recv, send) for all open channels.
()
| 66 | |
| 67 | |
| 68 | def list_all(): |
| 69 | """Return a list of (recv, send) for all open channels.""" |
| 70 | channels = [] |
| 71 | for cid, unboundop, _ in _channels.list_all(): |
| 72 | chan = _, send = RecvChannel(cid), SendChannel(cid) |
| 73 | if not hasattr(send, '_unboundop'): |
| 74 | send._set_unbound(unboundop) |
| 75 | else: |
| 76 | assert send._unbound[0] == unboundop |
| 77 | channels.append(chan) |
| 78 | return channels |
| 79 | |
| 80 | |
| 81 | class _ChannelEnd: |
nothing calls this directly
no test coverage detected