SetAttributes request

Author: siarheivesialou

ydb/core/grpc_services/service_ymq.h

@@ -21,5 +21,6 @@ void DoYmqDeleteMessageRequest(std::unique_ptr<IRequestOpCtx> p, const IFacility
 void DoYmqPurgeQueueRequest(std::unique_ptr<IRequestOpCtx> p, const IFacilityProvider& f);
 void DoYmqDeleteQueueRequest(std::unique_ptr<IRequestOpCtx> p, const IFacilityProvider& f);
 void DoYmqChangeMessageVisibilityRequest(std::unique_ptr<IRequestOpCtx> p, const IFacilityProvider& f);
+void DoYmqSetQueueAttributesRequest(std::unique_ptr<IRequestOpCtx> p, const IFacilityProvider& f);
 }
 }

ydb/core/http_proxy/http_req.cpp

@@ -528,6 +528,8 @@ namespace NKikimr::NHttpProxy {
                         action = NSQS::EAction::DeleteQueue;
                     } else if (Method == "ChangeMessageVisibility") {
                         action = NSQS::EAction::ChangeMessageVisibility;
+                    } else if (Method == "SetQueueAttributes") {
+                        action = NSQS::EAction::SetQueueAttributes;
                     }
 
                     requestHolder->SetRequestId(HttpContext.RequestId);
@@ -1063,6 +1065,7 @@ namespace NKikimr::NHttpProxy {
         DECLARE_YMQ_PROCESSOR_QUEUE_KNOWN(PurgeQueue);
         DECLARE_YMQ_PROCESSOR_QUEUE_KNOWN(DeleteQueue);
         DECLARE_YMQ_PROCESSOR_QUEUE_KNOWN(ChangeMessageVisibility);
+        DECLARE_YMQ_PROCESSOR_QUEUE_KNOWN(SetQueueAttributes);
         #undef DECLARE_YMQ_PROCESSOR_QUEUE_KNOWN
     }
 

ydb/core/http_proxy/ut/http_proxy_ut.h

@@ -1838,4 +1838,38 @@ Y_UNIT_TEST_SUITE(TestHttpProxy) {
         UNIT_ASSERT(NJson::ReadJsonTree(res.Body, &json));
         UNIT_ASSERT_VALUES_EQUAL(GetByPath<TString>(json, "__type"), "AWS.SimpleQueueService.NonExistentQueue");
     }
+
+    Y_UNIT_TEST_F(TestSetQueueAttributes, THttpProxyTestMock) {
+        auto createQueueReq = CreateSqsCreateQueueRequest();
+        NJson::TJsonValue attributes;
+        attributes["DelaySeconds"] = "1";
+        createQueueReq["Attributes"] = attributes;
+
+        auto res = SendHttpRequest("/Root", "AmazonSQS.CreateQueue", std::move(createQueueReq), FormAuthorizationStr("ru-central1"));
+        UNIT_ASSERT_VALUES_EQUAL(res.HttpCode, 200);
+        NJson::TJsonValue json;
+        UNIT_ASSERT(NJson::ReadJsonTree(res.Body, &json));
+
+        TString resultQueueUrl = GetByPath<TString>(json, "QueueUrl");
+
+        NJson::TJsonValue setQueueAttributes;
+        setQueueAttributes["QueueUrl"] = resultQueueUrl;
+        attributes = {};
+        attributes["DelaySeconds"] = "2";
+        setQueueAttributes["Attributes"] = attributes;
+
+        res = SendHttpRequest("/Root", "AmazonSQS.SetQueueAttributes", std::move(setQueueAttributes), FormAuthorizationStr("ru-central1"));
+        UNIT_ASSERT_VALUES_EQUAL(res.HttpCode, 200);
+
+        NJson::TJsonValue getQueueAttributes;
+        getQueueAttributes["QueueUrl"] = resultQueueUrl;
+        NJson::TJsonArray attributeNames = {"DelaySeconds"};
+        getQueueAttributes["AttributeNames"] = attributeNames;
+
+        res = SendHttpRequest("/Root", "AmazonSQS.GetQueueAttributes", std::move(getQueueAttributes), FormAuthorizationStr("ru-central1"));
+        UNIT_ASSERT_VALUES_EQUAL(res.HttpCode, 200);
+        NJson::TJsonValue resultJson;
+        UNIT_ASSERT(NJson::ReadJsonTree(res.Body, &resultJson));
+        UNIT_ASSERT_VALUES_EQUAL(resultJson["Attributes"]["DelaySeconds"], "2");
+    }
 } // Y_UNIT_TEST_SUITE(TestHttpProxy)

ydb/public/api/grpc/draft/ydb_ymq_v1.proto

@@ -20,4 +20,5 @@ service YmqService {
     rpc PurgeQueue(PurgeQueueRequest) returns (PurgeQueueResponse);
     rpc DeleteQueue(DeleteQueueRequest) returns (DeleteQueueResponse);
     rpc ChangeMessageVisibility(ChangeMessageVisibilityRequest) returns (ChangeMessageVisibilityResponse);
+    rpc SetQueueAttributes(SetQueueAttributesRequest) returns (SetQueueAttributesResponse);
 }

