// Ported from golang/nats-server/server/jetstream_test.go // Direct get API: message retrieval by sequence, last message by subject, // missing sequence handling, multi-message get, stream message API. using NATS.Server.JetStream.Models; namespace NATS.Server.Tests.JetStream; public class JetStreamDirectGetTests { // Go: TestJetStreamDirectGetBatch server/jetstream_test.go:16524 // Direct get retrieves a specific message by sequence number. [Fact] public async Task Direct_get_returns_correct_message_for_sequence() { await using var fx = await JetStreamApiFixture.StartWithStreamAsync("DG", "dg.>"); var a1 = await fx.PublishAndGetAckAsync("dg.first", "payload-one"); var a2 = await fx.PublishAndGetAckAsync("dg.second", "payload-two"); var a3 = await fx.PublishAndGetAckAsync("dg.third", "payload-three"); var resp = await fx.RequestLocalAsync( "$JS.API.DIRECT.GET.DG", $$$"""{ "seq": {{{a2.Seq}}} }"""); resp.Error.ShouldBeNull(); resp.DirectMessage.ShouldNotBeNull(); resp.DirectMessage!.Sequence.ShouldBe(a2.Seq); resp.DirectMessage.Subject.ShouldBe("dg.second"); resp.DirectMessage.Payload.ShouldBe("payload-two"); } // Go: TestJetStreamDirectGetBatch — first message in stream [Fact] public async Task Direct_get_retrieves_first_message_by_sequence() { await using var fx = await JetStreamApiFixture.StartWithStreamAsync("DGF", "dgf.>"); var a1 = await fx.PublishAndGetAckAsync("dgf.x", "first-data"); _ = await fx.PublishAndGetAckAsync("dgf.x", "second-data"); var resp = await fx.RequestLocalAsync( "$JS.API.DIRECT.GET.DGF", $$$"""{ "seq": {{{a1.Seq}}} }"""); resp.Error.ShouldBeNull(); resp.DirectMessage!.Payload.ShouldBe("first-data"); resp.DirectMessage.Subject.ShouldBe("dgf.x"); } // Go: TestJetStreamDirectGetBatch — last message in stream [Fact] public async Task Direct_get_retrieves_last_message_by_sequence() { await using var fx = await JetStreamApiFixture.StartWithStreamAsync("DGL", "dgl.>"); _ = await fx.PublishAndGetAckAsync("dgl.x", "first"); var last = await fx.PublishAndGetAckAsync("dgl.x", "last-data"); var resp = await fx.RequestLocalAsync( "$JS.API.DIRECT.GET.DGL", $$$"""{ "seq": {{{last.Seq}}} }"""); resp.Error.ShouldBeNull(); resp.DirectMessage!.Payload.ShouldBe("last-data"); } // Go: TestJetStreamDirectGetBatch — subject is preserved in response [Fact] public async Task Direct_get_response_includes_correct_subject() { await using var fx = await JetStreamApiFixture.StartWithStreamAsync("DGSUB", "dgsub.>"); _ = await fx.PublishAndGetAckAsync("dgsub.orders.created", "order-payload"); var a2 = await fx.PublishAndGetAckAsync("dgsub.events.logged", "event-payload"); var resp = await fx.RequestLocalAsync( "$JS.API.DIRECT.GET.DGSUB", $$$"""{ "seq": {{{a2.Seq}}} }"""); resp.Error.ShouldBeNull(); resp.DirectMessage!.Subject.ShouldBe("dgsub.events.logged"); resp.DirectMessage.Payload.ShouldBe("event-payload"); } // Go: TestJetStreamDirectGetBatch — requesting non-existent sequence returns not found [Fact] public async Task Direct_get_non_existent_sequence_returns_error() { await using var fx = await JetStreamApiFixture.StartWithStreamAsync("DGNE", "dgne.>"); _ = await fx.PublishAndGetAckAsync("dgne.x", "data"); var resp = await fx.RequestLocalAsync( "$JS.API.DIRECT.GET.DGNE", """{ "seq": 999999 }"""); resp.Error.ShouldNotBeNull(); resp.DirectMessage.ShouldBeNull(); } // Go: TestJetStreamDirectGetBatch — empty stream returns error [Fact] public async Task Direct_get_on_empty_stream_returns_error() { await using var fx = await JetStreamApiFixture.StartWithStreamAsync("DGEMPTY", "dgempty.>"); var resp = await fx.RequestLocalAsync( "$JS.API.DIRECT.GET.DGEMPTY", """{ "seq": 1 }"""); resp.Error.ShouldNotBeNull(); resp.DirectMessage.ShouldBeNull(); } // Go: TestJetStreamDirectGetBatch — missing stream returns not found [Fact] public async Task Direct_get_on_missing_stream_returns_not_found() { await using var fx = new JetStreamApiFixture(); var resp = await fx.RequestLocalAsync( "$JS.API.DIRECT.GET.NONEXISTENT", """{ "seq": 1 }"""); resp.Error.ShouldNotBeNull(); } // Go: TestJetStreamDirectGetBatch — sequence 0 in request returns error [Fact] public async Task Direct_get_with_zero_sequence_returns_error() { await using var fx = await JetStreamApiFixture.StartWithStreamAsync("DGZERO", "dgzero.>"); _ = await fx.PublishAndGetAckAsync("dgzero.x", "data"); var resp = await fx.RequestLocalAsync( "$JS.API.DIRECT.GET.DGZERO", """{ "seq": 0 }"""); resp.Error.ShouldNotBeNull(); } // Go: TestJetStreamDirectGetBatch — multiple retrieves are independent [Fact] public async Task Direct_get_multiple_sequences_independently() { await using var fx = await JetStreamApiFixture.StartWithStreamAsync("DGMULTI", "dgmulti.>"); var a1 = await fx.PublishAndGetAckAsync("dgmulti.a", "alpha"); var a2 = await fx.PublishAndGetAckAsync("dgmulti.b", "beta"); var a3 = await fx.PublishAndGetAckAsync("dgmulti.c", "gamma"); var r1 = await fx.RequestLocalAsync("$JS.API.DIRECT.GET.DGMULTI", $$$"""{ "seq": {{{a1.Seq}}} }"""); r1.DirectMessage!.Payload.ShouldBe("alpha"); var r3 = await fx.RequestLocalAsync("$JS.API.DIRECT.GET.DGMULTI", $$$"""{ "seq": {{{a3.Seq}}} }"""); r3.DirectMessage!.Payload.ShouldBe("gamma"); var r2 = await fx.RequestLocalAsync("$JS.API.DIRECT.GET.DGMULTI", $$$"""{ "seq": {{{a2.Seq}}} }"""); r2.DirectMessage!.Payload.ShouldBe("beta"); } // Go: TestJetStreamStreamMessageGet (STREAM.MSG.GET API) server/jetstream_test.go // Stream message get API (not direct) retrieves by sequence. [Fact] public async Task Stream_msg_get_returns_message_by_sequence() { await using var fx = await JetStreamApiFixture.StartWithStreamAsync("MSGGET", "msgget.>"); var a1 = await fx.PublishAndGetAckAsync("msgget.x", "data-one"); _ = await fx.PublishAndGetAckAsync("msgget.y", "data-two"); var resp = await fx.RequestLocalAsync( "$JS.API.STREAM.MSG.GET.MSGGET", $$$"""{ "seq": {{{a1.Seq}}} }"""); resp.Error.ShouldBeNull(); resp.StreamMessage.ShouldNotBeNull(); resp.StreamMessage!.Sequence.ShouldBe(a1.Seq); resp.StreamMessage.Subject.ShouldBe("msgget.x"); resp.StreamMessage.Payload.ShouldBe("data-one"); } // Go: TestJetStreamDeleteMsg — stream msg get after delete returns error [Fact] public async Task Stream_msg_get_after_delete_returns_error() { await using var fx = await JetStreamApiFixture.StartWithStreamAsync("GETDEL", "getdel.>"); var a1 = await fx.PublishAndGetAckAsync("getdel.x", "data"); _ = await fx.RequestLocalAsync( "$JS.API.STREAM.MSG.DELETE.GETDEL", $$$"""{ "seq": {{{a1.Seq}}} }"""); var get = await fx.RequestLocalAsync( "$JS.API.STREAM.MSG.GET.GETDEL", $$$"""{ "seq": {{{a1.Seq}}} }"""); get.StreamMessage.ShouldBeNull(); get.Error.ShouldNotBeNull(); } // Go: TestJetStreamDirectGetBatch — direct get sequence field in response [Fact] public async Task Direct_get_response_sequence_matches_requested_sequence() { await using var fx = await JetStreamApiFixture.StartWithStreamAsync("DGSEQ", "dgseq.>"); _ = await fx.PublishAndGetAckAsync("dgseq.a", "1"); _ = await fx.PublishAndGetAckAsync("dgseq.b", "2"); var a3 = await fx.PublishAndGetAckAsync("dgseq.c", "3"); _ = await fx.PublishAndGetAckAsync("dgseq.d", "4"); var resp = await fx.RequestLocalAsync( "$JS.API.DIRECT.GET.DGSEQ", $$$"""{ "seq": {{{a3.Seq}}} }"""); resp.Error.ShouldBeNull(); resp.DirectMessage!.Sequence.ShouldBe(a3.Seq); } // Go: TestJetStreamDirectGetBatch — payload is preserved verbatim [Fact] public async Task Direct_get_payload_is_preserved_verbatim() { await using var fx = await JetStreamApiFixture.StartWithStreamAsync("DGPAY", "dgpay.>"); const string payload = "Hello, JetStream Direct Get!"; var a1 = await fx.PublishAndGetAckAsync("dgpay.msg", payload); var resp = await fx.RequestLocalAsync( "$JS.API.DIRECT.GET.DGPAY", $$$"""{ "seq": {{{a1.Seq}}} }"""); resp.Error.ShouldBeNull(); resp.DirectMessage!.Payload.ShouldBe(payload); } // Go: TestJetStreamDirectGetBatch — direct get uses stream storage type correctly [Fact] public async Task Direct_get_works_with_memory_storage_stream() { await using var fx = await JetStreamApiFixture.StartWithStreamConfigAsync(new StreamConfig { Name = "DGMEM", Subjects = ["dgmem.>"], Storage = StorageType.Memory, }); var a1 = await fx.PublishAndGetAckAsync("dgmem.x", "in-memory"); var resp = await fx.RequestLocalAsync( "$JS.API.DIRECT.GET.DGMEM", $$$"""{ "seq": {{{a1.Seq}}} }"""); resp.Error.ShouldBeNull(); resp.DirectMessage!.Payload.ShouldBe("in-memory"); } // Go: TestJetStreamDirectGetBatch — backend type reported for memory stream [Fact] public async Task Stream_backend_type_is_memory_for_memory_storage() { await using var fx = await JetStreamApiFixture.StartWithStreamConfigAsync(new StreamConfig { Name = "BACKENDMEM", Subjects = ["backendmem.>"], Storage = StorageType.Memory, }); var backendType = await fx.GetStreamBackendTypeAsync("BACKENDMEM"); backendType.ShouldBe("memory"); } // Go: TestJetStreamDirectGetBatch — direct get after purge returns error [Fact] public async Task Direct_get_after_purge_returns_not_found() { await using var fx = await JetStreamApiFixture.StartWithStreamAsync("DGPURGE", "dgpurge.>"); var a1 = await fx.PublishAndGetAckAsync("dgpurge.x", "data"); _ = await fx.RequestLocalAsync("$JS.API.STREAM.PURGE.DGPURGE", "{}"); var resp = await fx.RequestLocalAsync( "$JS.API.DIRECT.GET.DGPURGE", $$$"""{ "seq": {{{a1.Seq}}} }"""); resp.Error.ShouldNotBeNull(); resp.DirectMessage.ShouldBeNull(); } // Go: TestJetStreamDirectGetBatch — sequence in middle of stream [Fact] public async Task Direct_get_retrieves_middle_sequence_correctly() { await using var fx = await JetStreamApiFixture.StartWithStreamAsync("DGMID", "dgmid.>"); for (var i = 1; i <= 10; i++) _ = await fx.PublishAndGetAckAsync("dgmid.x", $"msg-{i}"); // Get sequence 5 (middle) var resp = await fx.RequestLocalAsync("$JS.API.DIRECT.GET.DGMID", """{ "seq": 5 }"""); resp.Error.ShouldBeNull(); resp.DirectMessage!.Sequence.ShouldBe(5UL); resp.DirectMessage.Payload.ShouldBe("msg-5"); } // Go: TestJetStreamDirectGetBatch — stream msg get vs direct get both return same data [Fact] public async Task Stream_msg_get_and_direct_get_return_consistent_data() { await using var fx = await JetStreamApiFixture.StartWithStreamAsync("CONSISTENT", "consistent.>"); var a1 = await fx.PublishAndGetAckAsync("consistent.x", "consistent-data"); var directResp = await fx.RequestLocalAsync( "$JS.API.DIRECT.GET.CONSISTENT", $$$"""{ "seq": {{{a1.Seq}}} }"""); var msgGetResp = await fx.RequestLocalAsync( "$JS.API.STREAM.MSG.GET.CONSISTENT", $$$"""{ "seq": {{{a1.Seq}}} }"""); directResp.Error.ShouldBeNull(); msgGetResp.Error.ShouldBeNull(); directResp.DirectMessage!.Payload.ShouldBe("consistent-data"); msgGetResp.StreamMessage!.Payload.ShouldBe("consistent-data"); directResp.DirectMessage.Subject.ShouldBe(msgGetResp.StreamMessage.Subject); } }