intermediate changes

ref:cde9a383711a11544ce7e107a78147fb96cc4029

9 files changed, 231 insertions, 0 deletions

diff --git a/library/cpp/messagebus/test/example/client/client.cpp b/library/cpp/messagebus/test/example/client/client.cpp
new file mode 100644
index 0000000000..89b5f2c9be
--- /dev/null
+++ b/library/cpp/messagebus/test/example/client/client.cpp
@@ -0,0 +1,81 @@
+#include <library/cpp/messagebus/test/example/common/proto.h>
+
+#include <util/random/random.h>
+
+using namespace NBus;
+using namespace NCalculator;
+
+namespace NCalculator {
+    struct TCalculatorClient: public IBusClientHandler {
+        TCalculatorProtocol Proto;
+        TBusMessageQueuePtr MessageQueue;
+        TBusClientSessionPtr ClientSession;
+
+        TCalculatorClient() {
+            MessageQueue = CreateMessageQueue();
+            TBusClientSessionConfig config;
+            config.TotalTimeout = 2 * 1000;
+            ClientSession = TBusClientSession::Create(&Proto, this, config, MessageQueue);
+        }
+
+        ~TCalculatorClient() override {
+            MessageQueue->Stop();
+        }
+
+        void OnReply(TAutoPtr<TBusMessage> request, TAutoPtr<TBusMessage> response0) override {
+            Y_VERIFY(response0->GetHeader()->Type == TResponse::MessageType, "wrong response");
+            TResponse* response = VerifyDynamicCast<TResponse*>(response0.Get());
+            if (request->GetHeader()->Type == TRequestSum::MessageType) {
+                TRequestSum* requestSum = VerifyDynamicCast<TRequestSum*>(request.Get());
+                int a = requestSum->Record.GetA();
+                int b = requestSum->Record.GetB();
+                Cerr << a << " + " << b << " = " << response->Record.GetResult() << "\n";
+            } else if (request->GetHeader()->Type == TRequestMul::MessageType) {
+                TRequestMul* requestMul = VerifyDynamicCast<TRequestMul*>(request.Get());
+                int a = requestMul->Record.GetA();
+                int b = requestMul->Record.GetB();
+                Cerr << a << " * " << b << " = " << response->Record.GetResult() << "\n";
+            } else {
+                Y_FAIL("unknown request");
+            }
+        }
+
+        void OnError(TAutoPtr<TBusMessage>, EMessageStatus status) override {
+            Cerr << "got error " << status << "\n";
+        }
+    };
+
+}
+
+int main(int, char**) {
+    TCalculatorClient client;
+
+    for (;;) {
+        TNetAddr addr(TNetAddr("127.0.0.1", TCalculatorProtocol().GetPort()));
+
+        int a = RandomNumber<unsigned>(10);
+        int b = RandomNumber<unsigned>(10);
+        EMessageStatus ok;
+        if (RandomNumber<bool>()) {
+            TAutoPtr<TRequestSum> request(new TRequestSum);
+            request->Record.SetA(a);
+            request->Record.SetB(b);
+            Cerr << "sending " << a << " + " << b << "\n";
+            ok = client.ClientSession->SendMessageAutoPtr(request, &addr);
+        } else {
+            TAutoPtr<TRequestMul> request(new TRequestMul);
+            request->Record.SetA(a);
+            request->Record.SetB(b);
+            Cerr << "sending " << a << " * " << b << "\n";
+            ok = client.ClientSession->SendMessageAutoPtr(request, &addr);
+        }
+
+        if (ok != MESSAGE_OK) {
+            Cerr << "failed to send message " << ok << "\n";
+        }
+
+        Sleep(TDuration::Seconds(1));
+    }
+
+    return 0;
+}
diff --git a/library/cpp/messagebus/test/example/client/ya.make b/library/cpp/messagebus/test/example/client/ya.make
new file mode 100644
index 0000000000..a660a01698
--- /dev/null
+++ b/library/cpp/messagebus/test/example/client/ya.make
@@ -0,0 +1,13 @@
+PROGRAM(messagebus_example_client)
+
+OWNER(g:messagebus)
+
+PEERDIR(
+    library/cpp/messagebus/test/example/common
+)
+
+SRCS(
+    client.cpp
+)
+
+END()
diff --git a/library/cpp/messagebus/test/example/common/messages.proto b/library/cpp/messagebus/test/example/common/messages.proto
new file mode 100644
index 0000000000..16b858fc77
--- /dev/null
+++ b/library/cpp/messagebus/test/example/common/messages.proto
@@ -0,0 +1,15 @@
+package NCalculator;
+
+message TRequestSumRecord {
+    required int32 A = 1;
+    required int32 B = 2;
+}
+
+message TRequestMulRecord {
+    required int32 A = 1;
+    required int32 B = 2;
+}
+
+message TResponseRecord {
+    required int32 Result = 1;
+}
diff --git a/library/cpp/messagebus/test/example/common/proto.cpp b/library/cpp/messagebus/test/example/common/proto.cpp
new file mode 100644
index 0000000000..1d18aa77ea
--- /dev/null
+++ b/library/cpp/messagebus/test/example/common/proto.cpp
@@ -0,0 +1,12 @@
+#include "proto.h"
+
+using namespace NCalculator;
+using namespace NBus;
+
+TCalculatorProtocol::TCalculatorProtocol()
+    : TBusBufferProtocol("Calculator", 34567)
+{
+    RegisterType(new TRequestSum);
+    RegisterType(new TRequestMul);
+    RegisterType(new TResponse);
+}
diff --git a/library/cpp/messagebus/test/example/common/proto.h b/library/cpp/messagebus/test/example/common/proto.h
new file mode 100644
index 0000000000..a151aac468
--- /dev/null
+++ b/library/cpp/messagebus/test/example/common/proto.h
@@ -0,0 +1,17 @@
+#pragma once
+
+#include <library/cpp/messagebus/test/example/common/messages.pb.h>
+
+#include <library/cpp/messagebus/ybus.h>
+#include <library/cpp/messagebus/protobuf/ybusbuf.h>
+
+namespace NCalculator {
+    typedef ::NBus::TBusBufferMessage<TRequestSumRecord, 1> TRequestSum;
+    typedef ::NBus::TBusBufferMessage<TRequestMulRecord, 2> TRequestMul;
+    typedef ::NBus::TBusBufferMessage<TResponseRecord, 3> TResponse;
+
+    struct TCalculatorProtocol: public ::NBus::TBusBufferProtocol {
+        TCalculatorProtocol();
+    };
+
+}
diff --git a/library/cpp/messagebus/test/example/common/ya.make b/library/cpp/messagebus/test/example/common/ya.make
new file mode 100644
index 0000000000..4da16608fc
--- /dev/null
+++ b/library/cpp/messagebus/test/example/common/ya.make
@@ -0,0 +1,15 @@
+LIBRARY(messagebus_test_example_common)
+
+OWNER(g:messagebus)
+
+PEERDIR(
+    library/cpp/messagebus
+    library/cpp/messagebus/protobuf
+)
+
+SRCS(
+    proto.cpp
+    messages.proto
+)
+
+END()
diff --git a/library/cpp/messagebus/test/example/server/server.cpp b/library/cpp/messagebus/test/example/server/server.cpp
new file mode 100644
index 0000000000..13e52d75f5
--- /dev/null
+++ b/library/cpp/messagebus/test/example/server/server.cpp
@@ -0,0 +1,58 @@
+#include <library/cpp/messagebus/test/example/common/proto.h>
+
+using namespace NBus;
+using namespace NCalculator;
+
+namespace NCalculator {
+    struct TCalculatorServer: public IBusServerHandler {
+        TCalculatorProtocol Proto;
+        TBusMessageQueuePtr MessageQueue;
+        TBusServerSessionPtr ServerSession;
+
+        TCalculatorServer() {
+            MessageQueue = CreateMessageQueue();
+            TBusServerSessionConfig config;
+            ServerSession = TBusServerSession::Create(&Proto, this, config, MessageQueue);
+        }
+
+        ~TCalculatorServer() override {
+            MessageQueue->Stop();
+        }
+
+        void OnMessage(TOnMessageContext& request) override {
+            if (request.GetMessage()->GetHeader()->Type == TRequestSum::MessageType) {
+                TRequestSum* requestSum = VerifyDynamicCast<TRequestSum*>(request.GetMessage());
+                int a = requestSum->Record.GetA();
+                int b = requestSum->Record.GetB();
+                int result = a + b;
+                Cerr << "requested " << a << " + " << b << ", sending " << result << "\n";
+                TAutoPtr<TResponse> response(new TResponse);
+                response->Record.SetResult(result);
+                request.SendReplyMove(response);
+            } else if (request.GetMessage()->GetHeader()->Type == TRequestMul::MessageType) {
+                TRequestMul* requestMul = VerifyDynamicCast<TRequestMul*>(request.GetMessage());
+                int a = requestMul->Record.GetA();
+                int b = requestMul->Record.GetB();
+                int result = a * b;
+                Cerr << "requested " << a << " * " << b << ", sending " << result << "\n";
+                TAutoPtr<TResponse> response(new TResponse);
+                response->Record.SetResult(result);
+                request.SendReplyMove(response);
+            } else {
+                Y_FAIL("unknown request");
+            }
+        }
+    };
+}
+
+int main(int, char**) {
+    TCalculatorServer server;
+
+    Cerr << "listening on port " << server.ServerSession->GetActualListenPort() << "\n";
+
+    for (;;) {
+        Sleep(TDuration::Seconds(1));
+    }
+
+    return 0;
+}
diff --git a/library/cpp/messagebus/test/example/server/ya.make b/library/cpp/messagebus/test/example/server/ya.make
new file mode 100644
index 0000000000..8cdd97cb12
--- /dev/null
+++ b/library/cpp/messagebus/test/example/server/ya.make
@@ -0,0 +1,13 @@
+PROGRAM(messagebus_example_server)
+
+OWNER(g:messagebus)
+
+PEERDIR(
+    library/cpp/messagebus/test/example/common
+)
+
+SRCS(
+    server.cpp
+)
+
+END()
diff --git a/library/cpp/messagebus/test/example/ya.make b/library/cpp/messagebus/test/example/ya.make
new file mode 100644
index 0000000000..f275351c29
--- /dev/null
+++ b/library/cpp/messagebus/test/example/ya.make
@@ -0,0 +1,7 @@
+OWNER(g:messagebus)
+
+RECURSE(
+    client
+    common
+    server
+)