Topics improvements for 24-3 (ydb-platform#8605)

Co-authored-by: niksaveliev <[email protected]>
Co-authored-by: Alek5andr-Kotov <[email protected]>
Co-authored-by: Sergey Veselov <[email protected]>
Co-authored-by: FloatingCrowbar <[email protected]>

Author: 5 people

ydb/core/grpc_services/local_rpc/local_rpc.h

@@ -153,7 +153,12 @@ class TLocalRpcCtx : public TLocalRpcCtxImpl<TRpc, TCbWrapper, IsOperation> {
         if (key == NYdb::YDB_DATABASE_HEADER) {
             return GetDatabaseName();
         }
-        return TMaybe<TString>{};
+        auto valueIt = PeerMeta.find(key);
+        return valueIt == PeerMeta.end() ? Nothing() : TMaybe<TString>(valueIt->second);
+    }
+
+    void PutPeerMeta(const TString& key, const TString& value) {
+        PeerMeta.insert_or_assign(key, value);
     }
 
     TVector<TStringBuf> FindClientCert() const override {
@@ -278,6 +283,7 @@ class TLocalRpcCtx : public TLocalRpcCtxImpl<TRpc, TCbWrapper, IsOperation> {
     const bool InternalCall;
     TIntrusiveConstPtr<NACLib::TUserToken> InternalToken;
     const TString EmptySerializedTokenMessage_;
+    TMap<TString, TString> PeerMeta;
     google::protobuf::Arena Arena;
 };
 
@@ -318,6 +324,41 @@ NThreading::TFuture<typename TRpc::TResponse> DoLocalRpc(typename TRpc::TRequest
     return DoLocalRpc<TRpc>(std::move(proto), database, token, Nothing(), actorSystem, internalCall);
 }
 
+template<typename TRpc>
+NThreading::TFuture<typename TRpc::TResponse> DoLocalRpc(
+        typename TRpc::TRequest&& proto,
+        const TString& database,
+        const TMaybe<TString>& token,
+        const TMaybe<TString>& requestType,
+        TActorSystem* actorSystem,
+        const TMap<TString, TString>& peerMeta,
+        bool internalCall = false
+)
+{
+    auto promise = NThreading::NewPromise<typename TRpc::TResponse>();
+
+    SetRequestSyncOperationMode(proto);
+
+    using TCbWrapper = TPromiseWrapper<typename TRpc::TResponse>;
+    auto req = new TLocalRpcCtx<TRpc, TCbWrapper>(
+        std::move(proto),
+        TCbWrapper(promise),
+        database,
+        token,
+        requestType,
+        internalCall
+    );
+
+    for (const auto& [key, value] : peerMeta) {
+        req->PutPeerMeta(key, value);
+    }
+
+    auto actor = TRpc::CreateRpcActor(req);
+    actorSystem->Register(actor, TMailboxType::HTSwap, actorSystem->AppData<TAppData>()->UserPoolId);
+
+    return promise.GetFuture();
+}
+
 template<typename TRpc>
 TActorId DoLocalRpcSameMailbox(typename TRpc::TRequest&& proto, std::function<void(typename TRpc::TResponse)>&& cb,
         const TString& database, const TMaybe<TString>& token, const TMaybe<TString>& requestType,

ydb/core/grpc_services/service_ymq.h

@@ -0,0 +1,30 @@
+#pragma once
+#include <memory>
+
+namespace NActors {
+struct TActorId;
+}
+
+namespace NKikimr {
+namespace NGRpcService {
+
+class IRequestOpCtx;
+class IFacilityProvider;
+
+void DoYmqGetQueueUrlRequest(std::unique_ptr<IRequestOpCtx> p, const IFacilityProvider& f);
+void DoYmqCreateQueueRequest(std::unique_ptr<IRequestOpCtx> p, const IFacilityProvider& f);
+void DoYmqSendMessageRequest(std::unique_ptr<IRequestOpCtx> p, const IFacilityProvider& f);
+void DoYmqReceiveMessageRequest(std::unique_ptr<IRequestOpCtx> p, const IFacilityProvider& f);
+void DoYmqGetQueueAttributesRequest(std::unique_ptr<IRequestOpCtx> p, const IFacilityProvider& f);
+void DoYmqListQueuesRequest(std::unique_ptr<IRequestOpCtx> p, const IFacilityProvider& f);
+void DoYmqDeleteMessageRequest(std::unique_ptr<IRequestOpCtx> p, const IFacilityProvider& f);
+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);
+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);
+void DoYmqListDeadLetterSourceQueuesRequest(std::unique_ptr<IRequestOpCtx> p, const IFacilityProvider& f);
+}
+}

ydb/core/http_proxy/events.h

@@ -148,6 +148,40 @@ namespace NKikimr::NHttpProxy {
         };
     };
 
+    enum TEv {
+        EvYmqCloudAuthResponse
+    };
+
+    struct TEvYmqCloudAuthResponse: public TEventLocal<
+            TEvYmqCloudAuthResponse,
+            EvYmqCloudAuthResponse> {
+        struct TError {
+            TString ErrorCode;
+            ui32 HttpStatusCode;
+            TString Message;
+        };
+
+        bool IsSuccess;
+
+        TString CloudId;
+        TString FolderId;
+        TString Sid;
+
+        TMaybe<TError> Error;
+
+        TEvYmqCloudAuthResponse(const TString& cloudId, const TString& folderId, const TString& sid)
+            : IsSuccess(true)
+            , CloudId(cloudId)
+            , FolderId(folderId)
+            , Sid(sid)
+            , Error(Nothing())
+            {}
+
+        TEvYmqCloudAuthResponse(TError& error)
+            : IsSuccess(false)
+            , Error(error)
+            {}
+    };
 
     inline TActorId MakeAccessServiceID() {
         static const char x[12] = "accss_srvce";
@@ -184,6 +218,11 @@ namespace NKikimr::NHttpProxy {
         return TActorId(0, TStringBuf(x, 12));
     }
 
+    inline TActorId MakeFolderServiceID() {
+        static const char x[12] = "folder_svc";
+        return TActorId(0, TStringBuf(x, 12));
+    }
+
 #define LOG_SP_ERROR_S(actorCtxOrSystem, component, stream) LOG_ERROR_S(actorCtxOrSystem, component, LogPrefix() << " " << stream)
 #define LOG_SP_WARN_S(actorCtxOrSystem, component, stream) LOG_WARN_S(actorCtxOrSystem, component, LogPrefix() << " " << stream)
 #define LOG_SP_INFO_S(actorCtxOrSystem, component, stream) LOG_INFO_S(actorCtxOrSystem, component, LogPrefix() << " " << stream)