Skip to content

Commit

Permalink
Fix diskqueue_test.go (#15)
Browse files Browse the repository at this point in the history
  • Loading branch information
wentingg2023 committed Sep 26, 2023
1 parent 9ac83ef commit 2bab43a
Showing 1 changed file with 77 additions and 77 deletions.
154 changes: 77 additions & 77 deletions diskqueue_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -131,6 +131,83 @@ func TestDiskQueueRoll(t *testing.T) {
}
}

func TestDiskQueuePeek(t *testing.T) {
l := NewTestLogger(t)
dqName := "test_disk_queue_peek" + strconv.Itoa(int(time.Now().Unix()))
tmpDir, err := ioutil.TempDir("", fmt.Sprintf("nsq-test-%d", time.Now().UnixNano()))
if err != nil {
panic(err)
}
defer os.RemoveAll(tmpDir)
msg := bytes.Repeat([]byte{0}, 10)
ml := int64(len(msg))
dq := New(dqName, tmpDir, 10*(ml+4), int32(ml), 1<<10, 2500, 2*time.Second, l)
defer dq.Close()
NotNil(t, dq)
Equal(t, int64(0), dq.Depth())

t.Run("roll", func(t *testing.T) {
for i := 0; i < 10; i++ {
err := dq.Put(msg)
Nil(t, err)
Equal(t, int64(i+1), dq.Depth())
}

for i := 10; i > 0; i-- {
Equal(t, msg, <-dq.PeekChan())
Equal(t, int64(i), dq.Depth())

Equal(t, msg, <-dq.ReadChan())
Equal(t, int64(i-1), dq.Depth())
}

Nil(t, dq.Empty())
})

t.Run("peek-read", func(t *testing.T) {
for i := 0; i < 10; i++ {
err := dq.Put(msg)
Nil(t, err)
Equal(t, int64(i+1), dq.Depth())
}

for i := 10; i > 0; i-- {
Equal(t, msg, <-dq.PeekChan())
Equal(t, int64(i), dq.Depth())

Equal(t, msg, <-dq.PeekChan())
Equal(t, int64(i), dq.Depth())

Equal(t, msg, <-dq.ReadChan())
Equal(t, int64(i-1), dq.Depth())
}

Nil(t, dq.Empty())
})

t.Run("read-peek", func(t *testing.T) {
for i := 0; i < 10; i++ {
err := dq.Put(msg)
Nil(t, err)
Equal(t, int64(i+1), dq.Depth())
}

for i := 10; i > 1; i-- {
Equal(t, msg, <-dq.PeekChan())
Equal(t, int64(i), dq.Depth())

Equal(t, msg, <-dq.ReadChan())
Equal(t, int64(i-1), dq.Depth())

Equal(t, msg, <-dq.PeekChan())
Equal(t, int64(i-1), dq.Depth())
}

Nil(t, dq.Empty())
})

}

func assertFileNotExist(t *testing.T, fn string) {
f, err := os.OpenFile(fn, os.O_RDONLY, 0600)
Equal(t, (*os.File)(nil), f)
Expand Down Expand Up @@ -1260,80 +1337,3 @@ func benchmarkDiskQueueGet(size int64, b *testing.B) {
<-dq.ReadChan()
}
}

func TestDiskQueuePeek(t *testing.T) {
l := NewTestLogger(t)
dqName := "test_disk_queue_peek" + strconv.Itoa(int(time.Now().Unix()))
tmpDir, err := ioutil.TempDir("", fmt.Sprintf("nsq-test-%d", time.Now().UnixNano()))
if err != nil {
panic(err)
}
defer os.RemoveAll(tmpDir)
msg := bytes.Repeat([]byte{0}, 10)
ml := int64(len(msg))
dq := New(dqName, tmpDir, 10*(ml+4), int32(ml), 1<<10, 2500, 2*time.Second, l)
defer dq.Close()
NotNil(t, dq)
Equal(t, int64(0), dq.Depth())

t.Run("roll", func(t *testing.T) {
for i := 0; i < 10; i++ {
err := dq.Put(msg)
Nil(t, err)
Equal(t, int64(i+1), dq.Depth())
}

for i := 10; i > 0; i-- {
Equal(t, msg, <-dq.PeekChan())
Equal(t, int64(i), dq.Depth())

Equal(t, msg, <-dq.ReadChan())
Equal(t, int64(i-1), dq.Depth())
}

Nil(t, dq.Empty())
})

t.Run("peek-read", func(t *testing.T) {
for i := 0; i < 10; i++ {
err := dq.Put(msg)
Nil(t, err)
Equal(t, int64(i+1), dq.Depth())
}

for i := 10; i > 0; i-- {
Equal(t, msg, <-dq.PeekChan())
Equal(t, int64(i), dq.Depth())

Equal(t, msg, <-dq.PeekChan())
Equal(t, int64(i), dq.Depth())

Equal(t, msg, <-dq.ReadChan())
Equal(t, int64(i-1), dq.Depth())
}

Nil(t, dq.Empty())
})

t.Run("read-peek", func(t *testing.T) {
for i := 0; i < 10; i++ {
err := dq.Put(msg)
Nil(t, err)
Equal(t, int64(i+1), dq.Depth())
}

for i := 10; i > 1; i-- {
Equal(t, msg, <-dq.PeekChan())
Equal(t, int64(i), dq.Depth())

Equal(t, msg, <-dq.ReadChan())
Equal(t, int64(i-1), dq.Depth())

Equal(t, msg, <-dq.PeekChan())
Equal(t, int64(i-1), dq.Depth())
}

Nil(t, dq.Empty())
})

}

0 comments on commit 2bab43a

Please sign in to comment.