Clean up market processors (#803)

Author: ruixhuang

src/commonMain/kotlin/exchange.dydx.abacus/processor/markets/CandleProcessor.kt

@@ -3,7 +3,6 @@ package exchange.dydx.abacus.processor.markets
 import exchange.dydx.abacus.output.MarketCandle
 import exchange.dydx.abacus.processor.base.BaseProcessor
 import exchange.dydx.abacus.protocols.ParserProtocol
-import exchange.dydx.abacus.utils.ParsingHelper.Companion.transform
 import indexer.codegen.IndexerCandleResponseObject
 
 /*
@@ -43,25 +42,6 @@ internal interface CandleProcessorProtocol {
 internal class CandleProcessor(
     parser: ParserProtocol
 ) : BaseProcessor(parser), CandleProcessorProtocol {
-    private val candleKeyMap = mapOf(
-        "double" to mapOf(
-            "low" to "low",
-            "high" to "high",
-            "open" to "open",
-            "close" to "close",
-            "baseTokenVolume" to "baseTokenVolume",
-            "usdVolume" to "usdVolume",
-            "startingOpenInterest" to "startingOpenInterest",
-        ),
-        "datetime" to mapOf(
-            "startedAt" to "startedAt",
-            "updatedAt" to "updatedAt",
-        ),
-        "int" to mapOf(
-            "trades" to "trades",
-        ),
-    )
-
     override fun process(
         payload: IndexerCandleResponseObject?
     ): MarketCandle? {
@@ -91,11 +71,4 @@ internal class CandleProcessor(
             trades = parser.asInt(payload.trades),
         )
     }
-
-    override fun received(
-        existing: Map<String, Any>?,
-        payload: Map<String, Any>
-    ): Map<String, Any> {
-        return transform(existing, payload, candleKeyMap)
-    }
 }

src/commonMain/kotlin/exchange.dydx.abacus/processor/markets/CandlesProcessor.kt

@@ -3,8 +3,6 @@ package exchange.dydx.abacus.processor.markets
 import exchange.dydx.abacus.processor.base.BaseProcessor
 import exchange.dydx.abacus.protocols.ParserProtocol
 import exchange.dydx.abacus.state.internalstate.InternalMarketState
-import exchange.dydx.abacus.utils.mutable
-import exchange.dydx.abacus.utils.safeSet
 import indexer.codegen.IndexerCandleResponse
 import indexer.codegen.IndexerCandleResponseObject
 import kotlinx.datetime.Instant
@@ -95,88 +93,4 @@ internal class CandlesProcessor(
         }
         return existing
     }
-
-    override fun received(
-        existing: List<Any>?,
-        payload: List<Any>
-    ): List<Any>? {
-        return mergeDeprecated(
-            parser,
-            existing,
-            parser.asNativeList(payload)?.reversed()?.toList(),
-            "startedAt",
-            true,
-        )
-    }
-
-    internal fun subscribed(
-        existing: Map<String, Any>?,
-        resolution: String,
-        content: Map<String, Any>,
-    ): Map<String, Any>? {
-        val payload =
-            parser.asNativeList(content["candles"])
-        return if (payload != null) {
-            receivedChanges(existing, resolution, payload)
-        } else {
-            existing
-        }
-    }
-
-    @Suppress("FunctionName")
-    internal fun channel_data(
-        existing: Map<String, Any>?,
-        resolution: String,
-        content: Map<String, Any>,
-    ): Map<String, Any>? {
-        // content is a single candle update
-        if (content != null) {
-            val modified = existing?.mutable() ?: mutableMapOf()
-            val existingResolution = parser.asNativeList(existing?.get(resolution))
-            val candles = existingResolution?.mutable() ?: mutableListOf()
-            val lastExisting = parser.asNativeMap(candles.lastOrNull())
-            val lastStartAt = parser.asDatetime(lastExisting?.get("startedAt"))
-            val itemProcessor = itemProcessor as CandleProcessor
-            val incoming = itemProcessor.received(null, content)
-            val incomingStartAt = parser.asDatetime(incoming["startedAt"])
-            if (lastStartAt == incomingStartAt) {
-                candles.removeLast()
-            }
-            candles.add(incoming)
-            modified.safeSet(resolution, candles)
-            return modified
-        } else {
-            return existing
-        }
-    }
-
-    private fun receivedChanges(
-        existing: Map<String, Any>?,
-        resolution: String,
-        payload: List<Any>?,
-    ): Map<String, Any>? {
-        if (payload != null) {
-            val modified = existing?.mutable() ?: mutableMapOf()
-            val existingResolution = parser.asNativeList(existing?.get(resolution))
-            val candles = mutableListOf<Any>()
-            for (value in payload.reversed()) {
-                parser.asNativeMap(value)?.let {
-                    val candleProcessor = itemProcessor as CandleProcessor
-                    val candle = candleProcessor.received(null, it)
-                    candles.add(candle)
-                }
-            }
-            val mergedResolution = mergeDeprecated(
-                parser,
-                existingResolution,
-                candles,
-                "startedAt",
-                true,
-            )
-            modified.safeSet(resolution, mergedResolution)
-            return modified
-        } else {
-            return existing
-        }
-    }
 }

src/commonMain/kotlin/exchange.dydx.abacus/processor/markets/HistoricalFundingProcessor.kt

@@ -9,15 +9,6 @@ import kotlinx.datetime.Instant
 
 @Suppress("UNCHECKED_CAST")
 internal class HistoricalFundingsProcessor(parser: ParserProtocol) : BaseProcessor(parser) {
-    private val itemProcessor = HistoricalFundingProcessor(parser = parser)
-    internal fun received(
-        existing: List<Any>?,
-        content: Map<String, Any>,
-    ): List<Any>? {
-        val payload = parser.asNativeList(content["historicalFunding"]) as? List<Map<String, Any>>
-        return if (payload != null) receivedList(existing, payload) else null
-    }
-
     fun process(
         existing: InternalMarketState,
         payload: List<IndexerHistoricalFundingResponseObject>?,
@@ -51,28 +42,4 @@ internal class HistoricalFundingsProcessor(parser: ParserProtocol) : BaseProcess
 
         return existing
     }
-
-    private fun receivedList(
-        existing: List<Any>?,
-        payload: List<Any>?,
-    ): List<Any>? {
-        if (payload != null) {
-            val history = mutableListOf<Any>()
-            for (value in payload) {
-                parser.asNativeMap(value)?.let {
-                    val item = itemProcessor.received(null, it)
-                    history.add(item)
-                }
-            }
-            return mergeDeprecated(
-                parser,
-                existing,
-                history.reversed().toList(),
-                "effectiveAt",
-                true,
-            )
-        } else {
-            return existing
-        }
-    }
 }