diff options
| author | Devtools Arcadia <arcadia-devtools@yandex-team.ru> | 2022-02-07 18:08:42 +0300 |
|---|---|---|
| committer | Devtools Arcadia <arcadia-devtools@mous.vla.yp-c.yandex.net> | 2022-02-07 18:08:42 +0300 |
| commit | 1110808a9d39d4b808aef724c861a2e1a38d2a69 (patch) | |
| tree | e26c9fed0de5d9873cce7e00bc214573dc2195b7 /library/cpp/messagebus/test/ut/module_server_ut.cpp | |
| download | ydb-1110808a9d39d4b808aef724c861a2e1a38d2a69.tar.gz | |
intermediate changes
ref:cde9a383711a11544ce7e107a78147fb96cc4029
Diffstat (limited to 'library/cpp/messagebus/test/ut/module_server_ut.cpp')
| -rw-r--r-- | library/cpp/messagebus/test/ut/module_server_ut.cpp | 119 |
1 files changed, 119 insertions, 0 deletions
diff --git a/library/cpp/messagebus/test/ut/module_server_ut.cpp b/library/cpp/messagebus/test/ut/module_server_ut.cpp new file mode 100644 index 0000000000..88fe1dd9b6 --- /dev/null +++ b/library/cpp/messagebus/test/ut/module_server_ut.cpp @@ -0,0 +1,119 @@ +#include <library/cpp/testing/unittest/registar.h> + +#include "count_down_latch.h" +#include "moduletest.h" + +#include <library/cpp/messagebus/test/helper/example.h> +#include <library/cpp/messagebus/test/helper/example_module.h> +#include <library/cpp/messagebus/test/helper/object_count_check.h> +#include <library/cpp/messagebus/test/helper/wait_for.h> + +#include <library/cpp/messagebus/oldmodule/module.h> + +#include <util/generic/cast.h> + +using namespace NBus; +using namespace NBus::NTest; + +Y_UNIT_TEST_SUITE(ModuleServerTests) { + Y_UNIT_TEST(TestModule) { + TObjectCountCheck objectCountCheck; + + /// create or get instance of message queue, need one per application + TBusMessageQueuePtr bus(CreateMessageQueue()); + THostInfoHandler hostHandler(bus.Get()); + TDupDetectModule module(hostHandler.GetActualListenAddr()); + bool success; + success = module.Init(bus.Get()); + UNIT_ASSERT_C(success, "failed to initialize dupdetect module"); + + success = module.StartInput(); + UNIT_ASSERT_C(success, "failed to start dupdetect module"); + + TDupDetectHandler dupHandler(module.ListenAddr, bus.Get()); + dupHandler.Work(); + + UNIT_WAIT_FOR(dupHandler.NumMessages == dupHandler.NumReplies); + + module.Shutdown(); + dupHandler.DupDetect->Shutdown(); + } + + struct TParallelOnMessageModule: public TExampleServerModule { + TCountDownLatch WaitTwoRequestsLatch; + + TParallelOnMessageModule() + : WaitTwoRequestsLatch(2) + { + } + + TJobHandler Start(TBusJob* job, TBusMessage* mess) override { + WaitTwoRequestsLatch.CountDown(); + Y_VERIFY(WaitTwoRequestsLatch.Await(TDuration::Seconds(5)), "oops"); + + VerifyDynamicCast<TExampleRequest*>(mess); + + job->SendReply(new TExampleResponse(&Proto.ResponseCount)); + return nullptr; + } + }; + + Y_UNIT_TEST(TestOnMessageHandlerCalledInParallel) { + TObjectCountCheck objectCountCheck; + + TBusQueueConfig config; + config.NumWorkers = 5; + + TParallelOnMessageModule module; + module.StartModule(); + + TExampleClient client; + + client.SendMessagesWaitReplies(2, module.ServerAddr); + + module.Shutdown(); + } + + struct TDelayReplyServer: public TExampleServerModule { + TSystemEvent MessageReceivedEvent; + TSystemEvent ClientDiedEvent; + + TJobHandler Start(TBusJob* job, TBusMessage* mess) override { + Y_UNUSED(mess); + + MessageReceivedEvent.Signal(); + + Y_VERIFY(ClientDiedEvent.WaitT(TDuration::Seconds(5)), "oops"); + + job->SendReply(new TExampleResponse(&Proto.ResponseCount)); + return nullptr; + } + }; + + Y_UNIT_TEST(TestReplyCalledAfterClientDisconnected) { + TObjectCountCheck objectCountCheck; + + TBusQueueConfig config; + config.NumWorkers = 5; + + TDelayReplyServer server; + server.StartModule(); + + THolder<TExampleClient> client(new TExampleClient); + + client->SendMessages(1, server.ServerAddr); + + UNIT_ASSERT(server.MessageReceivedEvent.WaitT(TDuration::Seconds(5))); + + UNIT_ASSERT_VALUES_EQUAL(1, server.GetModuleSessionInFlight()); + + client.Destroy(); + + server.ClientDiedEvent.Signal(); + + // wait until all server message are delivered + UNIT_WAIT_FOR(0 == server.GetModuleSessionInFlight()); + + server.Shutdown(); + } +} |