ydb/public/api/protos/draft/ymq.proto

@@ -256,3 +256,16 @@ message SendMessageBatchResult {
     repeated BatchResultErrorEntry failed = 1;
     repeated SendMessageBatchResultEntry successful = 2;
 }
+
+message SetQueueAttributesRequest {
+    Ydb.Operations.OperationParams operation_params = 1;
+    map<string, string> attributes = 2;
+    string queue_url = 3;
+}
+
+message SetQueueAttributesResponse {
+    Ydb.Operations.Operation operation = 1;
+}
+
+message SetQueueAttributesResult {
+}

ydb/services/ymq/grpc_service.cpp

@@ -42,6 +42,7 @@ void TGRpcYmqService::SetupIncomingRequests(NYdbGrpc::TLoggerPtr logger)
     ADD_REQUEST(PurgeQueue, DoYmqPurgeQueueRequest, nullptr, Off)
     ADD_REQUEST(DeleteQueue, DoYmqDeleteQueueRequest, nullptr, Off)
     ADD_REQUEST(ChangeMessageVisibility, DoYmqChangeMessageVisibilityRequest, nullptr, Off)
+    ADD_REQUEST(SetQueueAttributes, DoYmqSetQueueAttributesRequest, nullptr, Off)
 
 #undef ADD_REQUEST
 }

ydb/services/ymq/ymq_proxy.cpp

@@ -694,6 +694,47 @@ namespace NKikimr::NYmq::V1 {
             return result;
         }
     };
+
+    class TSetQueueAttributesReplyCallback : public TReplyCallback<
+            NKikimr::NSQS::TSetQueueAttributesResponse,
+            Ydb::Ymq::V1::SetQueueAttributesResult> {
+    public:
+        using TReplyCallback::TReplyCallback;
+
+    private:
+        const NKikimr::NSQS::TSetQueueAttributesResponse& GetResponse(const NKikimrClient::TSqsResponse& resp) override {
+            return resp.GetSetQueueAttributes();
+        }
+
+        Ydb::Ymq::V1::SetQueueAttributesResult GetResult(const NKikimrClient::TSqsResponse&) override {
+            Ydb::Ymq::V1::SetQueueAttributesResult result;
+            return result;
+        }
+    };
+
+    void AddAttribute(THolder<TSqsRequest>& requestHolder, const TString& name, TString value) {
+        auto attribute = requestHolder->MutableSetQueueAttributes()->MutableAttributes()->Add();
+        attribute->SetName(name);
+        attribute->SetValue(value);
+    };
+
+    class TSetQueueAttributesActor : public TRpcRequestActor<
+            TEvYmqSetQueueAttributesRequest,
+            NKikimr::NSQS::TSetQueueAttributesRequest,
+            TSetQueueAttributesReplyCallback> {
+    public:
+        using TRpcRequestActor::TRpcRequestActor;
+
+    private:
+        NKikimr::NSQS::TSetQueueAttributesRequest* GetRequest(THolder<TSqsRequest>& requestHolder) override {
+            auto result = requestHolder->MutableSetQueueAttributes();
+            for (auto& [name, value]: GetProtoRequest()->Getattributes()) {
+                AddAttribute(requestHolder, name, value);
+            }
+            result->SetQueueName(CloudIdAndResourceIdFromQueueUrl(GetProtoRequest()->queue_url())->second);
+            return result;
+        }
+    };
 }
 
 namespace NKikimr::NGRpcService {
@@ -717,5 +758,6 @@ DECLARE_RPC(DeleteMessage);
 DECLARE_RPC(PurgeQueue);
 DECLARE_RPC(DeleteQueue);
 DECLARE_RPC(ChangeMessageVisibility);
+DECLARE_RPC(SetQueueAttributes);
 
 }

ydb/services/ymq/ymq_proxy.h

@@ -23,6 +23,7 @@ using TEvYmqDeleteMessageRequest = TGrpcRequestOperationCall<Ydb::Ymq::V1::Delet
 using TEvYmqPurgeQueueRequest = TGrpcRequestOperationCall<Ydb::Ymq::V1::PurgeQueueRequest, Ydb::Ymq::V1::PurgeQueueResponse>;
 using TEvYmqDeleteQueueRequest = TGrpcRequestOperationCall<Ydb::Ymq::V1::DeleteQueueRequest, Ydb::Ymq::V1::DeleteQueueResponse>;
 using TEvYmqChangeMessageVisibilityRequest = TGrpcRequestOperationCall<Ydb::Ymq::V1::ChangeMessageVisibilityRequest, Ydb::Ymq::V1::ChangeMessageVisibilityResponse>;
+using TEvYmqSetQueueAttributesRequest = TGrpcRequestOperationCall<Ydb::Ymq::V1::SetQueueAttributesRequest, Ydb::Ymq::V1::SetQueueAttributesResponse>;
 
 }
 }