unless you change the dispatcher of emmit

# coroutines

[myanmarking]

unless you change the dispatcher of emmit

[dave08]

What do you mean by this?

[myanmarking]

with flow, you cannot emit events in another dispatcher. So if you do that, it will crash

[myanmarking]

fun doSomething() = callbackFlow{
    withContext(<http://Dispatchers.IO|Dispatchers.IO>){
        offer(Unit)
    }
}

[myanmarking]

you cannot do this

[dave08]

No, I don't do that (I use channelFlow for that...), but I still have a `okio.SocketAsyncTimeout.newTimeoutException (SocketAsyncTimeout.java:159)`that's crashing the app ...

[myanmarking]

without seeing the code its hard to guess

[dave08]

I thought callback flow is for that...

[dave08]

yup, these things are a bit hard to pinpoint, I admit. It's crashing on a timeout here:

private fun Call.downloadAndSaveTo(
			output: Sink,
			bufferSize: Long = DEFAULT_BUFFER_SIZE.toLong() * 10,
			flushSize: Long = bufferSize * 100
	)Flow<Long> = callbackFlow<Long> {
		enqueue(object : Callback {
			override fun onFailure(call: Call?, e: IOException) {
				[e-mail](e)
			}

			override fun onResponse(call: Call?, response: Response) {
				if (!response.isSuccessful) {
					[e-mail](IllegalStateException("Unexpected HTTP code: ${response.code()}"))
				}

				try {
					response.body()?.use { body ->
						val contentLength = body.contentLength()
						val buffer = Buffer()
						var finished = false
						var totalLength = 0L

						var lastFlush = 0L

						output.buffer().use { out ->
							body.source().use { source ->
								while (isActive) {
									val read = source.read(buffer, bufferSize)
									if (read == -1L) {
										out.flush()

										finished = true
										break
									}

									out.write(buffer, read)
									totalLength += read

									if ((totalLength - lastFlush) / flushSize > 0) {
										out.flush()
										lastFlush = totalLength
									} else
										out.emit()

									offer(read)
								}
							}
						}
					}

					channel.close()
				} catch (e: Exception) {
					[e-mail](e)
				}
			}
		})

		awaitClose {
			// Cancel the call
			[e-mail]()

			// Close other resources..
		}
	}.flowOn(<http://Dispatchers.IO|Dispatchers.IO>)

[myanmarking]

ok give me a sec

[dave08]

I think on `

val read = source.read(buffer, bufferSize)

[myanmarking]

hm you dont need the try catch i think

[myanmarking]

the only place i see a problem, is awaitClose

[dave08]

What do you see wrong with the awaitClose?

[myanmarking]

wait i deleted the comment lol

[myanmarking]

you read it right 😛 ?

[myanmarking]

if there is a crash inside awaitClose, im not sure it wont be thrown

[dave08]

Right, could be, but all I have is close there... 🤔