48
49
// TestNamespaceReplicationQueue tests namespace replication queue operations
51
>
numMessages := 100
52
>
concurrentSenders := 10
53
>
54
>
messageChan := make(chan *replicationspb.ReplicationTask)
55
>
56
>
taskType := enumsspb.REPLICATION_TASK_TYPE_NAMESPACE_TASK
57
>
go func() {
58
>
for i := range numMessages {
59
>
messageChan <- &replicationspb.ReplicationTask{
60
>
TaskType: taskType,
61
>
Attributes: &replicationspb.ReplicationTask_NamespaceTaskAttributes{
62
>
NamespaceTaskAttributes: &replicationspb.NamespaceTaskAttributes{
63
>
Id: fmt.Sprintf("message-%v", i),
64
>
},
65
>
},
66
>
}
67
>
}
68
>
close(messageChan)
69
}()
70
72
>
wg.Add(concurrentSenders)
73
>
74
>
for i := range concurrentSenders {
75
>
go func(senderNum int) {
76
>
defer wg.Done()
77
>
for message := range messageChan {
78
>
err := s.Publish(s.ctx, message)
79
>
id := message.Attributes.(*replicationspb.ReplicationTask_NamespaceTaskAttributes).NamespaceTaskAttributes.Id
80
>
s.Nil(err, "Enqueue message failed when sender %d tried to send %s", senderNum, id)
81
>
}
82
}(i)
83
}
84
86
>
87
>
result, lastRetrievedMessageID, err := s.GetReplicationMessages(s.ctx, persistence.EmptyQueueMessageID, numMessages)
88
>
s.Nil(err, "GetReplicationMessages failed.")
89
>
s.Len(result, numMessages)
90
>
s.Equal(int64(numMessages-1), lastRetrievedMessageID)
91
}
92