Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ import com.segment.analytics.kotlin.core.platform.plugins.logger.log
import com.segment.analytics.kotlin.core.utilities.LenientJson
import com.segment.analytics.kotlin.core.utilities.safeJsonObject
import kotlinx.coroutines.launch
import kotlinx.coroutines.runInterruptible
import kotlinx.coroutines.withContext
import kotlinx.serialization.DeserializationStrategy
import kotlinx.serialization.Serializable
Expand Down Expand Up @@ -89,7 +90,7 @@ suspend fun Analytics.checkSettings() {

val settingsObj = withContext(networkIODispatcher) {
log("Fetching settings on ${Thread.currentThread().name}")
return@withContext fetchSettings(writeKey, cdnHost)
return@withContext runInterruptible { fetchSettings(writeKey, cdnHost) }
}

settingsObj?.let {
Expand Down
20 changes: 12 additions & 8 deletions core/src/main/java/com/segment/analytics/kotlin/core/Telemetry.kt
Original file line number Diff line number Diff line change
Expand Up @@ -263,15 +263,19 @@ object Telemetry: Subscriber {
// We're using this to leave off the 'log' parameter if unset.
val payload = Json.encodeToString(mapOf("series" to sendQueue))

val connection = httpClient.upload(host)
connection.outputStream?.use { outputStream ->
// Write the JSON string to the outputStream.
outputStream.write(payload.toByteArray(Charsets.UTF_8))
outputStream.flush() // Ensure all data is written
runBlocking {
runInterruptible {
val connection = httpClient.upload(host)
connection.outputStream?.use { outputStream ->
// Write the JSON string to the outputStream.
outputStream.write(payload.toByteArray(Charsets.UTF_8))
outputStream.flush() // Ensure all data is written
}
connection.inputStream?.close()
connection.outputStream?.close()
connection.close()
}
}
connection.inputStream?.close()
connection.outputStream?.close()
connection.close()
} catch (e: HTTPException) {
errorHandler?.invoke(e)
if (e.responseCode == 429) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED
import kotlinx.coroutines.channels.consumeEach
import kotlinx.coroutines.launch
import kotlinx.coroutines.runInterruptible
import kotlinx.coroutines.withContext
import kotlinx.serialization.encodeToString
import kotlinx.serialization.json.Json
Expand Down Expand Up @@ -135,14 +136,16 @@ open class EventPipeline(
var shouldCleanup = true
storage.readAsStream(url)?.use { data ->
try {
val connection = httpClient.upload(apiHost)
connection.outputStream?.let {
// Write the payloads into the OutputStream
data.copyTo(connection.outputStream)
connection.outputStream.close()

// Upload the payloads.
connection.close()
runInterruptible {
val connection = httpClient.upload(apiHost)
connection.outputStream?.let {
// Write the payloads into the OutputStream
data.copyTo(connection.outputStream)
connection.outputStream.close()

// Upload the payloads.
connection.close()
}
}
// Cleanup uploaded payloads
analytics.log("$logTag uploaded $url")
Expand Down
Loading