124
125
// TestNamespaceReplicationDLQ tests namespace DLQ operations
127
>
maxMessageID := int64(100)
128
>
numMessages := 100
129
>
concurrentSenders := 10
130
>
131
>
messageChan := make(chan *replicationspb.ReplicationTask)
132
>
133
>
taskType := enumsspb.REPLICATION_TASK_TYPE_NAMESPACE_TASK
134
>
go func() {
135
>
for i := range numMessages {
136
>
messageChan <- &replicationspb.ReplicationTask{
137
>
TaskType: taskType,
138
>
Attributes: &replicationspb.ReplicationTask_NamespaceTaskAttributes{
139
>
NamespaceTaskAttributes: &replicationspb.NamespaceTaskAttributes{
140
>
Id: fmt.Sprintf("message-%v", i),
141
>
},
142
>
},
143
>
}
144
>
}
145
>
close(messageChan)
146
}()
147
149
>
wg.Add(concurrentSenders)
150
>
151
>
for i := range concurrentSenders {
152
>
go func(senderNum int) {
153
>
defer wg.Done()
154
>
for message := range messageChan {
155
>
err := s.PublishToNamespaceDLQ(s.ctx, message)
156
>
id := message.Attributes.(*replicationspb.ReplicationTask_NamespaceTaskAttributes).NamespaceTaskAttributes.Id
157
>
s.Nil(err, "Enqueue message failed when sender %d tried to send %s", senderNum, id)
158
>
}
159
}(i)
160
}
161
163
>
164
>
result1, token, err := s.GetMessagesFromNamespaceDLQ(s.ctx, persistence.EmptyQueueMessageID, maxMessageID, numMessages/2, nil)
165
>
s.Nil(err, "GetReplicationMessages failed.")
166
>
s.NotNil(token)
167
>
result2, token, err := s.GetMessagesFromNamespaceDLQ(s.ctx, persistence.EmptyQueueMessageID, maxMessageID, numMessages, token)
168
>
s.Nil(err, "GetReplicationMessages failed.")
169
>
s.Equal(len(token), 0)
170
>
s.Equal(len(result1)+len(result2), numMessages)
171
>
_, _, err = s.GetMessagesFromNamespaceDLQ(s.ctx, persistence.EmptyQueueMessageID, 1<<63-1, numMessages, nil)
172
>
s.NoError(err, "GetReplicationMessages failed.")
173
>
s.Equal(len(token), 0)
174
>
175
>
lastMessageID := result2[len(result2)-1].SourceTaskId
176
>
err = s.DeleteMessageFromNamespaceDLQ(s.ctx, lastMessageID)
177
>
s.NoError(err)
178
>
result3, token, err := s.GetMessagesFromNamespaceDLQ(s.ctx, persistence.EmptyQueueMessageID, maxMessageID, numMessages, token)
179
>
s.Nil(err, "GetReplicationMessages failed.")
180
>
s.Equal(len(token), 0)
181
>
s.Equal(len(result3), numMessages-1)
182
>
183
>
err = s.RangeDeleteMessagesFromNamespaceDLQ(s.ctx, persistence.EmptyQueueMessageID, lastMessageID)
184
>
s.NoError(err)
185
>
result4, token, err := s.GetMessagesFromNamespaceDLQ(s.ctx, persistence.EmptyQueueMessageID, maxMessageID, numMessages, token)
186
>
s.Nil(err, "GetReplicationMessages failed.")
187
>
s.Equal(len(token), 0)
188
>
s.Equal(len(result4), 0)
189
}
190