Fix storing cache entities

Use ByteArrayContent instead of WriteChannelContent when executing HTTP PUT requests

core/src/main/kotlin/com/malinskiy/marathon/cache/CacheEntryWriter.kt

@@ -1,7 +1,7 @@
 package com.malinskiy.marathon.cache
 
-import io.ktor.utils.io.*
+import java.io.OutputStream
 
 interface CacheEntryWriter {
-    suspend fun writeTo(output: ByteWriteChannel)
+    fun writeTo(output: OutputStream)
 }

core/src/main/kotlin/com/malinskiy/marathon/cache/gradle/GradleHttpCacheService.kt

@@ -5,17 +5,18 @@ import com.malinskiy.marathon.cache.CacheEntryWriter
 import com.malinskiy.marathon.cache.CacheKey
 import com.malinskiy.marathon.cache.CacheService
 import com.malinskiy.marathon.cache.config.RemoteCacheConfiguration
+import com.malinskiy.marathon.log.MarathonLogging
 import io.ktor.client.*
 import io.ktor.client.engine.apache.*
 import io.ktor.client.plugins.auth.*
 import io.ktor.client.plugins.auth.providers.*
 import io.ktor.client.request.*
 import io.ktor.client.statement.*
+import io.ktor.content.*
 import io.ktor.http.*
-import io.ktor.http.content.*
-import io.ktor.utils.io.*
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.withContext
+import java.io.ByteArrayOutputStream
 import java.io.IOException
 import java.net.URI
 import java.net.URL
