501
})
502
t.Run("QueueSize", func(t *testing.T) {
504
>
queueName := "test-queue-" + t.Name()
505
>
_, err := queue.CreateQueue(ctx, &persistence.InternalCreateQueueRequest{
506
>
QueueType: queueType,
507
>
QueueName: queueName,
508
>
})
509
>
require.NoError(t, err)
510
>
response, err := queue.ListQueues(ctx, &persistence.InternalListQueuesRequest{
511
>
QueueType: queueType,
512
>
PageSize: 100,
513
>
NextPageToken: nil,
514
>
})
515
>
require.NoError(t, err)
516
>
var queueNames []string
517
>
for _, queue := range response.Queues {
518
>
queueNames = append(queueNames, queue.QueueName)
519
>
if queue.QueueName == queueName {
520
>
assert.Equal(t, int64(0), queue.MessageCount)
521
>
}
522
}
524
>
525
>
// Enqueue one message and verify QueueSize is 1
526
>
_, err = persistencetest.EnqueueMessage(ctx, queue, queueType, queueName)
527
>
require.NoError(t, err)
528
>
response, err = queue.ListQueues(ctx, &persistence.InternalListQueuesRequest{
529
>
QueueType: queueType,
530
>
PageSize: 100,
531
>
NextPageToken: nil,
532
>
})
533
>
require.NoError(t, err)
534
>
for _, queue := range response.Queues {
535
>
queueNames = append(queueNames, queue.QueueName)
536
>
if queue.QueueName == queueName {
537
>
assert.Equal(t, int64(1), queue.MessageCount)
538
>
assert.Equal(t, int64(0), queue.LastMessageID)
539
>
}
540
}
542
>
543
>
// Enqueue one more message and verify QueueSize is 2
544
>
_, err = persistencetest.EnqueueMessage(ctx, queue, queueType, queueName)
545
>
require.NoError(t, err)
546
>
response, err = queue.ListQueues(ctx, &persistence.InternalListQueuesRequest{
547
>
QueueType: queueType,
548
>
PageSize: 100,
549
>
NextPageToken: nil,
550
>
})
551
>
require.NoError(t, err)
552
>
for _, queue := range response.Queues {
553
>
queueNames = append(queueNames, queue.QueueName)
554
>
if queue.QueueName == queueName {
555
>
assert.Equal(t, int64(2), queue.MessageCount)
556
>
assert.Equal(t, int64(1), queue.LastMessageID)
557
>
}
558
}
560
>
561
>
// Delete one message and verify QueueSize is 1
562
>
_, err = queue.RangeDeleteMessages(ctx, &persistence.InternalRangeDeleteMessagesRequest{
563
>
QueueType: queueType,
564
>
QueueName: queueName,
565
>
InclusiveMaxMessageMetadata: persistence.MessageMetadata{
566
>
ID: persistence.FirstQueueMessageID,
567
>
},
568
>
})
569
>
require.NoError(t, err)
570
>
response, err = queue.ListQueues(ctx, &persistence.InternalListQueuesRequest{
571
>
QueueType: queueType,
572
>
PageSize: 100,
573
>
NextPageToken: nil,
574
>
})
575
>
require.NoError(t, err)
576
>
for _, queue := range response.Queues {
577
>
queueNames = append(queueNames, queue.QueueName)
578
>
if queue.QueueName == queueName {
579
>
assert.Equal(t, int64(1), queue.MessageCount)
580
>
assert.Equal(t, int64(1), queue.LastMessageID)
581
>
}
582
}
584
>
585
>
// Delete one more message and verify QueueSize is 0
586
>
_, err = queue.RangeDeleteMessages(ctx, &persistence.InternalRangeDeleteMessagesRequest{
587
>
QueueType: queueType,
588
>
QueueName: queueName,
589
>
InclusiveMaxMessageMetadata: persistence.MessageMetadata{
590
>
ID: persistence.FirstQueueMessageID + 1,
591
>
},
592
>
})
593
>
require.NoError(t, err)
594
>
response, err = queue.ListQueues(ctx, &persistence.InternalListQueuesRequest{
595
>
QueueType: queueType,
596
>
PageSize: 100,
597
>
NextPageToken: nil,
598
>
})
599
>
require.NoError(t, err)
600
>
for _, queue := range response.Queues {
601
>
queueNames = append(queueNames, queue.QueueName)
602
>
if queue.QueueName == queueName {
603
>
assert.Equal(t, int64(0), queue.MessageCount)
604
>
assert.Equal(t, int64(1), queue.LastMessageID)
605
>
}
606
}
608
>
609
>
// Enqueue one message once again and verify QueueSize is 1
610
>
_, err = persistencetest.EnqueueMessage(ctx, queue, queueType, queueName)
611
>
require.NoError(t, err)
612
>
response, err = queue.ListQueues(ctx, &persistence.InternalListQueuesRequest{
613
>
QueueType: queueType,
614
>
PageSize: 100,
615
>
NextPageToken: nil,
616
>
})
617
>
require.NoError(t, err)
618
>
for _, queue := range response.Queues {
619
>
queueNames = append(queueNames, queue.QueueName)
620
>
if queue.QueueName == queueName {
621
>
assert.Equal(t, int64(1), queue.MessageCount)
622
>
}
623
}
625
626
})