Add ChangeMessageVisibilityBatch request to SQS Json API progress

Author: siarheivesialou

ydb/core/grpc_services/service_ymq.h

@@ -24,5 +24,6 @@ void DoYmqChangeMessageVisibilityRequest(std::unique_ptr<IRequestOpCtx> p, const
 void DoYmqSetQueueAttributesRequest(std::unique_ptr<IRequestOpCtx> p, const IFacilityProvider& f);
 void DoYmqSendMessageBatchRequest(std::unique_ptr<IRequestOpCtx> p, const IFacilityProvider& f);
 void DoYmqDeleteMessageBatchRequest(std::unique_ptr<IRequestOpCtx> p, const IFacilityProvider& f);
+void DoYmqChangeMessageVisibilityBatchRequest(std::unique_ptr<IRequestOpCtx> p, const IFacilityProvider& f);
 }
 }

ydb/core/http_proxy/http_req.cpp

@@ -534,6 +534,8 @@ namespace NKikimr::NHttpProxy {
                         action = NSQS::EAction::SendMessageBatch;
                     }else if (Method == "DeleteMessageBatch") {
                         action = NSQS::EAction::DeleteMessageBatch;
+                    } else if (Method == "ChangeMessageVisibilityBatch") {
+                        action = NSQS::EAction::ChangeMessageVisibilityBatch;
                     }
 
                     requestHolder->SetRequestId(HttpContext.RequestId);
@@ -1072,6 +1074,7 @@ namespace NKikimr::NHttpProxy {
         DECLARE_YMQ_PROCESSOR_QUEUE_KNOWN(SetQueueAttributes);
         DECLARE_YMQ_PROCESSOR_QUEUE_KNOWN(SendMessageBatch);
         DECLARE_YMQ_PROCESSOR_QUEUE_KNOWN(DeleteMessageBatch);
+        DECLARE_YMQ_PROCESSOR_QUEUE_KNOWN(ChangeMessageVisibilityBatch);
         #undef DECLARE_YMQ_PROCESSOR_QUEUE_KNOWN
     }
 

ydb/core/http_proxy/ut/http_proxy_ut.h

@@ -1995,4 +1995,5 @@ Y_UNIT_TEST_SUITE(TestHttpProxy) {
         UNIT_ASSERT_VALUES_EQUAL(json["Messages"].GetArray().size(), 0);
 
     }
+
 } // Y_UNIT_TEST_SUITE(TestHttpProxy)

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

@@ -23,4 +23,5 @@ service YmqService {
     rpc SetQueueAttributes(SetQueueAttributesRequest) returns (SetQueueAttributesResponse);
     rpc SendMessageBatch(SendMessageBatchRequest) returns (SendMessageBatchResponse);
     rpc DeleteMessageBatch(DeleteMessageBatchRequest) returns (DeleteMessageBatchResponse);
+    rpc ChangeMessageVisibilityBatch(ChangeMessageVisibilityBatchRequest) returns (ChangeMessageVisibilityBatchResponse);
 }

ydb/services/ymq/grpc_service.cpp

@@ -45,6 +45,7 @@ void TGRpcYmqService::SetupIncomingRequests(NYdbGrpc::TLoggerPtr logger)
     ADD_REQUEST(SetQueueAttributes, DoYmqSetQueueAttributesRequest, nullptr, Off)
     ADD_REQUEST(SendMessageBatch, DoYmqSendMessageBatchRequest, nullptr, Off)
     ADD_REQUEST(DeleteMessageBatch, DoYmqDeleteMessageBatchRequest, nullptr, Off)
+    ADD_REQUEST(ChangeMessageVisibilityBatch, DoYmqChangeMessageVisibilityBatchRequest, nullptr, Off)
 
 #undef ADD_REQUEST
 }

ydb/services/ymq/ymq_proxy.cpp

@@ -854,6 +854,58 @@ namespace NKikimr::NYmq::V1 {
             return result;
         }
     };
