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