Skip to content

Commit e0afd44

Browse files
polbinspolbins
authored
refactor: remove-previous-retry-logic (#258)
* fix warnings on file storage and add KDocs * remove previous retry logic - see changes original PR at #113 - we're going to update the retry logic in a separate PR * fix ktlint * fix failing test * inline sorting function inside read()
1 parent a49b6a7 commit e0afd44

File tree

7 files changed

+151
-135
lines changed

7 files changed

+151
-135
lines changed

‎android/src/main/java/com/amplitude/android/storage/AndroidStorageV2.kt

+11-5
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@ package com.amplitude.android.storage
22

33
import android.content.Context
44
import android.content.SharedPreferences
5+
import androidx.core.content.edit
56
import com.amplitude.android.utilities.AndroidKVS
67
import com.amplitude.common.Logger
78
import com.amplitude.core.Amplitude
@@ -62,11 +63,15 @@ class AndroidStorageV2(
6263
key: Storage.Constants,
6364
value: String,
6465
) {
65-
sharedPreferences.edit().putString(key.rawVal, value).apply()
66+
sharedPreferences.edit {
67+
putString(key.rawVal, value)
68+
}
6669
}
6770

6871
override suspend fun remove(key: Storage.Constants) {
69-
sharedPreferences.edit().remove(key.rawVal).apply()
72+
sharedPreferences.edit {
73+
remove(key.rawVal)
74+
}
7075
}
7176

7277
override suspend fun rollover() {
@@ -85,8 +90,8 @@ class AndroidStorageV2(
8590
eventsFile.release(filePath)
8691
}
8792

88-
override suspend fun getEventsString(content: Any): String {
89-
return eventsFile.getEventString(content as String)
93+
override suspend fun getEventsString(filePath: Any): String {
94+
return eventsFile.getEventString(filePath as String)
9095
}
9196

9297
override fun getResponseHandler(
@@ -136,7 +141,8 @@ class AndroidEventsStorageProviderV2 : StorageProvider {
136141
): Storage {
137142
val configuration = amplitude.configuration as com.amplitude.android.Configuration
138143
val sharedPreferencesName = "amplitude-events-${configuration.instanceName}"
139-
val sharedPreferences = configuration.context.getSharedPreferences(sharedPreferencesName, Context.MODE_PRIVATE)
144+
val sharedPreferences =
145+
configuration.context.getSharedPreferences(sharedPreferencesName, Context.MODE_PRIVATE)
140146
return AndroidStorageV2(
141147
configuration.instanceName,
142148
configuration.loggerProvider.getLogger(amplitude),

‎android/src/main/java/com/amplitude/android/utilities/AndroidStorage.kt

+2-2
Original file line numberDiff line numberDiff line change
@@ -74,8 +74,8 @@ class AndroidStorage(
7474
storageV2.releaseFile(filePath)
7575
}
7676

77-
override suspend fun getEventsString(content: Any): String {
78-
return storageV2.getEventsString(content)
77+
override suspend fun getEventsString(filePath: Any): String {
78+
return storageV2.getEventsString(filePath)
7979
}
8080

8181
override fun getResponseHandler(

‎core/src/main/java/com/amplitude/core/Amplitude.kt

+2-2
Original file line numberDiff line numberDiff line change
@@ -156,10 +156,10 @@ open class Amplitude internal constructor(
156156
options: EventOptions? = null,
157157
callback: EventCallBack? = null,
158158
): Amplitude {
159-
options ?. let {
159+
options?.let {
160160
event.mergeEventOptions(it)
161161
}
162-
callback ?. let {
162+
callback?.let {
163163
event.callback = it
164164
}
165165
process(event)

‎core/src/main/java/com/amplitude/core/platform/EventPipeline.kt

+11-25
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,6 @@ import com.amplitude.core.Amplitude
44
import com.amplitude.core.events.BaseEvent
55
import com.amplitude.core.utilities.http.HttpClient
66
import com.amplitude.core.utilities.http.HttpClientInterface
7-
import com.amplitude.core.utilities.http.ResponseHandler
87
import com.amplitude.core.utilities.logWithStackTrace
98
import kotlinx.coroutines.channels.Channel
109
import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED
@@ -20,35 +19,30 @@ class EventPipeline(
2019
private val amplitude: Amplitude,
2120
) {
2221
private val writeChannel: Channel<WriteQueueMessage>
23-
2422
private val uploadChannel: Channel<String>
25-
2623
private val eventCount: AtomicInteger = AtomicInteger(0)
27-
2824
private val httpClient: HttpClientInterface = amplitude.configuration.httpClient
2925
?: HttpClient(amplitude.configuration)
30-
3126
private val storage get() = amplitude.storage
32-
3327
private val scope get() = amplitude.amplitudeScope
3428

35-
var flushInterval = amplitude.configuration.flushIntervalMillis.toLong()
36-
var flushQueueSize = amplitude.configuration.flushQueueSize
37-
3829
private var running: Boolean
39-
4030
private var scheduled: Boolean
41-
4231
var flushSizeDivider: AtomicInteger = AtomicInteger(1)
4332

44-
var exceededRetries = false
33+
private val responseHandler by lazy {
34+
storage.getResponseHandler(
35+
this@EventPipeline,
36+
amplitude.configuration,
37+
scope,
38+
amplitude.retryDispatcher,
39+
)
40+
}
4541

4642
companion object {
4743
internal const val UPLOAD_SIG = "#!upload"
4844
}
4945

50-
private val responseHandler: ResponseHandler
51-
5246
init {
5347
running = false
5448
scheduled = false
@@ -57,14 +51,6 @@ class EventPipeline(
5751
uploadChannel = Channel(UNLIMITED)
5852

5953
registerShutdownHook()
60-
61-
responseHandler =
62-
storage.getResponseHandler(
63-
this@EventPipeline,
64-
amplitude.configuration,
65-
scope,
66-
amplitude.retryDispatcher,
67-
)
6854
}
6955

7056
fun put(event: BaseEvent) {
@@ -153,15 +139,15 @@ class EventPipeline(
153139
}
154140

155141
private fun getFlushCount(): Int {
156-
val count = flushQueueSize / flushSizeDivider.get()
142+
val count = amplitude.configuration.flushQueueSize / flushSizeDivider.get()
157143
return count.takeUnless { it == 0 } ?: 1
158144
}
159145

160146
private fun schedule() =
161147
scope.launch(amplitude.storageIODispatcher) {
162-
if (isActive && running && !scheduled && !exceededRetries) {
148+
if (isActive && running && !scheduled) {
163149
scheduled = true
164-
delay(flushInterval)
150+
delay(amplitude.configuration.flushIntervalMillis.toLong())
165151
flush()
166152
scheduled = false
167153
}

‎core/src/main/java/com/amplitude/core/utilities/EventsFileManager.kt

+18-24
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,6 @@ import kotlinx.coroutines.sync.withLock
99
import org.json.JSONArray
1010
import org.json.JSONException
1111
import org.json.JSONObject
12-
import java.io.BufferedReader
1312
import java.io.File
1413
import java.io.FileNotFoundException
1514
import java.io.FileOutputStream
@@ -28,8 +27,8 @@ class EventsFileManager(
2827
) {
2928
private val fileIndexKey = "amplitude.events.file.index.$storageKey"
3029
private val storageVersionKey = "amplitude.events.file.version.$storageKey"
31-
val filePathSet: MutableSet<String> = Collections.newSetFromMap(ConcurrentHashMap())
32-
val curFile: MutableMap<String, File> = ConcurrentHashMap<String, File>()
30+
private val filePathSet: MutableSet<String> = Collections.newSetFromMap(ConcurrentHashMap())
31+
private val curFile: MutableMap<String, File> = ConcurrentHashMap<String, File>()
3332

3433
companion object {
3534
const val MAX_FILE_SIZE = 975_000 // 975KB
@@ -38,7 +37,7 @@ class EventsFileManager(
3837
val readMutexMap = ConcurrentHashMap<String, Mutex>()
3938
}
4039

41-
val writeMutex = writeMutexMap.getOrPut(storageKey) { Mutex() }
40+
private val writeMutex = writeMutexMap.getOrPut(storageKey) { Mutex() }
4241
private val readMutex = readMutexMap.getOrPut(storageKey) { Mutex() }
4342

4443
init {
@@ -100,20 +99,24 @@ class EventsFileManager(
10099
*/
101100
fun read(): List<String> {
102101
// we need to filter out .temp file, since it's operating on the writing thread
103-
val fileList =
104-
directory.listFiles { _, name ->
105-
name.contains(storageKey) && !name.endsWith(".tmp") && !name.endsWith(".properties")
106-
} ?: emptyArray()
107-
return fileList.sortedBy { it ->
108-
getSortKeyForFile(it)
102+
val fileList = directory.listFiles { _, name ->
103+
name.contains(storageKey) && !name.endsWith(".tmp") && !name.endsWith(".properties")
104+
} ?: emptyArray()
105+
106+
return fileList.sortedBy { file ->
107+
val name = file.nameWithoutExtension.replace("$storageKey-", "")
108+
109+
val dashIndex = name.indexOf('-')
110+
if (dashIndex >= 0) {
111+
name.substring(0, dashIndex).padStart(10, '0') + name.substring(dashIndex)
112+
} else {
113+
name
114+
}
109115
}.map {
110116
it.absolutePath
111117
}
112118
}
113119

114-
/**
115-
* deletes the file at filePath
116-
*/
117120
fun remove(filePath: String): Boolean {
118121
filePathSet.remove(filePath)
119122
return File(filePath).delete()
@@ -160,8 +163,8 @@ class EventsFileManager(
160163
return@withLock ""
161164
}
162165
filePathSet.add(filePath)
163-
File(filePath).bufferedReader().use<BufferedReader, String> {
164-
val content = it.readText()
166+
File(filePath).bufferedReader().use { reader ->
167+
val content = reader.readText()
165168
val isCurrentVersion = content.endsWith(DELIMITER)
166169
if (isCurrentVersion) {
167170
// handle current version
@@ -247,15 +250,6 @@ class EventsFileManager(
247250
return curFile[storageKey]!!
248251
}
249252

250-
private fun getSortKeyForFile(file: File): String {
251-
val name = file.nameWithoutExtension.replace("$storageKey-", "")
252-
val dashIndex = name.indexOf('-')
253-
if (dashIndex >= 0) {
254-
return name.substring(0, dashIndex).padStart(10, '0') + name.substring(dashIndex)
255-
}
256-
return name
257-
}
258-
259253
// write to underlying file
260254
private fun writeToFile(
261255
content: ByteArray,

0 commit comments

Comments
 (0)