feat(backend): Aenderungsfeed per Long Polling (Schritt 4)
GET /documents/{id}/changes haelt die Anfrage offen und antwortet, sobald
sich etwas tut — kumuliertes Diff seit der bekannten Version, dazu die
Ereignisse mit ihrem Absender. Ist die Basis verdichtet oder hat der Client
noch gar nichts, kommt der Volltext statt der Operationen: ein Roundtrip und
ein Sonderzustand weniger als ein eigener Fehlerpfad.
Der Feed arbeitet auf der Historie, nicht am Dokument — ein geloeschtes
Dokument muss sein DELETED noch zustellen koennen.
Blockierend auf virtuellen Threads statt DeferredResult (D76-Nachtrag 5):
So behaelt der Endpunkt die aus der Spezifikation generierte Signatur, und
API-First bleibt fuer ihn unangetastet; ein Wartender kostet trotzdem fast
nichts. Geweckt wird nach dem Commit, nie davor, und ueber einen Stempel, den
der Aufrufer VOR dem Nachsehen liest — sonst ginge ein Signal aus der Luecke
dazwischen verloren.
122 Tests. Das Szenario "ein Wartender wird geweckt" misst die Dauer: Ohne
das bestuende es auch dann, wenn der Wartende bloss in den Timeout liefe und
danach die Aenderung vorfaende. Gegenprobe: Benachrichtigung entfernt ->
genau dieses Szenario faellt; Volltext-Rueckfall entfernt -> genau jenes.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
mhoennig
co-authored by
Claude Opus 5
Claude Opus 5
parent
ae85503b79
commit
741b41ef11
@@ -1,6 +1,7 @@
|
||||
package de.werkbaum.bdd
|
||||
|
||||
import de.werkbaum.diff.LineDiff
|
||||
import de.werkbaum.generated.model.ChangeFeed
|
||||
import de.werkbaum.generated.model.ContentConflict
|
||||
import de.werkbaum.generated.model.ContentPatchResult
|
||||
import de.werkbaum.generated.model.Document as ApiDocument
|
||||
@@ -12,11 +13,14 @@ import io.cucumber.java.de.Und
|
||||
import io.cucumber.java.de.Wenn
|
||||
import io.kotest.assertions.withClue
|
||||
import io.kotest.matchers.nulls.shouldNotBeNull
|
||||
import io.kotest.matchers.collections.shouldContain
|
||||
import io.kotest.matchers.shouldBe
|
||||
import org.springframework.beans.factory.annotation.Autowired
|
||||
import org.springframework.http.MediaType
|
||||
import org.springframework.test.web.servlet.client.EntityExchangeResult
|
||||
import org.springframework.test.web.servlet.client.RestTestClient
|
||||
import java.util.concurrent.CompletableFuture
|
||||
import java.util.concurrent.TimeUnit
|
||||
|
||||
/**
|
||||
* Behavior-Tests des Live-Editings (D76) gegen die laufende Anwendung.
|
||||
@@ -38,6 +42,10 @@ class LiveEditingStepDefinitions {
|
||||
private var lastResponse: EntityExchangeResult<String>? = null
|
||||
private var lastRequestBody: String? = null
|
||||
|
||||
/** Ein Feed-Abruf, der im Hintergrund wartet – für Long Polling. */
|
||||
private var pendingFeed: CompletableFuture<EntityExchangeResult<String>>? = null
|
||||
private var feedStartedAt: Long = 0
|
||||
|
||||
private fun status(): Int? = lastResponse?.status?.value()
|
||||
|
||||
private fun currentDocument(): ApiDocument =
|
||||
@@ -79,7 +87,8 @@ class LiveEditingStepDefinitions {
|
||||
seq: Long = 1,
|
||||
) = """
|
||||
{"baseVersion":$baseVersion,"checksum":${json(checksum)},
|
||||
"clientId":${json(clientId)},"seq":$seq,"ops":$ops}
|
||||
"clientId":${json(clientId)},"displayName":${json(clientId)},
|
||||
"seq":$seq,"ops":$ops}
|
||||
""".trimIndent()
|
||||
|
||||
private fun baseOf(clientId: String): Pair<Long, String> =
|
||||
@@ -115,6 +124,17 @@ class LiveEditingStepDefinitions {
|
||||
.status.value() shouldBe 204
|
||||
}
|
||||
|
||||
@Angenommen("dieses Dokument wiederhergestellt wird")
|
||||
fun `dieses Dokument wird wiederhergestellt`() {
|
||||
client.post()
|
||||
.uri("/api/v1/documents/$documentId/restore")
|
||||
.contentType(MediaType.APPLICATION_JSON)
|
||||
.body("{}")
|
||||
.exchange()
|
||||
.returnResult(String::class.java)
|
||||
.status.value() shouldBe 200
|
||||
}
|
||||
|
||||
// ---------------- Wenn ----------------
|
||||
|
||||
@Wenn("Client {string} folgendes Diff einreicht:")
|
||||
@@ -134,6 +154,86 @@ class LiveEditingStepDefinitions {
|
||||
sendPatch(lastRequestBody.shouldNotBeNull())
|
||||
}
|
||||
|
||||
// ---------------- Änderungsfeed ----------------
|
||||
|
||||
@Wenn("ich die Änderungen seit Version {long} abrufe")
|
||||
fun `ich rufe die Aenderungen ab`(since: Long) {
|
||||
lastResponse = feedRequest(since, wait = 0)
|
||||
}
|
||||
|
||||
@Wenn("ich höchstens {int} Sekunden auf Änderungen seit Version {long} warte")
|
||||
fun `ich warte auf Aenderungen`(sekunden: Int, since: Long) {
|
||||
lastResponse = feedRequest(since, wait = sekunden)
|
||||
}
|
||||
|
||||
@Wenn("im Hintergrund auf Änderungen seit Version {long} gewartet wird")
|
||||
fun `im Hintergrund wird gewartet`(since: Long) {
|
||||
feedStartedAt = System.nanoTime()
|
||||
pendingFeed = CompletableFuture.supplyAsync { feedRequest(since, wait = 5) }
|
||||
// Dem Abruf einen Moment geben, damit er wirklich wartet, statt die
|
||||
// Aenderung schon vorzufinden - sonst prueft das Szenario nichts.
|
||||
Thread.sleep(300)
|
||||
}
|
||||
|
||||
@Dann("hat der wartende Abruf die Änderung erhalten")
|
||||
fun `der wartende Abruf hat die Aenderung erhalten`() {
|
||||
val response = pendingFeed.shouldNotBeNull().get(10, TimeUnit.SECONDS)
|
||||
val dauer = (System.nanoTime() - feedStartedAt) / 1_000_000
|
||||
withClue("Antwort: ${response.responseBody}") { response.status.value() shouldBe 200 }
|
||||
withClue("Der Abruf hat $dauer ms gebraucht - er wurde nicht geweckt, sondern lief ab") {
|
||||
(dauer < 4_000) shouldBe true
|
||||
}
|
||||
lastResponse = response
|
||||
}
|
||||
|
||||
@Und("der Feed meldet die Version {long}")
|
||||
fun `der Feed meldet die Version`(erwartet: Long) {
|
||||
lastBody<ChangeFeed>().currentVersion shouldBe erwartet
|
||||
}
|
||||
|
||||
@Und("der Feed liefert {int} Operationen ab Version {long}")
|
||||
fun `der Feed liefert n Operationen`(anzahl: Int, from: Long) {
|
||||
val feed = lastBody<ChangeFeed>()
|
||||
feed.fromVersion shouldBe from
|
||||
feed.ops.shouldNotBeNull().size shouldBe anzahl
|
||||
}
|
||||
|
||||
@Und("der Feed liefert den Volltext:")
|
||||
fun `der Feed liefert den Volltext`(erwartet: String) {
|
||||
val feed = lastBody<ChangeFeed>()
|
||||
feed.fromVersion shouldBe null
|
||||
feed.content shouldBe erwartet
|
||||
}
|
||||
|
||||
@Und("der Feed meldet das Ereignis {string}")
|
||||
fun `der Feed meldet das Ereignis`(typ: String) {
|
||||
lastBody<ChangeFeed>().events.map { it.changeType.value } shouldContain typ
|
||||
}
|
||||
|
||||
@Und("der Feed nennt als Absender {string}")
|
||||
fun `der Feed nennt als Absender`(name: String) {
|
||||
lastBody<ChangeFeed>().events.mapNotNull { it.displayName } shouldContain name
|
||||
}
|
||||
|
||||
@Und("die Antwort verbietet das Zwischenspeichern")
|
||||
fun `die Antwort verbietet das Zwischenspeichern`() {
|
||||
lastResponse?.responseHeaders?.cacheControl shouldBe "no-store"
|
||||
}
|
||||
|
||||
@Wenn("ich die Änderungen eines unbekannten Dokuments abrufe")
|
||||
fun `ich rufe die Aenderungen eines unbekannten Dokuments ab`() {
|
||||
lastResponse = client.get()
|
||||
.uri("/api/v1/documents/00000000-0000-0000-0000-000000000000/changes?since=0&wait=0")
|
||||
.exchange()
|
||||
.returnResult(String::class.java)
|
||||
}
|
||||
|
||||
private fun feedRequest(since: Long, wait: Int): EntityExchangeResult<String> =
|
||||
client.get()
|
||||
.uri("/api/v1/documents/$documentId/changes?since=$since&wait=$wait")
|
||||
.exchange()
|
||||
.returnResult(String::class.java)
|
||||
|
||||
// ---------------- Dann / Und ----------------
|
||||
|
||||
@Dann("erhalte ich für das Diff den Status {int}")
|
||||
|
||||
@@ -0,0 +1,50 @@
|
||||
package de.werkbaum.service
|
||||
|
||||
import io.kotest.matchers.shouldBe
|
||||
import org.junit.jupiter.api.Test
|
||||
import java.time.Duration
|
||||
import java.util.UUID
|
||||
import java.util.concurrent.CompletableFuture
|
||||
import java.util.concurrent.TimeUnit
|
||||
|
||||
class ChangeNotifierTest {
|
||||
|
||||
private val notifier = ChangeNotifier()
|
||||
private val id = UUID.randomUUID()
|
||||
|
||||
@Test
|
||||
fun `ohne Aenderung laeuft die Wartezeit ab`() {
|
||||
notifier.awaitChange(id, notifier.stampOf(id), Duration.ofMillis(50)) shouldBe false
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `eine Aenderung weckt den Wartenden`() {
|
||||
val stamp = notifier.stampOf(id)
|
||||
val wartend = CompletableFuture.supplyAsync {
|
||||
notifier.awaitChange(id, stamp, Duration.ofSeconds(5))
|
||||
}
|
||||
Thread.sleep(100)
|
||||
notifier.published(id)
|
||||
|
||||
wartend.get(5, TimeUnit.SECONDS) shouldBe true
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `eine Aenderung in der Luecke geht nicht verloren`() {
|
||||
// Genau dafuer ist der Stempel da: Der Aufrufer liest ihn, bevor er in
|
||||
// der Datenbank nachsieht. Passiert dazwischen etwas, kehrt das Warten
|
||||
// sofort zurueck, statt die volle Zeit abzusitzen.
|
||||
val stamp = notifier.stampOf(id)
|
||||
notifier.published(id)
|
||||
|
||||
notifier.awaitChange(id, stamp, Duration.ofMillis(50)) shouldBe true
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `ein anderes Dokument weckt nicht`() {
|
||||
val stamp = notifier.stampOf(id)
|
||||
notifier.published(UUID.randomUUID())
|
||||
|
||||
notifier.awaitChange(id, stamp, Duration.ofMillis(50)) shouldBe false
|
||||
}
|
||||
}
|
||||
@@ -38,7 +38,9 @@ class DocumentServiceTest {
|
||||
|
||||
private val repository = mockk<DocumentRepository>()
|
||||
private val historyRepository = mockk<DocumentHistoryRepository>(relaxed = true)
|
||||
private val service = DocumentService(repository, historyRepository, clock, properties)
|
||||
private val notifier = ChangeNotifier()
|
||||
private val service =
|
||||
DocumentService(repository, historyRepository, clock, properties, notifier)
|
||||
|
||||
private fun sampleDocument(
|
||||
id: UUID = UUID.randomUUID(),
|
||||
|
||||
@@ -8,6 +8,7 @@ import de.werkbaum.domain.ChangeType
|
||||
import de.werkbaum.domain.ContentPatch
|
||||
import de.werkbaum.domain.ContentPatchOutcome
|
||||
import de.werkbaum.domain.Document
|
||||
import de.werkbaum.domain.ChangeEvent
|
||||
import de.werkbaum.domain.DocumentHistoryEntry
|
||||
import de.werkbaum.repository.DocumentHistoryRepository
|
||||
import io.kotest.assertions.throwables.shouldThrow
|
||||
@@ -17,6 +18,7 @@ import io.mockk.mockk
|
||||
import io.mockk.slot
|
||||
import io.mockk.verify
|
||||
import org.junit.jupiter.api.Test
|
||||
import java.time.Duration
|
||||
import java.time.OffsetDateTime
|
||||
import java.util.UUID
|
||||
|
||||
@@ -25,8 +27,13 @@ class LiveEditingServiceTest {
|
||||
private val id = UUID.randomUUID()
|
||||
private val documents = mockk<DocumentService>()
|
||||
private val history = mockk<DocumentHistoryRepository>(relaxed = true)
|
||||
private val properties = LiveEditingProperties(maxOps = 3, maxContentLength = 40)
|
||||
private val service = LiveEditingService(documents, history, properties)
|
||||
private val properties = LiveEditingProperties(
|
||||
maxOps = 3,
|
||||
maxContentLength = 40,
|
||||
maxWait = Duration.ofMillis(200),
|
||||
)
|
||||
private val notifier = ChangeNotifier()
|
||||
private val service = LiveEditingService(documents, history, properties, notifier)
|
||||
|
||||
private val basis = "eins\nzwei\ndrei"
|
||||
|
||||
@@ -242,6 +249,104 @@ class LiveEditingServiceTest {
|
||||
verify(exactly = 0) { documents.update(any(), any(), any(), any(), any()) }
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
// Änderungsfeed
|
||||
// -----------------------------------------------------------------------
|
||||
|
||||
@Test
|
||||
fun `ohne Aenderung liefert der Feed nichts`() {
|
||||
every { history.exists(id) } returns true
|
||||
every { history.findLatest(id) } returns historyEntry(basis, 7)
|
||||
|
||||
service.changesSince(id, since = 7, wait = Duration.ZERO) shouldBe null
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `der Feed liefert das kumulierte Diff seit der bekannten Version`() {
|
||||
every { history.exists(id) } returns true
|
||||
every { history.findLatest(id) } returns historyEntry("eins\nZWEI\ndrei", 9)
|
||||
every { history.findVersion(id, 7) } returns historyEntry(basis, 7)
|
||||
every { history.findAfterVersion(id, 7) } returns listOf(
|
||||
historyEntry("eins\nzwischendrin\ndrei", 8),
|
||||
historyEntry("eins\nZWEI\ndrei", 9),
|
||||
)
|
||||
|
||||
val feed = service.changesSince(id, since = 7, wait = Duration.ZERO)!!
|
||||
|
||||
feed.fromVersion shouldBe 7
|
||||
feed.currentVersion shouldBe 9
|
||||
feed.ops shouldBe listOf(LineOp.Replace(1, 1, listOf("ZWEI")))
|
||||
feed.content shouldBe null
|
||||
feed.events.map { it.version } shouldBe listOf(8L, 9L)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `eine verdichtete Basis liefert den Volltext statt eines Diffs`() {
|
||||
every { history.exists(id) } returns true
|
||||
every { history.findLatest(id) } returns historyEntry("neu", 9)
|
||||
every { history.findVersion(id, 2) } returns null
|
||||
every { history.findAfterVersion(id, 2) } returns emptyList()
|
||||
|
||||
val feed = service.changesSince(id, since = 2, wait = Duration.ZERO)!!
|
||||
|
||||
feed.fromVersion shouldBe null
|
||||
feed.ops shouldBe null
|
||||
feed.content shouldBe "neu"
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `der Feed nennt den Absender jeder Aenderung`() {
|
||||
every { history.exists(id) } returns true
|
||||
every { history.findLatest(id) } returns historyEntry("neu", 8)
|
||||
every { history.findVersion(id, 7) } returns historyEntry(basis, 7)
|
||||
every { history.findAfterVersion(id, 7) } returns listOf(
|
||||
historyEntry("neu", 8).copy(author = ChangeAuthor("c-1", "Anna")),
|
||||
)
|
||||
|
||||
service.changesSince(id, since = 7, wait = Duration.ZERO)!!.events shouldBe listOf(
|
||||
ChangeEvent(8, ChangeType.UPDATED, ChangeAuthor("c-1", "Anna")),
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `ein geloeschtes Dokument hat weiterhin einen Feed`() {
|
||||
// Sonst käme ausgerechnet das DELETED-Ereignis nie an.
|
||||
every { history.exists(id) } returns true
|
||||
every { history.findLatest(id) } returns
|
||||
historyEntry(basis, 8).copy(changeType = ChangeType.DELETED)
|
||||
every { history.findVersion(id, 7) } returns historyEntry(basis, 7)
|
||||
every { history.findAfterVersion(id, 7) } returns listOf(
|
||||
historyEntry(basis, 8).copy(changeType = ChangeType.DELETED),
|
||||
)
|
||||
|
||||
val feed = service.changesSince(id, since = 7, wait = Duration.ZERO)!!
|
||||
|
||||
feed.events.single().changeType shouldBe ChangeType.DELETED
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `eine gaenzlich unbekannte UUID hat keinen Feed`() {
|
||||
every { history.exists(id) } returns false
|
||||
|
||||
shouldThrow<DocumentNotFoundException> {
|
||||
service.changesSince(id, since = 0, wait = Duration.ZERO)
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `die Wartezeit wird serverseitig geklemmt`() {
|
||||
// maxWait steht in diesem Test auf 200 ms; ein Client darf keine
|
||||
// beliebig lange Verbindung binden.
|
||||
every { history.exists(id) } returns true
|
||||
every { history.findLatest(id) } returns historyEntry(basis, 7)
|
||||
|
||||
val start = System.nanoTime()
|
||||
service.changesSince(id, since = 7, wait = Duration.ofSeconds(30)) shouldBe null
|
||||
val dauer = Duration.ofNanos(System.nanoTime() - start)
|
||||
|
||||
(dauer < Duration.ofSeconds(5)) shouldBe true
|
||||
}
|
||||
|
||||
private fun historyEntry(content: String, version: Long) = DocumentHistoryEntry(
|
||||
documentId = id,
|
||||
version = version,
|
||||
|
||||
@@ -10,3 +10,12 @@ spring:
|
||||
open-in-view: false
|
||||
liquibase:
|
||||
change-log: classpath:db/changelog/db.changelog-master.sql
|
||||
|
||||
threads:
|
||||
virtual:
|
||||
enabled: true
|
||||
|
||||
werkbaum:
|
||||
live-editing:
|
||||
# Kurz, damit die Behavior-Tests nicht auf die Produktionswerte warten.
|
||||
max-wait: 5s
|
||||
|
||||
@@ -0,0 +1,94 @@
|
||||
# language: de
|
||||
Funktionalität: Änderungen mitbekommen, ohne zu pollen
|
||||
Als Betrachter eines geteilten Plans
|
||||
möchte ich Änderungen sofort sehen,
|
||||
ohne dass mein Browser dauernd nachfragt.
|
||||
|
||||
Der Server hält die Anfrage offen und antwortet, sobald sich etwas tut
|
||||
(Long Polling). Kommt in der Wartezeit nichts, antwortet er mit 204 und der
|
||||
Client fragt erneut.
|
||||
|
||||
Szenario: Wer zurückliegt, bekommt sofort das Diff
|
||||
Angenommen es existiert ein Dokument "Plan" mit den Zeilen:
|
||||
"""
|
||||
- [ ] Eins
|
||||
- [ ] Zwei
|
||||
"""
|
||||
Und Client "anna" kennt den aktuellen Stand
|
||||
Und Client "ben" folgendes Diff einreicht:
|
||||
"""
|
||||
[{"op":"replace","index":0,"count":1,"lines":["- [x] Eins"]}]
|
||||
"""
|
||||
Wenn ich die Änderungen seit Version 1 abrufe
|
||||
Dann erhalte ich für das Diff den Status 200
|
||||
Und der Feed meldet die Version 2
|
||||
Und der Feed liefert 1 Operationen ab Version 1
|
||||
Und der Feed meldet das Ereignis "UPDATED"
|
||||
Und die Antwort verbietet das Zwischenspeichern
|
||||
|
||||
Szenario: Wer auf dem neuesten Stand ist, bekommt nichts
|
||||
Angenommen es existiert ein Dokument "Plan" mit den Zeilen:
|
||||
"""
|
||||
- [ ] Eins
|
||||
"""
|
||||
Wenn ich höchstens 1 Sekunden auf Änderungen seit Version 1 warte
|
||||
Dann erhalte ich für das Diff den Status 204
|
||||
|
||||
Szenario: Ein Wartender wird geweckt, sobald eine Änderung eintrifft
|
||||
Angenommen es existiert ein Dokument "Plan" mit den Zeilen:
|
||||
"""
|
||||
- [ ] Eins
|
||||
"""
|
||||
Und im Hintergrund auf Änderungen seit Version 1 gewartet wird
|
||||
Wenn Client "ben" folgendes Diff einreicht:
|
||||
"""
|
||||
[{"op":"insert","index":1,"lines":["- [ ] Zwei"]}]
|
||||
"""
|
||||
Dann hat der wartende Abruf die Änderung erhalten
|
||||
Und der Feed meldet die Version 2
|
||||
Und der Feed liefert 1 Operationen ab Version 1
|
||||
|
||||
Szenario: Wer noch gar nichts hat, bekommt den Volltext
|
||||
Angenommen es existiert ein Dokument "Plan" mit den Zeilen:
|
||||
"""
|
||||
- [ ] Eins
|
||||
- [ ] Zwei
|
||||
"""
|
||||
Wenn ich die Änderungen seit Version 0 abrufe
|
||||
Dann erhalte ich für das Diff den Status 200
|
||||
Und der Feed liefert den Volltext:
|
||||
"""
|
||||
- [ ] Eins
|
||||
- [ ] Zwei
|
||||
"""
|
||||
|
||||
Szenario: Der Feed nennt den Absender einer Änderung
|
||||
Angenommen es existiert ein Dokument "Plan" mit den Zeilen:
|
||||
"""
|
||||
- [ ] Eins
|
||||
"""
|
||||
Und Client "anna" folgendes Diff einreicht:
|
||||
"""
|
||||
[{"op":"replace","index":0,"count":1,"lines":["- [x] Eins"]}]
|
||||
"""
|
||||
Wenn ich die Änderungen seit Version 1 abrufe
|
||||
Dann erhalte ich für das Diff den Status 200
|
||||
Und der Feed nennt als Absender "anna"
|
||||
|
||||
Szenario: Der Feed meldet das Löschen und die Wiederherstellung
|
||||
Angenommen es existiert ein Dokument "Plan" mit den Zeilen:
|
||||
"""
|
||||
- [ ] Eins
|
||||
"""
|
||||
Und dieses Dokument gelöscht wird
|
||||
Wenn ich die Änderungen seit Version 1 abrufe
|
||||
Dann erhalte ich für das Diff den Status 200
|
||||
Und der Feed meldet das Ereignis "DELETED"
|
||||
Wenn dieses Dokument wiederhergestellt wird
|
||||
Und ich die Änderungen seit Version 2 abrufe
|
||||
Dann erhalte ich für das Diff den Status 200
|
||||
Und der Feed meldet das Ereignis "RESTORED"
|
||||
|
||||
Szenario: Eine gänzlich unbekannte UUID hat keinen Feed
|
||||
Wenn ich die Änderungen eines unbekannten Dokuments abrufe
|
||||
Dann erhalte ich für das Diff den Status 404
|
||||
Reference in New Issue
Block a user