intermediate changes

ref:cde9a383711a11544ce7e107a78147fb96cc4029

1 files changed, 59 insertions, 0 deletions

diff --git a/library/cpp/messagebus/remote_server_session_semaphore.cpp b/library/cpp/messagebus/remote_server_session_semaphore.cpp
new file mode 100644
index 0000000000..6094a3586e
--- /dev/null
+++ b/library/cpp/messagebus/remote_server_session_semaphore.cpp
@@ -0,0 +1,59 @@
+#include "remote_server_session_semaphore.h"
+
+#include <util/stream/output.h>
+#include <util/system/yassert.h>
+
+using namespace NBus;
+using namespace NBus::NPrivate;
+
+TRemoteServerSessionSemaphore::TRemoteServerSessionSemaphore(
+    TAtomicBase limitCount, TAtomicBase limitSize, const char* name)
+    : Name(name)
+    , LimitCount(limitCount)
+    , LimitSize(limitSize)
+    , CurrentCount(0)
+    , CurrentSize(0)
+    , PausedByUser(0)
+    , StopSignal(0)
+{
+    Y_VERIFY(limitCount > 0, "limit must be > 0");
+    Y_UNUSED(Name);
+}
+
+TRemoteServerSessionSemaphore::~TRemoteServerSessionSemaphore() {
+    Y_VERIFY(AtomicGet(CurrentCount) == 0);
+    // TODO: fix spider and enable
+    //Y_VERIFY(AtomicGet(CurrentSize) == 0);
+}
+
+bool TRemoteServerSessionSemaphore::TryWait() {
+    if (Y_UNLIKELY(AtomicGet(StopSignal)))
+        return true;
+    if (AtomicGet(PausedByUser))
+        return false;
+    if (AtomicGet(CurrentCount) < LimitCount && (LimitSize < 0 || AtomicGet(CurrentSize) < LimitSize))
+        return true;
+    return false;
+}
+
+void TRemoteServerSessionSemaphore::IncrementMultiple(TAtomicBase count, TAtomicBase size) {
+    AtomicAdd(CurrentCount, count);
+    AtomicAdd(CurrentSize, size);
+    Updated();
+}
+
+void TRemoteServerSessionSemaphore::ReleaseMultiple(TAtomicBase count, TAtomicBase size) {
+    AtomicSub(CurrentCount, count);
+    AtomicSub(CurrentSize, size);
+    Updated();
+}
+
+void TRemoteServerSessionSemaphore::Stop() {
+    AtomicSet(StopSignal, 1);
+    Updated();
+}
+
+void TRemoteServerSessionSemaphore::PauseByUsed(bool pause) {
+    AtomicSet(PausedByUser, pause);
+    Updated();
+}