117
}
118
120
>
queueType := persistence.NamespaceReplicationQueueType
121
>
messageID := rand.Int63()
122
>
123
>
message := s.newRandomQueueMessageRow(queueType, messageID)
124
>
result, err := s.store.InsertIntoMessages(newExecutionContext(), []sqlplugin.QueueMessageRow{message})
125
>
s.NoError(err)
126
>
rowsAffected, err := result.RowsAffected()
127
>
s.NoError(err)
128
>
s.Equal(1, int(rowsAffected))
129
>
130
>
filter := sqlplugin.QueueMessagesFilter{
131
>
QueueType: queueType,
132
>
MessageID: messageID,
133
>
}
134
>
rows, err := s.store.SelectFromMessages(newExecutionContext(), filter)
135
>
s.NoError(err)
136
>
for index := range rows {
137
>
rows[index].QueueType = queueType
138
>
}
139
>
s.Equal([]sqlplugin.QueueMessageRow{message}, rows)
140
}
141