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