Function TestQueue_concurrently

pkg/queue/queue_test.go:482–530

(t *testing.T)

}

func TestQueue_concurrently(t *testing.T) {
	dir := filepath.Join(t.TempDir(), t.Name())

	q, err := NewQueue(dir, 128*1024*1024)
	assert.NoError(t, err)

	var (
		messages sync.Map
		wait     sync.WaitGroup
	)

	sendMessages := make([]map[string][]byte, 4)

	for i := 0; i < 4; i++ {
		sendMessages[i] = mockMessageData(i, 100)
	}
	wait.Add(4)

	for i := 0; i < 4; i++ {
		msg := sendMessages[i]

		go func() {
			defer wait.Done()

			for k, v := range msg {
				err := q.Put(v)
				messages.Store(k, v)

				if err != nil {
					panic("get err")
				}
			}
		}()
	}

	wait.Wait()

	for i := 0; i < 400; i++ {
		data, err := q.Get(int64(i))
		assert.NoError(t, err)
		messages.Delete(string(data))
	}

	messages.Range(func(key, value interface{}) bool {
		panic("get data")
	})

	q.Close()
}

func TestQueue_GC(t *testing.T) {
	dir := filepath.Join(t.TempDir(), t.Name())