gogrlx/nats-server
1
0
Fork 0
mirror of https://github.com/gogrlx/nats-server.git synced 2026-04-17 11:24:44 -07:00
Code Issues Packages Projects Releases Wiki Activity

Simplified a bit

Browse Source 
Signed-off-by: Derek Collison <derek@nats.io>
This commit is contained in:
Derek Collison
2021-10-12 06:12:56 -07:00
parent 3b9c358a81
commit c789ad9211
1 changed files with 37 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files

100
server/jetstream_test.go
View File

@@ -12389,75 +12389,49 @@ func TestJetStreamDeliverLastPerSubject(t *testing.T) {
}
func TestJetStreamDeliverLastPerSubjectNumPending(t *testing.T) {
for _, st := range []nats.StorageType{nats.FileStorage} { //}, nats.MemoryStorage} {
t.Run(st.String(), func(t *testing.T) {
s := RunBasicJetStreamServer()
defer s.Shutdown()
s := RunBasicJetStreamServer()
defer s.Shutdown()
if config := s.JetStreamConfig(); config != nil {
defer removeDir(t, config.StoreDir)
}
if config := s.JetStreamConfig(); config != nil {
defer removeDir(t, config.StoreDir)
}
// Client for API requests.
nc, js := jsClientConnect(t, s)
defer nc.Close()
// Client for API requests.
nc, js := jsClientConnect(t, s)
defer nc.Close()
if _, err := js.AddStream(&nats.StreamConfig{
Name: "KV",
Subjects: []string{"KV.>"},
Storage: st,
MaxMsgsPerSubject: 5,
Replicas: 1,
}); err != nil {
t.Fatalf("Error adding stream: %v", err)
}
if _, err := js.AddStream(&nats.StreamConfig{
Name: "KV",
Subjects: []string{"KV.>"},
MaxMsgsPerSubject: 5,
Replicas: 1,
}); err != nil {
t.Fatalf("Error adding stream: %v", err)
}
for i := 0; i < 5; i++ {
msg := []byte(fmt.Sprintf("msg%d", i))
js.PublishAsync("KV.foo", msg)
js.PublishAsync("KV.bar", msg)
js.PublishAsync("KV.baz", msg)
js.PublishAsync("KV.bat", msg)
}
for i := 0; i < 5; i++ {
msg := []byte(fmt.Sprintf("msg%d", i))
js.Publish("KV.foo", msg)
js.Publish("KV.bar", msg)
js.Publish("KV.baz", msg)
js.Publish("KV.bat", msg)
}
select {
case <-js.PublishAsyncComplete():
case <-time.After(2 * time.Second):
t.Fatalf("Did not receive completion signal")
}
// Delete some messages
js.DeleteMsg("KV", 2)
js.DeleteMsg("KV", 5)
// Delete some messages
js.DeleteMsg("KV", 2)
js.DeleteMsg("KV", 5)
checkFor(t, time.Second, 15*time.Millisecond, func() error {
si, err := js.StreamInfo("KV")
if err != nil {
return err
}
if si.State.Msgs != 18 {
t.Fatalf("Expected 18 messages, got %v", si.State.Msgs)
}
return nil
})
inbox := nats.NewInbox()
sub, _ := nc.SubscribeSync(inbox)
defer sub.Unsubscribe()
ci, err := js.AddConsumer("KV", &nats.ConsumerConfig{
DeliverSubject: inbox,
AckPolicy: nats.AckExplicitPolicy,
DeliverPolicy: nats.DeliverLastPerSubjectPolicy,
FilterSubject: "KV.>",
})
if err != nil {
t.Fatalf("Error adding consumer: %v", err)
}
if ci.NumPending != 4 {
t.Fatalf("Expected 4 pending msgs, got %v", ci.NumPending)
}
})
ci, err := js.AddConsumer("KV", &nats.ConsumerConfig{
DeliverSubject: nats.NewInbox(),
AckPolicy: nats.AckExplicitPolicy,
DeliverPolicy: nats.DeliverLastPerSubjectPolicy,
FilterSubject: "KV.>",
})
if err != nil {
t.Fatalf("Error adding consumer: %v", err)
}
if ci.NumPending != 4 {
t.Fatalf("Expected 4 pending msgs, got %v", ci.NumPending)
}
}