fix(infra): 消除 StubRepositories 编译警告
- KDoc 移到注解之前(2 处) - upsertSchd 改显式 null 检查避免 !! 警告 - findBackfillDue 链式调用转 asSequence - 移除 persistFullState 中 always-null 的 fallback - markDeleted/revive 改 !! 断言(mains/snapshots 同步写入)
This commit is contained in:
@@ -54,9 +54,9 @@ class StubPipelineLock : PipelineLockRepository {
|
|||||||
override fun lock() = Unit
|
override fun lock() = Unit
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/** 内存版 PROC_STATE:入队幂等,写终态时一并写下回填待办。 */
|
||||||
@Singleton
|
@Singleton
|
||||||
@Requires(property = "msgx.stubs", value = "true")
|
@Requires(property = "msgx.stubs", value = "true")
|
||||||
/** 内存版 PROC_STATE:入队幂等,写终态时一并写下回填待办。 */
|
|
||||||
class StubProcState(private val clock: Clock = Clock.systemUTC()) : ProcStateRepository {
|
class StubProcState(private val clock: Clock = Clock.systemUTC()) : ProcStateRepository {
|
||||||
val rows = linkedMapOf<Long, ProcState>()
|
val rows = linkedMapOf<Long, ProcState>()
|
||||||
val bound = linkedMapOf<String, Long>()
|
val bound = linkedMapOf<String, Long>()
|
||||||
@@ -186,6 +186,7 @@ class StubProcState(private val clock: Clock = Clock.systemUTC()) : ProcStateRep
|
|||||||
|
|
||||||
override fun findBackfillDue(now: Instant, overdueBefore: Instant, limit: Int): List<BackfillDue> =
|
override fun findBackfillDue(now: Instant, overdueBefore: Instant, limit: Int): List<BackfillDue> =
|
||||||
rows.values
|
rows.values
|
||||||
|
.asSequence()
|
||||||
.filter { it.state.isTerminal() && it.backfillAt == null && it.backfillAbandonedAt == null }
|
.filter { it.state.isTerminal() && it.backfillAt == null && it.backfillAbandonedAt == null }
|
||||||
// 超期判据是本地入队时间,与 JDBC 实现保持一致(不用库方时钟的 receivedAt)。
|
// 超期判据是本地入队时间,与 JDBC 实现保持一致(不用库方时钟的 receivedAt)。
|
||||||
.map { it to (it.enqueuedAt?.let { e -> e < overdueBefore } ?: false) }
|
.map { it to (it.enqueuedAt?.let { e -> e < overdueBefore } ?: false) }
|
||||||
@@ -194,6 +195,7 @@ class StubProcState(private val clock: Clock = Clock.systemUTC()) : ProcStateRep
|
|||||||
.sortedWith(compareBy({ it.first.backfillAttempts }, { it.first.msgId }))
|
.sortedWith(compareBy({ it.first.backfillAttempts }, { it.first.msgId }))
|
||||||
.take(limit)
|
.take(limit)
|
||||||
.map { (row, overdue) -> BackfillDue(row.msgId, row.backfillAttempts, overdue) }
|
.map { (row, overdue) -> BackfillDue(row.msgId, row.backfillAttempts, overdue) }
|
||||||
|
.toList()
|
||||||
|
|
||||||
override fun hasAny(): Boolean = rows.isNotEmpty()
|
override fun hasAny(): Boolean = rows.isNotEmpty()
|
||||||
|
|
||||||
@@ -251,11 +253,12 @@ class StubMsgEvents : MsgEventRepository {
|
|||||||
*/
|
*/
|
||||||
private fun upsertSchd(e: MsgEvent): Long {
|
private fun upsertSchd(e: MsgEvent): Long {
|
||||||
val existing = rows.values.firstOrNull { it.target == Targets.KAFKA_SCHD && it.partitionKey == e.partitionKey }
|
val existing = rows.values.firstOrNull { it.target == Targets.KAFKA_SCHD && it.partitionKey == e.partitionKey }
|
||||||
val accepted = existing == null ||
|
if (existing != null) {
|
||||||
e.stateVersion > existing.stateVersion ||
|
val accepted = e.stateVersion > existing.stateVersion ||
|
||||||
(e.stateVersion == existing.stateVersion && e.eventType == EventType.TOMBSTONE && existing.eventType != EventType.TOMBSTONE)
|
(e.stateVersion == existing.stateVersion && e.eventType == EventType.TOMBSTONE && existing.eventType != EventType.TOMBSTONE)
|
||||||
if (!accepted) return existing!!.eventId!!
|
if (!accepted) return existing.eventId!!
|
||||||
existing?.let { rows.remove(it.eventId!!) }
|
rows.remove(existing.eventId!!)
|
||||||
|
}
|
||||||
val id = ids.incrementAndGet()
|
val id = ids.incrementAndGet()
|
||||||
rows[id] = e.copy(
|
rows[id] = e.copy(
|
||||||
eventId = id, state = EventStatus.PENDING, attempts = 0, nextAttemptAt = null,
|
eventId = id, state = EventStatus.PENDING, attempts = 0, nextAttemptAt = null,
|
||||||
@@ -327,7 +330,7 @@ class StubFlightState : FlightStateRepository {
|
|||||||
}
|
}
|
||||||
val main = FlightMainRow(
|
val main = FlightMainRow(
|
||||||
flid = snapshot.flid,
|
flid = snapshot.flid,
|
||||||
operationDay = snapshot.operationDay ?: existing?.operationDay,
|
operationDay = snapshot.operationDay,
|
||||||
state = snapshot.state,
|
state = snapshot.state,
|
||||||
stateVersion = snapshot.stateVersion,
|
stateVersion = snapshot.stateVersion,
|
||||||
lastMsgId = msgId,
|
lastMsgId = msgId,
|
||||||
@@ -342,7 +345,7 @@ class StubFlightState : FlightStateRepository {
|
|||||||
val main = mains[flid] ?: return false
|
val main = mains[flid] ?: return false
|
||||||
if (main.state != FlightState.ACTIVE) return false
|
if (main.state != FlightState.ACTIVE) return false
|
||||||
mains[flid] = main.copy(state = FlightState.DELETED, stateVersion = main.stateVersion + 1, lastMsgId = msgId, updatedAt = now)
|
mains[flid] = main.copy(state = FlightState.DELETED, stateVersion = main.stateVersion + 1, lastMsgId = msgId, updatedAt = now)
|
||||||
snapshots[flid] = (snapshots[flid] ?: return true).copy(state = FlightState.DELETED, stateVersion = main.stateVersion + 1)
|
snapshots[flid] = snapshots[flid]!!.copy(state = FlightState.DELETED, stateVersion = main.stateVersion + 1)
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -350,7 +353,7 @@ class StubFlightState : FlightStateRepository {
|
|||||||
val main = mains[flid] ?: return false
|
val main = mains[flid] ?: return false
|
||||||
if (main.state != FlightState.DELETED) return false
|
if (main.state != FlightState.DELETED) return false
|
||||||
mains[flid] = main.copy(state = FlightState.ACTIVE, stateVersion = main.stateVersion + 1, lastMsgId = msgId, updatedAt = now)
|
mains[flid] = main.copy(state = FlightState.ACTIVE, stateVersion = main.stateVersion + 1, lastMsgId = msgId, updatedAt = now)
|
||||||
snapshots[flid] = (snapshots[flid] ?: return true).copy(state = FlightState.ACTIVE, stateVersion = main.stateVersion + 1)
|
snapshots[flid] = snapshots[flid]!!.copy(state = FlightState.ACTIVE, stateVersion = main.stateVersion + 1)
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -450,9 +453,9 @@ class StubReqTrack : ReqTrackRepository {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/** 内存版共享信箱:可以模拟上游写入、库方清除,并记录哪些行被打上了处理标记。 */
|
||||||
@Singleton
|
@Singleton
|
||||||
@Requires(property = "msgx.stubs", value = "true")
|
@Requires(property = "msgx.stubs", value = "true")
|
||||||
/** 内存版共享信箱:可以模拟上游写入、库方清除,并记录哪些行被打上了处理标记。 */
|
|
||||||
class StubInbox(private val clock: Clock = Clock.systemUTC()) : CminmsgInboxRepository {
|
class StubInbox(private val clock: Clock = Clock.systemUTC()) : CminmsgInboxRepository {
|
||||||
val raws = linkedMapOf<Long, String>()
|
val raws = linkedMapOf<Long, String>()
|
||||||
private val received = linkedMapOf<Long, Instant>()
|
private val received = linkedMapOf<Long, Instant>()
|
||||||
|
|||||||
Reference in New Issue
Block a user