Sergey Mashkov | 86f7061 | 2017-07-25 10:59:44 +0300 | [diff] [blame] | 1 | package kotlinx.coroutines.experimental.io |
| 2 | |
Sergey Mashkov | 72e9150 | 2017-09-03 00:17:14 +0300 | [diff] [blame] | 3 | import kotlinx.coroutines.experimental.io.internal.* |
Sergey Mashkov | a22ed42 | 2017-09-05 19:24:30 +0300 | [diff] [blame] | 4 | import kotlinx.coroutines.experimental.io.packet.* |
Sergey Mashkov | 6d05993 | 2017-10-06 18:46:23 +0300 | [diff] [blame] | 5 | import kotlinx.coroutines.experimental.io.packet.ByteReadPacket |
| 6 | import kotlinx.io.core.* |
Sergey Mashkov | 86f7061 | 2017-07-25 10:59:44 +0300 | [diff] [blame] | 7 | |
Roman Elizarov | 469cad3 | 2017-08-15 15:54:56 +0300 | [diff] [blame] | 8 | /** |
| 9 | * Channel for asynchronous reading of sequences of bytes. |
| 10 | * This is a **single-reader channel**. |
| 11 | * |
| 12 | * Operations on this channel cannot be invoked concurrently. |
| 13 | */ |
| 14 | public interface ByteReadChannel { |
Sergey Mashkov | 86f7061 | 2017-07-25 10:59:44 +0300 | [diff] [blame] | 15 | /** |
Roman Elizarov | 469cad3 | 2017-08-15 15:54:56 +0300 | [diff] [blame] | 16 | * Returns number of bytes that can be read without suspension. Read operations do no suspend and return |
| 17 | * immediately when this number is at least the number of bytes requested for read. |
Sergey Mashkov | 86f7061 | 2017-07-25 10:59:44 +0300 | [diff] [blame] | 18 | */ |
Roman Elizarov | 469cad3 | 2017-08-15 15:54:56 +0300 | [diff] [blame] | 19 | public val availableForRead: Int |
Sergey Mashkov | 86f7061 | 2017-07-25 10:59:44 +0300 | [diff] [blame] | 20 | |
| 21 | /** |
Roman Elizarov | 469cad3 | 2017-08-15 15:54:56 +0300 | [diff] [blame] | 22 | * Returns `true` if the channel is closed and no remaining bytes are available for read. |
| 23 | * It implies that [availableForRead] is zero. |
Sergey Mashkov | 86f7061 | 2017-07-25 10:59:44 +0300 | [diff] [blame] | 24 | */ |
Roman Elizarov | 469cad3 | 2017-08-15 15:54:56 +0300 | [diff] [blame] | 25 | public val isClosedForRead: Boolean |
Sergey Mashkov | 86f7061 | 2017-07-25 10:59:44 +0300 | [diff] [blame] | 26 | |
Sergey Mashkov | 4cd0014 | 2017-09-05 19:25:32 +0300 | [diff] [blame] | 27 | public val isClosedForWrite: Boolean |
| 28 | |
Sergey Mashkov | 86f7061 | 2017-07-25 10:59:44 +0300 | [diff] [blame] | 29 | /** |
Roman Elizarov | 469cad3 | 2017-08-15 15:54:56 +0300 | [diff] [blame] | 30 | * Byte order that is used for multi-byte read operations |
| 31 | * (such as [readShort], [readInt], [readLong], [readFloat], and [readDouble]). |
| 32 | */ |
| 33 | public var readByteOrder: ByteOrder |
| 34 | |
| 35 | /** |
Sergey Mashkov | aae062d | 2017-09-07 15:12:50 +0300 | [diff] [blame] | 36 | * Number of bytes read from the channel. |
| 37 | * It is not guaranteed to be atomic so could be updated in the middle of long running read operation. |
| 38 | */ |
| 39 | public val totalBytesRead: Long |
| 40 | |
| 41 | /** |
Roman Elizarov | 469cad3 | 2017-08-15 15:54:56 +0300 | [diff] [blame] | 42 | * Reads all available bytes to [dst] buffer and returns immediately or suspends if no bytes available |
Sergey Mashkov | 86f7061 | 2017-07-25 10:59:44 +0300 | [diff] [blame] | 43 | * @return number of bytes were read or `-1` if the channel has been closed |
| 44 | */ |
Roman Elizarov | 469cad3 | 2017-08-15 15:54:56 +0300 | [diff] [blame] | 45 | suspend fun readAvailable(dst: ByteArray, offset: Int, length: Int): Int |
| 46 | suspend fun readAvailable(dst: ByteArray) = readAvailable(dst, 0, dst.size) |
| 47 | suspend fun readAvailable(dst: ByteBuffer): Int |
Sergey Mashkov | c4694cc | 2017-10-09 12:51:34 +0300 | [diff] [blame] | 48 | suspend fun readAvailable(dst: BufferView): Int |
Sergey Mashkov | 86f7061 | 2017-07-25 10:59:44 +0300 | [diff] [blame] | 49 | |
| 50 | /** |
| 51 | * Reads all [length] bytes to [dst] buffer or fails if channel has been closed. |
| 52 | * Suspends if not enough bytes available. |
| 53 | */ |
| 54 | suspend fun readFully(dst: ByteArray, offset: Int, length: Int) |
Sergey Mashkov | 86f7061 | 2017-07-25 10:59:44 +0300 | [diff] [blame] | 55 | suspend fun readFully(dst: ByteArray) = readFully(dst, 0, dst.size) |
Roman Elizarov | 469cad3 | 2017-08-15 15:54:56 +0300 | [diff] [blame] | 56 | suspend fun readFully(dst: ByteBuffer): Int |
Sergey Mashkov | 86f7061 | 2017-07-25 10:59:44 +0300 | [diff] [blame] | 57 | |
| 58 | /** |
Sergey Mashkov | a10b692 | 2017-08-03 13:41:54 +0300 | [diff] [blame] | 59 | * Reads the specified amount of bytes and makes a byte packet from them. Fails if channel has been closed |
Sergey Mashkov | 57906eb | 2017-09-19 14:08:06 +0300 | [diff] [blame] | 60 | * and not enough bytes available. Accepts [headerSizeHint] to be provided, see [WritePacket]. |
Sergey Mashkov | a10b692 | 2017-08-03 13:41:54 +0300 | [diff] [blame] | 61 | */ |
Sergey Mashkov | 57906eb | 2017-09-19 14:08:06 +0300 | [diff] [blame] | 62 | suspend fun readPacket(size: Int, headerSizeHint: Int = 0): ByteReadPacket |
Sergey Mashkov | a10b692 | 2017-08-03 13:41:54 +0300 | [diff] [blame] | 63 | |
| 64 | /** |
Sergey Mashkov | 86f7061 | 2017-07-25 10:59:44 +0300 | [diff] [blame] | 65 | * Reads a long number (suspending if not enough bytes available) or fails if channel has been closed |
| 66 | * and not enough bytes. |
| 67 | */ |
| 68 | suspend fun readLong(): Long |
| 69 | |
| 70 | /** |
| 71 | * Reads an int number (suspending if not enough bytes available) or fails if channel has been closed |
| 72 | * and not enough bytes. |
| 73 | */ |
| 74 | suspend fun readInt(): Int |
| 75 | |
| 76 | /** |
| 77 | * Reads a short number (suspending if not enough bytes available) or fails if channel has been closed |
| 78 | * and not enough bytes. |
| 79 | */ |
| 80 | suspend fun readShort(): Short |
| 81 | |
| 82 | /** |
| 83 | * Reads a byte (suspending if no bytes available yet) or fails if channel has been closed |
| 84 | * and not enough bytes. |
| 85 | */ |
| 86 | suspend fun readByte(): Byte |
| 87 | |
| 88 | /** |
| 89 | * Reads a boolean value (suspending if no bytes available yet) or fails if channel has been closed |
| 90 | * and not enough bytes. |
| 91 | */ |
| 92 | suspend fun readBoolean(): Boolean |
| 93 | |
| 94 | /** |
Sergey Mashkov | 86f7061 | 2017-07-25 10:59:44 +0300 | [diff] [blame] | 95 | * Reads double number (suspending if not enough bytes available) or fails if channel has been closed |
| 96 | * and not enough bytes. |
| 97 | */ |
| 98 | suspend fun readDouble(): Double |
| 99 | |
| 100 | /** |
| 101 | * Reads float number (suspending if not enough bytes available) or fails if channel has been closed |
| 102 | * and not enough bytes. |
| 103 | */ |
| 104 | suspend fun readFloat(): Float |
| 105 | |
Sergey Mashkov | 4cd0014 | 2017-09-05 19:25:32 +0300 | [diff] [blame] | 106 | /** |
| 107 | * For every available bytes range invokes [visitor] function until it return false or end of stream encountered |
| 108 | */ |
| 109 | suspend fun consumeEachBufferRange(visitor: (buffer: ByteBuffer, last: Boolean) -> Boolean) |
| 110 | |
| 111 | fun <R> lookAhead(visitor: LookAheadSession.() -> R): R |
| 112 | suspend fun <R> lookAheadSuspend(visitor: suspend LookAheadSuspendSession.() -> R): R |
Sergey Mashkov | 86f7061 | 2017-07-25 10:59:44 +0300 | [diff] [blame] | 113 | |
| 114 | /** |
| 115 | * Reads a line of UTF-8 characters to the specified [out] buffer up to [limit] characters. |
| 116 | * Supports both CR-LF and LF line endings. |
| 117 | * Throws an exception if the specified [limit] has been exceeded. |
| 118 | * |
| 119 | * @return `true` if line has been read (possibly empty) or `false` if channel has been closed |
| 120 | * and no characters were read. |
| 121 | */ |
| 122 | suspend fun <A : Appendable> readUTF8LineTo(out: A, limit: Int = Int.MAX_VALUE): Boolean |
Sergey Mashkov | 7510117 | 2017-09-25 19:41:08 +0300 | [diff] [blame] | 123 | |
| 124 | /** |
| 125 | * Invokes [block] when it will be possible to read at least [min] bytes |
| 126 | * providing byte buffer to it so lambda can read from the buffer |
| 127 | * up to [ByteBuffer.remaining] bytes. If there are no [min] bytes available then the invocation could |
| 128 | * suspend until the requirement will be met. |
| 129 | * |
Sergey Mashkov | e86eb08 | 2017-12-04 18:45:05 +0300 | [diff] [blame] | 130 | * If [min] is zero then the invocation will suspend until at least one byte available. |
| 131 | * |
Sergey Mashkov | 7510117 | 2017-09-25 19:41:08 +0300 | [diff] [blame] | 132 | * Warning: it is not guaranteed that all of remaining bytes will be represented as a single byte buffer |
| 133 | * eg: it could be 4 bytes available for read but the provided byte buffer could have only 2 remaining bytes: |
| 134 | * in this case you have to invoke read again (with decreased [min] accordingly). |
| 135 | * |
Sergey Mashkov | e86eb08 | 2017-12-04 18:45:05 +0300 | [diff] [blame] | 136 | * It will fail with [EOFException] if not enough bytes ([availableForRead] < [min]) available |
| 137 | * in the channel after it is closed. |
| 138 | * |
| 139 | * [block] lambda should modify buffer's position accordingly. It also could temporarily modify limit however |
| 140 | * it should restore it before return. It is not recommended to access any bytes of the buffer outside of the |
| 141 | * provided byte range [position(); limit()) as there could be any garbage or incomplete data. |
| 142 | * |
| 143 | * @param min amount of bytes available for read, should be positive or zero |
Sergey Mashkov | 7510117 | 2017-09-25 19:41:08 +0300 | [diff] [blame] | 144 | * @param block to be invoked when at least [min] bytes available for read |
| 145 | */ |
| 146 | suspend fun read(min: Int = 1, block: (ByteBuffer) -> Unit) |
Sergey Mashkov | eee3aaf | 2017-12-05 15:51:55 +0300 | [diff] [blame^] | 147 | |
| 148 | suspend fun cancel(cause: Throwable?): Boolean |
Roman Elizarov | 469cad3 | 2017-08-15 15:54:56 +0300 | [diff] [blame] | 149 | } |
| 150 | |
Sergey Mashkov | c51ecd3 | 2017-10-27 22:42:39 +0300 | [diff] [blame] | 151 | suspend fun ByteReadChannel.joinTo(dst: ByteWriteChannel, closeOnEnd: Boolean) { |
| 152 | require(dst !== this) |
| 153 | |
| 154 | if (this is ByteBufferChannel && dst is ByteBufferChannel) { |
| 155 | return dst.joinFrom(this, closeOnEnd) |
| 156 | } |
| 157 | |
Sergey Mashkov | f75ec15 | 2017-11-14 14:11:46 +0300 | [diff] [blame] | 158 | return joinToImplSuspend(dst, closeOnEnd) |
Sergey Mashkov | c51ecd3 | 2017-10-27 22:42:39 +0300 | [diff] [blame] | 159 | } |
| 160 | |
Sergey Mashkov | f75ec15 | 2017-11-14 14:11:46 +0300 | [diff] [blame] | 161 | private suspend fun ByteReadChannel.joinToImplSuspend(dst: ByteWriteChannel, close: Boolean) { |
Sergey Mashkov | fd85fb9 | 2017-11-08 22:27:18 +0300 | [diff] [blame] | 162 | copyToImpl(dst, Long.MAX_VALUE) |
Sergey Mashkov | c51ecd3 | 2017-10-27 22:42:39 +0300 | [diff] [blame] | 163 | if (close) { |
Sergey Mashkov | c51ecd3 | 2017-10-27 22:42:39 +0300 | [diff] [blame] | 164 | dst.close() |
Sergey Mashkov | fd85fb9 | 2017-11-08 22:27:18 +0300 | [diff] [blame] | 165 | } else { |
| 166 | dst.flush() |
Sergey Mashkov | c51ecd3 | 2017-10-27 22:42:39 +0300 | [diff] [blame] | 167 | } |
| 168 | } |
| 169 | |
Sergey Mashkov | 72e9150 | 2017-09-03 00:17:14 +0300 | [diff] [blame] | 170 | /** |
| 171 | * Reads up to [limit] bytes from receiver channel and writes them to [dst] channel. |
| 172 | * Closes [dst] channel if fails to read or write with cause exception. |
| 173 | * @return a number of copied bytes |
| 174 | */ |
| 175 | suspend fun ByteReadChannel.copyTo(dst: ByteWriteChannel, limit: Long = Long.MAX_VALUE): Long { |
Sergey Mashkov | 88951ac | 2017-10-27 22:36:08 +0300 | [diff] [blame] | 176 | require(this !== dst) |
| 177 | require(limit >= 0L) |
| 178 | |
| 179 | if (this is ByteBufferChannel && dst is ByteBufferChannel) { |
Sergey Mashkov | fd85fb9 | 2017-11-08 22:27:18 +0300 | [diff] [blame] | 180 | return dst.copyDirect(this, limit, null) |
Sergey Mashkov | 88951ac | 2017-10-27 22:36:08 +0300 | [diff] [blame] | 181 | } |
| 182 | |
| 183 | return copyToImpl(dst, limit) |
| 184 | } |
| 185 | |
| 186 | private suspend fun ByteReadChannel.copyToImpl(dst: ByteWriteChannel, limit: Long): Long { |
Sergey Mashkov | 72e9150 | 2017-09-03 00:17:14 +0300 | [diff] [blame] | 187 | val buffer = BufferPool.borrow() |
Sergey Mashkov | 88951ac | 2017-10-27 22:36:08 +0300 | [diff] [blame] | 188 | val dstNeedsFlush = !dst.autoFlush |
| 189 | |
Roman Elizarov | 469cad3 | 2017-08-15 15:54:56 +0300 | [diff] [blame] | 190 | try { |
| 191 | var copied = 0L |
Roman Elizarov | 469cad3 | 2017-08-15 15:54:56 +0300 | [diff] [blame] | 192 | |
Sergey Mashkov | 88951ac | 2017-10-27 22:36:08 +0300 | [diff] [blame] | 193 | while (true) { |
Sergey Mashkov | 72e9150 | 2017-09-03 00:17:14 +0300 | [diff] [blame] | 194 | buffer.clear() |
Sergey Mashkov | 88951ac | 2017-10-27 22:36:08 +0300 | [diff] [blame] | 195 | |
| 196 | val bufferLimit = limit - copied |
| 197 | if (bufferLimit <= 0) break |
| 198 | if (bufferLimit < buffer.limit()) { |
| 199 | buffer.limit(bufferLimit.toInt()) |
Sergey Mashkov | 72e9150 | 2017-09-03 00:17:14 +0300 | [diff] [blame] | 200 | } |
| 201 | val size = readAvailable(buffer) |
Roman Elizarov | 469cad3 | 2017-08-15 15:54:56 +0300 | [diff] [blame] | 202 | if (size == -1) break |
| 203 | |
Sergey Mashkov | 72e9150 | 2017-09-03 00:17:14 +0300 | [diff] [blame] | 204 | buffer.flip() |
| 205 | dst.writeFully(buffer) |
Roman Elizarov | 469cad3 | 2017-08-15 15:54:56 +0300 | [diff] [blame] | 206 | copied += size |
Sergey Mashkov | 88951ac | 2017-10-27 22:36:08 +0300 | [diff] [blame] | 207 | |
| 208 | if (dstNeedsFlush && availableForRead == 0) { |
| 209 | dst.flush() |
| 210 | } |
Roman Elizarov | 469cad3 | 2017-08-15 15:54:56 +0300 | [diff] [blame] | 211 | } |
Roman Elizarov | 469cad3 | 2017-08-15 15:54:56 +0300 | [diff] [blame] | 212 | return copied |
| 213 | } catch (t: Throwable) { |
| 214 | dst.close(t) |
| 215 | throw t |
| 216 | } finally { |
Sergey Mashkov | 72e9150 | 2017-09-03 00:17:14 +0300 | [diff] [blame] | 217 | BufferPool.recycle(buffer) |
Roman Elizarov | 469cad3 | 2017-08-15 15:54:56 +0300 | [diff] [blame] | 218 | } |
| 219 | } |
Sergey Mashkov | 7c8b155 | 2017-08-31 15:13:51 +0300 | [diff] [blame] | 220 | |
Sergey Mashkov | 72e9150 | 2017-09-03 00:17:14 +0300 | [diff] [blame] | 221 | /** |
| 222 | * Reads all the bytes from receiver channel and writes them to [dst] channel and then closes it. |
| 223 | * Closes [dst] channel if fails to read or write with cause exception. |
| 224 | * @return a number of copied bytes |
| 225 | */ |
Sergey Mashkov | 7c8b155 | 2017-08-31 15:13:51 +0300 | [diff] [blame] | 226 | suspend fun ByteReadChannel.copyAndClose(dst: ByteWriteChannel): Long { |
| 227 | val count = copyTo(dst) |
| 228 | dst.close() |
| 229 | return count |
| 230 | } |
Sergey Mashkov | a22ed42 | 2017-09-05 19:24:30 +0300 | [diff] [blame] | 231 | |
| 232 | /** |
| 233 | * Reads all the bytes from receiver channel and builds a packet that is returned unless the specified [limit] exceeded. |
| 234 | * It will simply stop reading and return packet of size [limit] in this case |
| 235 | */ |
Sergey Mashkov | 2c3a0dc | 2017-09-06 22:58:08 +0300 | [diff] [blame] | 236 | suspend fun ByteReadChannel.readRemaining(limit: Int = Int.MAX_VALUE): ByteReadPacket { |
Sergey Mashkov | a22ed42 | 2017-09-05 19:24:30 +0300 | [diff] [blame] | 237 | val buffer = BufferPool.borrow() |
| 238 | val packet = WritePacket() |
| 239 | |
| 240 | try { |
| 241 | var copied = 0L |
| 242 | |
| 243 | while (copied < limit) { |
| 244 | buffer.clear() |
| 245 | if (limit - copied < buffer.limit()) { |
| 246 | buffer.limit((limit - copied).toInt()) |
| 247 | } |
| 248 | val size = readAvailable(buffer) |
| 249 | if (size == -1) break |
| 250 | |
| 251 | buffer.flip() |
| 252 | packet.writeFully(buffer) |
| 253 | copied += size |
| 254 | } |
| 255 | |
| 256 | return packet.build() |
| 257 | } catch (t: Throwable) { |
| 258 | packet.release() |
| 259 | throw t |
| 260 | } finally { |
| 261 | BufferPool.recycle(buffer) |
| 262 | } |
| 263 | } |