Skip to content

Commit 8b1b84a

Browse files
qyryqqyryq
authored
YMQ: fix ReceiveMessage with attributes (#10142)
1 parent 2e92a1c commit 8b1b84a

File tree

2 files changed

+115
-37
lines changed

2 files changed

+115
-37
lines changed

‎ydb/core/http_proxy/ut/http_proxy_ut.h

+92-11
Original file line numberDiff line numberDiff line change
@@ -1665,7 +1665,7 @@ Y_UNIT_TEST_SUITE(TestHttpProxy) {
16651665

16661666
Y_UNIT_TEST_F(TestReceiveMessage, THttpProxyTestMock) {
16671667
auto createQueueReq = CreateSqsCreateQueueRequest();
1668-
auto res = SendHttpRequest("/Root", "AmazonSQS.CreateQueue", std::move(createQueueReq), FormAuthorizationStr("ru-central1"));
1668+
auto res = SendHttpRequest("/Root", "AmazonSQS.CreateQueue", createQueueReq, FormAuthorizationStr("ru-central1"));
16691669
UNIT_ASSERT_VALUES_EQUAL(res.HttpCode, 200);
16701670
NJson::TJsonValue json;
16711671
UNIT_ASSERT(NJson::ReadJsonTree(res.Body, &json));
@@ -1674,29 +1674,110 @@ Y_UNIT_TEST_SUITE(TestHttpProxy) {
16741674

16751675
NJson::TJsonValue sendMessageReq;
16761676
sendMessageReq["QueueUrl"] = resultQueueUrl;
1677-
auto body = "MessageBody-0";
1678-
sendMessageReq["MessageBody"] = body;
1679-
sendMessageReq["MessageBody"] = body;
1677+
auto body0 = "MessageBody-0";
1678+
sendMessageReq["MessageBody"] = body0;
16801679

1681-
res = SendHttpRequest("/Root", "AmazonSQS.SendMessage", std::move(sendMessageReq), FormAuthorizationStr("ru-central1"));
1680+
res = SendHttpRequest("/Root", "AmazonSQS.SendMessage", sendMessageReq, FormAuthorizationStr("ru-central1"));
16821681
UNIT_ASSERT(NJson::ReadJsonTree(res.Body, &json));
16831682
UNIT_ASSERT(!GetByPath<TString>(json, "MD5OfMessageBody").empty());
16841683
UNIT_ASSERT_VALUES_EQUAL(res.HttpCode, 200);
16851684

1685+
NJson::TJsonValue receiveMessageReq;
1686+
receiveMessageReq["QueueUrl"] = resultQueueUrl;
16861687
for (int i = 0; i < 20; ++i) {
1687-
NJson::TJsonValue receiveMessageReq;
1688-
receiveMessageReq["QueueUrl"] = resultQueueUrl;
1689-
res = SendHttpRequest("/Root", "AmazonSQS.ReceiveMessage", std::move(receiveMessageReq), FormAuthorizationStr("ru-central1"));
1690-
if (res.Body != TString("{}")) {
1691-
break;;
1688+
res = SendHttpRequest("/Root", "AmazonSQS.ReceiveMessage", receiveMessageReq, FormAuthorizationStr("ru-central1"));
1689+
if (res.Body != "{}") {
1690+
break;
16921691
}
16931692
std::this_thread::sleep_for(std::chrono::milliseconds(1000));
16941693
}
16951694

16961695
UNIT_ASSERT(NJson::ReadJsonTree(res.Body, &json));
16971696
UNIT_ASSERT_VALUES_EQUAL(res.HttpCode, 200);
16981697
UNIT_ASSERT_VALUES_EQUAL(json["Messages"].GetArray().size(), 1);
1699-
UNIT_ASSERT_VALUES_EQUAL(json["Messages"][0]["Body"], body);
1698+
UNIT_ASSERT_VALUES_EQUAL(json["Messages"][0]["Body"], body0);
1699+
}
1700+
1701+
Y_UNIT_TEST_F(TestReceiveMessageWithAttributes, THttpProxyTestMock) {
1702+
// Test if we process AttributeNames, MessageSystemAttributeNames, MessageAttributeNames correctly.
1703+
1704+
auto createQueueReq = CreateSqsCreateQueueRequest();
1705+
auto res = SendHttpRequest("/Root", "AmazonSQS.CreateQueue", createQueueReq, FormAuthorizationStr("ru-central1"));
1706+
UNIT_ASSERT_VALUES_EQUAL(res.HttpCode, 200);
1707+
NJson::TJsonValue json;
1708+
UNIT_ASSERT(NJson::ReadJsonTree(res.Body, &json));
1709+
TString resultQueueUrl = GetByPath<TString>(json, "QueueUrl");
1710+
UNIT_ASSERT(resultQueueUrl.EndsWith("ExampleQueueName"));
1711+
1712+
auto sendMessage = [this, resultQueueUrl](const TString& body) {
1713+
NJson::TJsonValue sendMessageReq;
1714+
sendMessageReq["QueueUrl"] = resultQueueUrl;
1715+
sendMessageReq["MessageBody"] = body;
1716+
1717+
auto res = SendHttpRequest("/Root", "AmazonSQS.SendMessage", sendMessageReq, FormAuthorizationStr("ru-central1"));
1718+
NJson::TJsonValue json;
1719+
UNIT_ASSERT(NJson::ReadJsonTree(res.Body, &json));
1720+
UNIT_ASSERT(!GetByPath<TString>(json, "MD5OfMessageBody").empty());
1721+
UNIT_ASSERT_VALUES_EQUAL(res.HttpCode, 200);
1722+
};
1723+
1724+
TString body = "MessageBody-0";
1725+
sendMessage(body);
1726+
1727+
auto receiveMessage = [this](NJson::TJsonValue request, const TString& expectedBody) -> NJson::TJsonValue {
1728+
request["VisibilityTimeout"] = 0; // Keep the message visible for next ReceiveMessage requests.
1729+
THttpResult res;
1730+
for (int i = 0; i < 20; ++i) {
1731+
res = SendHttpRequest("/Root", "AmazonSQS.ReceiveMessage", request, FormAuthorizationStr("ru-central1"));
1732+
if (res.Body != "{}") {
1733+
break;
1734+
}
1735+
std::this_thread::sleep_for(std::chrono::milliseconds(1000));
1736+
}
1737+
1738+
NJson::TJsonValue json;
1739+
UNIT_ASSERT(NJson::ReadJsonTree(res.Body, &json));
1740+
UNIT_ASSERT_VALUES_EQUAL(res.HttpCode, 200);
1741+
UNIT_ASSERT_VALUES_EQUAL(json["Messages"].GetArray().size(), 1);
1742+
UNIT_ASSERT_VALUES_EQUAL(json["Messages"][0]["Body"], expectedBody);
1743+
return json;
1744+
};
1745+
1746+
{
1747+
// Request SentTimestamp message system attribute using deprecated AttributeNames field.
1748+
NJson::TJsonValue receiveMessageReq;
1749+
receiveMessageReq["QueueUrl"] = resultQueueUrl;
1750+
receiveMessageReq["AttributeNames"] = NJson::TJsonArray{"SentTimestamp"};
1751+
json = receiveMessage(receiveMessageReq, body);
1752+
UNIT_ASSERT(!json["Messages"][0]["Attributes"]["SentTimestamp"].GetString().empty());
1753+
}
1754+
1755+
{
1756+
// Request SentTimestamp message system attribute using MessageSystemAttributeNames field.
1757+
NJson::TJsonValue receiveMessageReq;
1758+
receiveMessageReq["QueueUrl"] = resultQueueUrl;
1759+
receiveMessageReq["MessageSystemAttributeNames"] = NJson::TJsonArray{"SentTimestamp"};
1760+
json = receiveMessage(receiveMessageReq, body);
1761+
UNIT_ASSERT(!json["Messages"][0]["Attributes"]["SentTimestamp"].GetString().empty());
1762+
}
1763+
1764+
{
1765+
// Request All message system attributes using deprecated AttributeNames field.
1766+
NJson::TJsonValue receiveMessageReq;
1767+
receiveMessageReq["QueueUrl"] = resultQueueUrl;
1768+
receiveMessageReq["AttributeNames"] = NJson::TJsonArray{"All"};
1769+
json = receiveMessage(receiveMessageReq, body);
1770+
UNIT_ASSERT(!json["Messages"][0]["Attributes"]["SentTimestamp"].GetString().empty());
1771+
}
1772+
1773+
{
1774+
// Request All message system attributes using MessageSystemAttributeNames field.
1775+
NJson::TJsonValue receiveMessageReq;
1776+
receiveMessageReq["QueueUrl"] = resultQueueUrl;
1777+
receiveMessageReq["MessageSystemAttributeNames"] = NJson::TJsonArray{"All"};
1778+
json = receiveMessage(receiveMessageReq, body);
1779+
UNIT_ASSERT(!json["Messages"][0]["Attributes"]["SentTimestamp"].GetString().empty());
1780+
}
17001781
}
17011782

17021783
Y_UNIT_TEST_F(TestGetQueueAttributes, THttpProxyTestMock) {

‎ydb/services/ymq/ymq_proxy.cpp

+23-26
Original file line numberDiff line numberDiff line change
@@ -322,29 +322,26 @@ namespace NKikimr::NYmq::V1 {
322322
for (const auto& srcMessage: GetResponse(resp).GetMessages()) {
323323
Ydb::Ymq::V1::Message dstMessage;
324324

325-
for (TString& attributeName : AttributesNames) {
326-
if (attributeName == APPROXIMATE_RECEIVE_COUNT) {
327-
dstMessage.Mutableattributes()->at(attributeName)
328-
.assign(srcMessage.GetApproximateReceiveCount());
329-
} else if (attributeName == APPROXIMATE_FIRST_RECEIVE_TIMESTAMP) {
330-
dstMessage.Mutableattributes()->at(attributeName)
331-
.assign(srcMessage.GetApproximateFirstReceiveTimestamp());
332-
} else if (attributeName == MESSAGE_DEDUPLICATION_ID) {
333-
dstMessage.Mutableattributes()->at(attributeName)
334-
.assign(srcMessage.GetMessageDeduplicationId());
335-
} else if (attributeName == MESSAGE_GROUP_ID) {
336-
dstMessage.Mutableattributes()->at(attributeName)
337-
.assign(srcMessage.GetMessageGroupId());
338-
} else if (attributeName == SENDER_ID) {
339-
dstMessage.Mutableattributes()->at(attributeName)
340-
.assign(srcMessage.GetSenderId());
341-
} else if (attributeName == SENT_TIMESTAMP) {
342-
dstMessage.Mutableattributes()->at(attributeName)
343-
.assign(srcMessage.GetSentTimestamp());
344-
} else if (attributeName == SEQUENCE_NUMBER) {
345-
dstMessage.Mutableattributes()->at(attributeName)
346-
.assign(srcMessage.GetSequenceNumber());
347-
}
325+
if (srcMessage.HasApproximateReceiveCount()) {
326+
dstMessage.Mutableattributes()->insert({APPROXIMATE_RECEIVE_COUNT, std::to_string(srcMessage.GetApproximateReceiveCount())});
327+
}
328+
if (srcMessage.HasApproximateFirstReceiveTimestamp()) {
329+
dstMessage.Mutableattributes()->insert({APPROXIMATE_FIRST_RECEIVE_TIMESTAMP, std::to_string(srcMessage.GetApproximateFirstReceiveTimestamp())});
330+
}
331+
if (srcMessage.HasMessageDeduplicationId()) {
332+
dstMessage.Mutableattributes()->insert({MESSAGE_DEDUPLICATION_ID, srcMessage.GetMessageDeduplicationId()});
333+
}
334+
if (srcMessage.HasMessageGroupId()) {
335+
dstMessage.Mutableattributes()->insert({MESSAGE_GROUP_ID, srcMessage.GetMessageGroupId()});
336+
}
337+
if (srcMessage.HasSenderId()) {
338+
dstMessage.Mutableattributes()->insert({SENDER_ID, srcMessage.GetSenderId()});
339+
}
340+
if (srcMessage.HasSentTimestamp()) {
341+
dstMessage.Mutableattributes()->insert({SENT_TIMESTAMP, std::to_string(srcMessage.GetSentTimestamp())});
342+
}
343+
if (srcMessage.HasSequenceNumber()) {
344+
dstMessage.Mutableattributes()->insert({SEQUENCE_NUMBER, std::to_string(srcMessage.GetSequenceNumber())});
348345
}
349346

350347
dstMessage.set_body(srcMessage.GetData());
@@ -392,17 +389,17 @@ namespace NKikimr::NYmq::V1 {
392389
// because AttributeNames is deprecated in favour of SystemAttributeNames
393390
if (systemAttributeNames.size() > 0) {
394391
for (int i = 0; i < systemAttributeNames.size(); i++) {
395-
result->SetAttributeName(i, systemAttributeNames[i]);
392+
result->AddAttributeName(systemAttributeNames[i]);
396393
}
397394
} else {
398395
auto attributeNames = GetProtoRequest()->Getattribute_names();
399396
for (int i = 0; i < attributeNames.size(); i++) {
400-
result->SetAttributeName(i, attributeNames[i]);
397+
result->AddAttributeName(attributeNames[i]);
401398
}
402399
}
403400

404401
for (int i = 0; i < GetProtoRequest()->Getmessage_attribute_names().size(); i++) {
405-
result->SetMessageAttributeName(i, GetProtoRequest()->Getmessage_attribute_names()[i]);
402+
result->AddMessageAttributeName(GetProtoRequest()->Getmessage_attribute_names()[i]);
406403
}
407404

408405
return result;

0 commit comments

Comments
 (0)