Check whether we have reached the end of a stream. Although it is more effective to read and catch exceptions, this function Parameters ---------- substrate: :py:class:`IOBase` Stream to check Returns ------- : :py:class:`bool`
(substrate)
| 116 | |
| 117 | |
| 118 | def isEndOfStream(substrate): |
| 119 | """Check whether we have reached the end of a stream. |
| 120 | |
| 121 | Although it is more effective to read and catch exceptions, this |
| 122 | function |
| 123 | |
| 124 | Parameters |
| 125 | ---------- |
| 126 | substrate: :py:class:`IOBase` |
| 127 | Stream to check |
| 128 | |
| 129 | Returns |
| 130 | ------- |
| 131 | : :py:class:`bool` |
| 132 | """ |
| 133 | if isinstance(substrate, io.BytesIO): |
| 134 | cp = substrate.tell() |
| 135 | substrate.seek(0, os.SEEK_END) |
| 136 | result = substrate.tell() == cp |
| 137 | substrate.seek(cp, os.SEEK_SET) |
| 138 | yield result |
| 139 | |
| 140 | else: |
| 141 | received = substrate.read(1) |
| 142 | if received is None: |
| 143 | yield |
| 144 | |
| 145 | if received: |
| 146 | substrate.seek(-1, os.SEEK_CUR) |
| 147 | |
| 148 | yield not received |
| 149 | |
| 150 | |
| 151 | def peekIntoStream(substrate, size=-1): |
no test coverage detected