dummy pq emulation: cosmetics (ydb-platform#12992)

Author: yumkam

ydb/library/yql/providers/pq/gateway/dummy/yql_pq_file_topic_client.cpp

@@ -13,14 +13,17 @@ namespace NYql {
 
 class TFileTopicReadSession : public NYdb::NTopic::IReadSession {
 
-constexpr static auto FILE_POLL_PERIOD = TDuration::MilliSeconds(5);    
+constexpr static auto FILE_POLL_PERIOD = TDuration::MilliSeconds(5);
 
 public:
-    TFileTopicReadSession(TFile file, NYdb::NTopic::TPartitionSession::TPtr session, const TString& producerId = ""): 
-        File_(std::move(file)), Session_(std::move(session)), ProducerId_(producerId), 
-        FilePoller_([this] () {
-            PollFileForChanges();
-        }), Counters_()
+    TFileTopicReadSession(TFile file, NYdb::NTopic::TPartitionSession::TPtr session, const TString& producerId = "")
+        : File_(std::move(file))
+        , Session_(std::move(session))
+        , ProducerId_(producerId)
+        , FilePoller_([this] () {
+                PollFileForChanges();
+            })
+        , Counters_()
     {
         Pool_.Start(1);
     }
@@ -37,7 +40,7 @@ constexpr static auto FILE_POLL_PERIOD = TDuration::MilliSeconds(5);
         Y_UNUSED(maxByteSize);
 
         TVector<NYdb::NTopic::TReadSessionEvent::TEvent> res;
-        for (auto event = EventsQ_.Pop(block); !event.Empty() &&  res.size() < maxEventsCount.GetOrElse(std::numeric_limits<size_t>::max()); event = EventsQ_.Pop(/*block=*/ false)) {
+        for (auto event = EventsQ_.Pop(block); !event.Empty() && res.size() < maxEventsCount.GetOrElse(std::numeric_limits<size_t>::max()); event = EventsQ_.Pop(/*block=*/ false)) {
             res.push_back(std::move(*event));
         }
         return res;
@@ -60,13 +63,15 @@ constexpr static auto FILE_POLL_PERIOD = TDuration::MilliSeconds(5);
 
     bool Close(TDuration timeout = TDuration::Max()) override {
         Y_UNUSED(timeout);
+        // TODO send TSessionClosedEvent
+        // XXX (... but if we stop queues, nobody will receive it, needs rethinking)
         EventsQ_.Stop();
         Pool_.Stop();
 
         if (FilePoller_.joinable()) {
             FilePoller_.join();
         }
-        return true;
+        return true; // TODO incorrect if EventQ_ was non-empty
     }
 
     NYdb::NTopic::TReaderCounters::TPtr GetCounters() const override {
@@ -86,11 +91,11 @@ constexpr static auto FILE_POLL_PERIOD = TDuration::MilliSeconds(5);
     }
 
 private:
-    using TMessageInformation = NYdb::NTopic::TReadSessionEvent::TDataReceivedEvent::TMessageInformation; 
-    using TMessage = NYdb::NTopic::TReadSessionEvent::TDataReceivedEvent::TMessage; 
+    using TMessageInformation = NYdb::NTopic::TReadSessionEvent::TDataReceivedEvent::TMessageInformation;
+    using TMessage = NYdb::NTopic::TReadSessionEvent::TDataReceivedEvent::TMessage;
 
-    TMessageInformation MakeNextMessageInformation(size_t offset, size_t uncompressedSize, const TString& messageGroupId = "") { 
-        auto now = TInstant::Now(); 
+    TMessageInformation MakeNextMessageInformation(size_t offset, size_t uncompressedSize, const TString& messageGroupId = "") {
+        auto now = TInstant::Now();
         TMessageInformation msgInfo(
             offset,
             ProducerId_,
@@ -104,7 +109,7 @@ constexpr static auto FILE_POLL_PERIOD = TDuration::MilliSeconds(5);
         );
         return msgInfo;
     }
-    
+
     TMessage MakeNextMessage(const TString& msgBuff) {
         TMessage msg(msgBuff, nullptr, MakeNextMessageInformation(MsgOffset_, msgBuff.size()), Session_);
         return msg;
@@ -146,12 +151,14 @@ constexpr static auto FILE_POLL_PERIOD = TDuration::MilliSeconds(5);
     ui64 SeqNo_ = 0;
 };
 
-class TFileTopicWriteSession : public NYdb::NTopic::IWriteSession, private  NYdb::NTopic::TContinuationTokenIssuer {
+class TFileTopicWriteSession : public NYdb::NTopic::IWriteSession, private NYdb::NTopic::TContinuationTokenIssuer {
 public:
-    TFileTopicWriteSession(TFile file): 
-        File_(std::move(file)), FileWriter_([this] () {
-            PushToFile();
-        }), Counters_()
+    explicit TFileTopicWriteSession(TFile file)
+        : File_(std::move(file))
+        , FileWriter_([this] () {
+                PushToFile();
+            })
+        , Counters_()
     {
         Pool_.Start(1);
         EventsQ_.Push(NYdb::NTopic::TWriteSessionEvent::TReadyToAcceptEvent{IssueContinuationToken()});
@@ -170,7 +177,7 @@ class TFileTopicWriteSession : public NYdb::NTopic::IWriteSession, private  NYdb
 
     TVector<NYdb::NTopic::TWriteSessionEvent::TEvent> GetEvents(bool block, TMaybe<size_t> maxEventsCount) override {
         TVector<NYdb::NTopic::TWriteSessionEvent::TEvent> res;
-        for (auto event = EventsQ_.Pop(block); !event.Empty() &&  res.size() < maxEventsCount.GetOrElse(std::numeric_limits<size_t>::max()); event = EventsQ_.Pop(/*block=*/ false)) {
+        for (auto event = EventsQ_.Pop(block); !event.Empty() && res.size() < maxEventsCount.GetOrElse(std::numeric_limits<size_t>::max()); event = EventsQ_.Pop(/*block=*/ false)) {
             res.push_back(std::move(*event));
         }
         return res;
@@ -180,10 +187,10 @@ class TFileTopicWriteSession : public NYdb::NTopic::IWriteSession, private  NYdb
         return NThreading::MakeFuture(SeqNo_);
     }
 
-    void Write(NYdb::NTopic::TContinuationToken&&, NYdb::NTopic::TWriteMessage&& message, 
+    void Write(NYdb::NTopic::TContinuationToken&&, NYdb::NTopic::TWriteMessage&& message,
                NYdb::NTable::TTransaction* tx) override {
         Y_UNUSED(tx);
-        
+
         auto size = message.Data.size();
         EventsMsgQ_.Push(TOwningWriteMessage(std::move(message)), size);
     }
@@ -238,14 +245,16 @@ class TFileTopicWriteSession : public NYdb::NTopic::IWriteSession, private  NYdb
 
     bool Close(TDuration timeout = TDuration::Max()) override {
         Y_UNUSED(timeout);
+        // TODO send TSessionClosedEvent
+        // XXX (... but if we stop queues, nobody will receive it, needs rethinking)
         EventsQ_.Stop();
         EventsMsgQ_.Stop();
         Pool_.Stop();
 
         if (FileWriter_.joinable()) {
             FileWriter_.join();
         }
-        return true;
+        return true; // TODO incorrect if Event*Q_ was non-empty
     }
 
     NYdb::NTopic::TWriterCounters::TPtr GetCounters() override {
@@ -289,16 +298,19 @@ class TFileTopicWriteSession : public NYdb::NTopic::IWriteSession, private  NYdb
             }
         }
     }
-    
+
     TFile File_;
-    
+
     // We acquire ownership of messages immediately
     // TODO: remove extra message copying to and from queue
     struct TOwningWriteMessage {
         TString content;
         NYdb::NTopic::TWriteMessage msg;
-        
-        TOwningWriteMessage(NYdb::NTopic::TWriteMessage&& msg): content(msg.Data), msg(std::move(msg)) {
+
+        explicit TOwningWriteMessage(NYdb::NTopic::TWriteMessage&& msg)
+            : content(msg.Data)
+            , msg(std::move(msg))
+        {
             msg.Data = content;
         }
     };
@@ -368,7 +380,7 @@ NYdb::TAsyncStatus TFileTopicClient::DropTopic(const TString& path, const NYdb::
     return NThreading::MakeFuture(NYdb::TStatus(NYdb::EStatus::SUCCESS, {}));
 }
 
-NYdb::NTopic::TAsyncDescribeTopicResult TFileTopicClient::DescribeTopic(const TString& path, 
+NYdb::NTopic::TAsyncDescribeTopicResult TFileTopicClient::DescribeTopic(const TString& path,
     const NYdb::NTopic::TDescribeTopicSettings& settings) {
     Y_UNUSED(path);
     Y_UNUSED(settings);
@@ -377,7 +389,7 @@ NYdb::NTopic::TAsyncDescribeTopicResult TFileTopicClient::DescribeTopic(const TS
     return NThreading::MakeFuture(NYdb::NTopic::TDescribeTopicResult(std::move(success), {}));
 }
 
-NYdb::NTopic::TAsyncDescribeConsumerResult TFileTopicClient::DescribeConsumer(const TString& path, const TString& consumer, 
+NYdb::NTopic::TAsyncDescribeConsumerResult TFileTopicClient::DescribeConsumer(const TString& path, const TString& consumer,
     const NYdb::NTopic::TDescribeConsumerSettings& settings) {
     Y_UNUSED(path);
     Y_UNUSED(consumer);
@@ -387,7 +399,7 @@ NYdb::NTopic::TAsyncDescribeConsumerResult TFileTopicClient::DescribeConsumer(co
     return NThreading::MakeFuture(NYdb::NTopic::TDescribeConsumerResult(std::move(success), {}));
 }
 
-NYdb::NTopic::TAsyncDescribePartitionResult TFileTopicClient::DescribePartition(const TString& path, i64 partitionId, 
+NYdb::NTopic::TAsyncDescribePartitionResult TFileTopicClient::DescribePartition(const TString& path, i64 partitionId,
     const NYdb::NTopic::TDescribePartitionSettings& settings) {
     Y_UNUSED(path);
     Y_UNUSED(partitionId);