+
+    class TChangeMessageVisibilityBatchReplyCallback : public TReplyCallback<
+            NKikimr::NSQS::TChangeMessageVisibilityBatchResponse,
+            Ydb::Ymq::V1::ChangeMessageVisibilityBatchResult> {
+    public:
+        using TReplyCallback::TReplyCallback;
+
+    private:
+        const NKikimr::NSQS::TChangeMessageVisibilityBatchResponse& GetResponse(const NKikimrClient::TSqsResponse& resp) override {
+            return resp.GetChangeMessageVisibilityBatch();
+        }
+
+        Ydb::Ymq::V1::ChangeMessageVisibilityBatchResult GetResult(const NKikimrClient::TSqsResponse& response) override {
+            Ydb::Ymq::V1::ChangeMessageVisibilityBatchResult result;
+            for (auto& entry : response.GetChangeMessageVisibilityBatch().GetEntries()) {
+                if (entry.GetError().HasErrorCode()) {
+                    auto currentFailed = result.Addfailed();
+                    currentFailed->Setcode(entry.GetError().GetErrorCode());
+                    currentFailed->Setid(entry.GetId());
+                    currentFailed->Setmessage(entry.GetError().GetMessage());
+
+                    ui32 httpStatus = NSQS::TErrorClass::GetHttpStatus(entry.GetError().GetErrorCode()).GetOrElse(400);
+                    currentFailed->Setsender_fault(400 <= httpStatus && httpStatus < 500);
+                } else {
+                    auto currentSuccessful = result.Addsuccessful();
+                    currentSuccessful->Setid(entry.GetId());
+                }
+            }
+            return result;
+        }
+    };
+
+    class TChangeMessageVisibilityBatchActor : public TRpcRequestActor<
+            TEvYmqChangeMessageVisibilityBatchRequest,
+            NKikimr::NSQS::TChangeMessageVisibilityBatchRequest,
+            TChangeMessageVisibilityBatchReplyCallback> {
+    public:
+        using TRpcRequestActor::TRpcRequestActor;
+
+    private:
+        NKikimr::NSQS::TChangeMessageVisibilityBatchRequest* GetRequest(THolder<TSqsRequest>& requestHolder) override {
+            auto result = requestHolder->MutableChangeMessageVisibilityBatch();
+            result->SetQueueName(CloudIdAndResourceIdFromQueueUrl(GetProtoRequest()->Getqueue_url())->second);
+            for (auto& requestEntry : GetProtoRequest()->Getentries()) {
+                auto entry = requestHolder->MutableChangeMessageVisibilityBatch()->MutableEntries()->Add();
+                entry->SetId(requestEntry.Getid());
+                entry->SetVisibilityTimeout(requestEntry.Getvisibility_timeout());
+                entry->SetReceiptHandle(requestEntry.Getreceipt_handle());
+            }
+            return result;
+        }
+    };
 }
 
 namespace NKikimr::NGRpcService {
@@ -880,5 +932,6 @@ DECLARE_RPC(ChangeMessageVisibility);
 DECLARE_RPC(SetQueueAttributes);
 DECLARE_RPC(SendMessageBatch);
 DECLARE_RPC(DeleteMessageBatch);
+DECLARE_RPC(ChangeMessageVisibilityBatch);
 
 }

ydb/services/ymq/ymq_proxy.h

@@ -26,6 +26,7 @@ using TEvYmqChangeMessageVisibilityRequest = TGrpcRequestOperationCall<Ydb::Ymq:
 using TEvYmqSetQueueAttributesRequest = TGrpcRequestOperationCall<Ydb::Ymq::V1::SetQueueAttributesRequest, Ydb::Ymq::V1::SetQueueAttributesResponse>;
 using TEvYmqSendMessageBatchRequest = TGrpcRequestOperationCall<Ydb::Ymq::V1::SendMessageBatchRequest, Ydb::Ymq::V1::SendMessageBatchResponse>;
 using TEvYmqDeleteMessageBatchRequest = TGrpcRequestOperationCall<Ydb::Ymq::V1::DeleteMessageBatchRequest, Ydb::Ymq::V1::DeleteMessageBatchResponse>;
+using TEvYmqChangeMessageVisibilityBatchRequest = TGrpcRequestOperationCall<Ydb::Ymq::V1::ChangeMessageVisibilityBatchRequest, Ydb::Ymq::V1::ChangeMessageVisibilityBatchResponse>;
 
 }
 }