@@ -25,33 +26,38 @@ class GradleHttpCacheService(private val configuration: RemoteCacheConfiguration
     private val httpClient = createClient()
     private val baseUri = URI.create(configuration.url)
 
+    private val logger = MarathonLogging.logger("GradleHttpCacheService")
+
     override suspend fun load(key: CacheKey, reader: CacheEntryReader): Boolean =
         withContext(Dispatchers.IO) {
             try {
                 val response = httpClient.get(url = key.entryUrl())
                 if (response.status != HttpStatusCode.OK) {
+                    logger.warn("Got response status when loading cache entry for ${key.key} : ${response.status}")
                     false
                 } else {
                     reader.readFrom(response.bodyAsChannel())
                     true
                 }
             } catch (exception: IOException) {
+                logger.warn("Error during loading cache entry for ${key.key}", exception)
                 false
             }
         }
 
     override suspend fun store(key: CacheKey, writer: CacheEntryWriter) {
         withContext(Dispatchers.IO) {
+            val stream = ByteArrayOutputStream()
             try {
-                httpClient.put(url = key.entryUrl()) {
-                    setBody(object : OutgoingContent.WriteChannelContent() {
-                        override suspend fun writeTo(channel: ByteWriteChannel) {
-                            writer.writeTo(channel)
-                        }
-                    })
+                writer.writeTo(stream)
+                val response = httpClient.put(url = key.entryUrl()) { setBody(ByteArrayContent(stream.toByteArray())) }
+                if (!response.status.isSuccess()) {
+                    logger.warn("Got response status when storing cache entry for ${key.key} with ${key.entryUrl()} : ${response.status}")
                 }
             } catch (exception: IOException) {
-                // ignore
+                logger.warn("Error during storing cache entry for ${key.key}", exception)
+            } finally {
+                stream.close()
             }
         }
     }

core/src/main/kotlin/com/malinskiy/marathon/cache/test/serialization/TestResultEntryWriter.kt

@@ -4,45 +4,42 @@ import com.malinskiy.marathon.cache.CacheEntryWriter
 import com.malinskiy.marathon.device.DeviceInfo
 import com.malinskiy.marathon.execution.Attachment
 import com.malinskiy.marathon.execution.TestResult
-import io.ktor.util.cio.*
-import io.ktor.utils.io.*
 import io.ktor.utils.io.core.*
+import io.ktor.utils.io.streams.*
+import java.io.DataOutputStream
 import java.io.File
+import java.io.OutputStream
 
 class TestResultEntryWriter(private val testResult: TestResult) : CacheEntryWriter {
 
-    override suspend fun writeTo(output: ByteWriteChannel) {
-        output.writeDeviceInfo(testResult.device)
-        output.writeInt(testResult.status.ordinal)
-        output.writeLong(testResult.startTime)
-        output.writeLong(testResult.endTime)
-        output.writeString(testResult.batchId)
-        output.writeString(testResult.stacktrace)
+    override fun writeTo(output: OutputStream) {
+        DataOutputStream(output).use {
+            it.writeDeviceInfo(testResult.device)
+            it.writeInt(testResult.status.ordinal)
+            it.writeLong(testResult.startTime)
+            it.writeLong(testResult.endTime)
+            it.writeString(testResult.batchId)
+            it.writeString(testResult.stacktrace)
 
-        output.writeInt(testResult.attachments.size)
-        testResult.attachments.forEach {
-            output.writeAttachment(it)
+            it.writeInt(testResult.attachments.size)
+            testResult.attachments.forEach { attachment ->
+                it.writeAttachment(attachment)
+            }
         }
     }
 
-    private suspend fun ByteWriteChannel.writeAttachment(attachment: Attachment) {
+    private fun DataOutputStream.writeAttachment(attachment: Attachment) {
         writeInt(attachment.type.ordinal)
         writeInt(attachment.fileType.ordinal)
         writeFile(attachment.file)
     }
 
-    private suspend fun ByteWriteChannel.writeFile(file: File) {
-        val readChannel = file.readChannel()
-        try {
-            val fileLength = file.length()
-            writeLong(file.length())
-            readChannel.copyTo(this, limit = fileLength)
-        } finally {
-            readChannel.cancel()
-        }
+    private fun DataOutputStream.writeFile(file: File) {
+        writeLong(file.length())
+        write(file.readBytes())
     }
 
-    private suspend fun ByteWriteChannel.writeDeviceInfo(deviceInfo: DeviceInfo) {
+    private fun DataOutputStream.writeDeviceInfo(deviceInfo: DeviceInfo) {
         writeString(deviceInfo.operatingSystem.version)
         writeString(deviceInfo.serialNumber)
         writeString(deviceInfo.model)
@@ -52,14 +49,14 @@ class TestResultEntryWriter(private val testResult: TestResult) : CacheEntryWrit
         writeBoolean(deviceInfo.healthy)
     }
 
-    private suspend fun ByteWriteChannel.writeEnumCollection(collection: Collection<Enum<*>>) {
+    private fun DataOutputStream.writeEnumCollection(collection: Collection<Enum<*>>) {
         writeInt(collection.size)
         collection.forEach {
             writeInt(it.ordinal)
         }
     }
 
-    private suspend fun ByteWriteChannel.writeString(str: String?) {
+    private fun DataOutputStream.writeString(str: String?) {
         if (str == null) {
             writeBoolean(false)
             return

core/src/test/kotlin/com/malinskiy/marathon/cache/MemoryCacheService.kt

@@ -2,7 +2,7 @@ package com.malinskiy.marathon.cache
 
 import io.ktor.utils.io.*
 import io.ktor.utils.io.core.*
-import java.lang.RuntimeException
+import java.io.ByteArrayOutputStream
 
 class MemoryCacheService : CacheService {
 
@@ -20,10 +20,10 @@ class MemoryCacheService : CacheService {
     override suspend fun store(key: CacheKey, writer: CacheEntryWriter) {
         throwable?.let { throw it }
 
-        val channel = ByteChannel()
-        writer.writeTo(channel)
-        channel.close()
-        cache[key] = channel.readRemaining().readBytes()
+        ByteArrayOutputStream().use {
+            writer.writeTo(it)
+            cache[key] = it.toByteArray()
+        }
     }
 
     override fun close() {

core/src/test/kotlin/com/malinskiy/marathon/cache/SimpleEntryWriter.kt

@@ -1,13 +1,13 @@
 package com.malinskiy.marathon.cache
 
-import io.ktor.utils.io.*
+import java.io.OutputStream
 
 class SimpleEntryWriter(data: String) : CacheEntryWriter {
 
     private val bytes = data.toByteArray()
 
-    override suspend fun writeTo(output: ByteWriteChannel) {
-        output.writeFully(bytes)
+    override fun writeTo(output: OutputStream) {
+       output.write(bytes)
     }
 
 }