diff --git a/Import/GacUI.cpp b/Import/GacUI.cpp index 3f32eef4..cf943d7d 100644 --- a/Import/GacUI.cpp +++ b/Import/GacUI.cpp @@ -39010,10 +39010,9 @@ GuiRemoteProtocolJsonChannelRenderer_Async #undef ERROR_MESSAGE_PREFIX } - void GuiRemoteProtocolJsonChannelRenderer_Async::SendToClient(vint senderClientId, vint receiverClientId, const JsonPackage& package) + void GuiRemoteProtocolJsonChannelRenderer_Async::SendToClient(vint receiverClientId, const JsonPackage& package) { QueuedPackage queuedPackage; - queuedPackage.senderClientId = senderClientId; queuedPackage.receiverClientId = receiverClientId; queuedPackage.package = package; @@ -39023,11 +39022,18 @@ GuiRemoteProtocolJsonChannelRenderer_Async } } - void GuiRemoteProtocolJsonChannelRenderer_Async::BroadcastFromClient(vint senderClientId, const JsonPackage& package) + void GuiRemoteProtocolJsonChannelRenderer_Async::BroadcastFromClient(const JsonPackage& package) + { + List blockedReceivers; + BroadcastFromClient(package, blockedReceivers); + } + + void GuiRemoteProtocolJsonChannelRenderer_Async::BroadcastFromClient(const JsonPackage& package, const List& blockedReceivers) { QueuedPackage queuedPackage; - queuedPackage.senderClientId = senderClientId; queuedPackage.package = package; + queuedPackage.blockedReceivers = Ptr(new List); + CopyFrom(*queuedPackage.blockedReceivers.Obj(), blockedReceivers); SPIN_LOCK(lockPendingPackages) { @@ -39086,11 +39092,18 @@ GuiRemoteProtocolJsonChannelRenderer_Async { if (queuedPackage.receiverClientId) { - channel->SendToClient(queuedPackage.senderClientId, queuedPackage.receiverClientId.Value(), queuedPackage.package); + channel->SendToClient(queuedPackage.receiverClientId.Value(), queuedPackage.package); } else { - channel->BroadcastFromClient(queuedPackage.senderClientId, queuedPackage.package); + if (queuedPackage.blockedReceivers) + { + channel->BroadcastFromClient(queuedPackage.package, *queuedPackage.blockedReceivers.Obj()); + } + else + { + channel->BroadcastFromClient(queuedPackage.package); + } } } channel->BatchWrite(disconnected); @@ -39279,14 +39292,19 @@ GuiRemoteProtocolAsyncJsonChannelRenderer } } - void GuiRemoteProtocolAsyncJsonChannelRenderer::SendToClient(vint senderClientId, vint receiverClientId, const JsonPackage& package) + void GuiRemoteProtocolAsyncJsonChannelRenderer::SendToClient(vint receiverClientId, const JsonPackage& package) { - channel->SendToClient(senderClientId, receiverClientId, package); + channel->SendToClient(receiverClientId, package); } - void GuiRemoteProtocolAsyncJsonChannelRenderer::BroadcastFromClient(vint senderClientId, const JsonPackage& package) + void GuiRemoteProtocolAsyncJsonChannelRenderer::BroadcastFromClient(const JsonPackage& package) { - channel->BroadcastFromClient(senderClientId, package); + channel->BroadcastFromClient(package); + } + + void GuiRemoteProtocolAsyncJsonChannelRenderer::BroadcastFromClient(const JsonPackage& package, const List& blockedReceivers) + { + channel->BroadcastFromClient(package, blockedReceivers); } void GuiRemoteProtocolAsyncJsonChannelRenderer::BatchWrite(bool& disconnected) @@ -39553,9 +39571,9 @@ GuiRemoteProtocolCoreChannel for (auto&& pendingPackage : packages) { - channel->SendToClient(client->GetClientId(), receiverClientId, pendingPackage); + channel->SendToClient(receiverClientId, pendingPackage); } - channel->SendToClient(client->GetClientId(), receiverClientId, package); + channel->SendToClient(receiverClientId, package); } } @@ -39733,7 +39751,7 @@ GuiRemoteProtocolCoreChannel for (auto&& package : packages) { - channel->SendToClient(client->GetClientId(), receiverClientId, package); + channel->SendToClient(receiverClientId, package); } channel->BatchWrite(disconnected); } @@ -39836,7 +39854,7 @@ GuiRemoteProtocolRendererChannel void GuiRemoteProtocolRendererChannel::Write(Ptr package) { - channel->SendToClient(client->GetClientId(), GacUIRemoteProtocolCoreClientId, package); + channel->SendToClient(GacUIRemoteProtocolCoreClientId, package); if (!receiving) { diff --git a/Import/GacUI.h b/Import/GacUI.h index f80ee569..672e94c8 100644 --- a/Import/GacUI.h +++ b/Import/GacUI.h @@ -22822,6 +22822,7 @@ namespace vl::presentation::remoteprotocol::channeling using IJsonChannel = inter_process::IChannel; using IJsonChannelClient = inter_process::IChannelClient; using IJsonChannelServer = inter_process::IChannelServer; + using IJsonLocalChannelServer = inter_process::INetworkProtocolLocalChannelServer; /*********************************************************************** ChannelPackageSemantic @@ -22847,7 +22848,39 @@ ChannelPackageSemantic extern void JsonChannelUnpack(Ptr package, ChannelPackageInfo& info, Ptr& arguments); extern void JsonChannelUnpack(Ptr package, ChannelPackageInfo& info, Ptr& arguments); - using GuiRemoteProtocolChannelServer = inter_process::NetworkProtocolChannelServer; + class GuiRemoteProtocolLocalChannelServerBase + : public Object + , public virtual inter_process::INetworkProtocolServer + { + private: + bool stopped = true; + + public: + inter_process::WaitForClientResult OnClientConnected(inter_process::INetworkProtocolConnection* connection) override + { + return inter_process::WaitForClientResult::Reject; + } + + void Start() override + { + stopped = false; + } + + void Stop() override + { + stopped = true; + } + + bool IsStopped() override + { + return stopped; + } + }; + + template + using GuiRemoteProtocolNetworkChannelServer = inter_process::NetworkProtocolChannelServer; + + using GuiRemoteProtocolChannelServer = GuiRemoteProtocolNetworkChannelServer; class GuiRemoteProtocolChannelClient : public inter_process::NetworkProtocolChannelClient @@ -23052,8 +23085,8 @@ GuiRemoteProtocolJsonChannelRenderer_Async protected: struct QueuedPackage { - vint senderClientId = -1; Nullable receiverClientId; + Ptr> blockedReceivers; JsonPackage package; }; @@ -23104,8 +23137,9 @@ GuiRemoteProtocolJsonChannelRenderer_Async const WString& GetChannelName() override; IJsonChannelReader* GetReader() override; void Initialize(IJsonChannelReader* _reader) override; - void SendToClient(vint senderClientId, vint receiverClientId, const JsonPackage& package) override; - void BroadcastFromClient(vint senderClientId, const JsonPackage& package) override; + void SendToClient(vint receiverClientId, const JsonPackage& package) override; + void BroadcastFromClient(const JsonPackage& package) override; + void BroadcastFromClient(const JsonPackage& package, const collections::List& blockedReceivers) override; void BatchWrite(bool& disconnected) override; IGuiRemoteEventProcessor* GetRemoteEventProcessor(); @@ -23185,8 +23219,9 @@ GuiRemoteProtocolAsyncJsonChannelRenderer const WString& GetChannelName() override; IJsonChannelReader* GetReader() override; void Initialize(IJsonChannelReader* _reader) override; - void SendToClient(vint senderClientId, vint receiverClientId, const JsonPackage& package) override; - void BroadcastFromClient(vint senderClientId, const JsonPackage& package) override; + void SendToClient(vint receiverClientId, const JsonPackage& package) override; + void BroadcastFromClient(const JsonPackage& package) override; + void BroadcastFromClient(const JsonPackage& package, const collections::List& blockedReceivers) override; void BatchWrite(bool& disconnected) override; void SetInvokeInMainThread(IGuiRemoteProtocolAsyncRendererInvoker* _invokeInMainThread); diff --git a/Import/Vlpp.Linux.cpp b/Import/Vlpp.Linux.cpp index 8ab7effc..3c1ba508 100644 --- a/Import/Vlpp.Linux.cpp +++ b/Import/Vlpp.Linux.cpp @@ -34,11 +34,27 @@ Console std::wcout << s << std::flush; } - WString Console::Read() + Nullable Console::TryRead() { std::wstring s; - std::getline(std::wcin, s, L'\n'); - return s.c_str(); + if (!std::getline(std::wcin, s, L'\n')) + { + if (s.empty()) + { + return {}; + } + } + if (!s.empty() && s[s.size() - 1] == L'\r') + { + s.pop_back(); + } + return WString::CopyFrom(s.c_str(), (vint)s.size()); + } + + WString Console::Read() + { + auto result = TryRead(); + return result ? result.Value() : WString::Empty; } void Console::SetColor(bool red, bool green, bool blue, bool light) diff --git a/Import/Vlpp.Windows.cpp b/Import/Vlpp.Windows.cpp index 5e979f0d..44c69d69 100644 --- a/Import/Vlpp.Windows.cpp +++ b/Import/Vlpp.Windows.cpp @@ -48,29 +48,93 @@ Console } } + Nullable Console::TryRead() + { + auto inHandle = GetStdHandle(STD_INPUT_HANDLE); + if (inHandle == INVALID_HANDLE_VALUE || inHandle == NULL) + { + return {}; + } + + WString result; + DWORD fileMode = 0; + if ((GetFileType(inHandle) & FILE_TYPE_CHAR) && GetConsoleMode(inHandle, &fileMode)) + { + for (;;) + { + wchar_t buffer = 0; + DWORD count = 0; + if (!ReadConsole(inHandle, &buffer, 1, &count, 0) || count == 0) + { + return result.Length() == 0 ? Nullable() : Nullable(result); + } + + if (buffer == L'\r') + { + if (!ReadConsole(inHandle, &buffer, 1, &count, 0) || count == 0) + { + return result; + } + break; + } + else if (buffer == L'\n') + { + break; + } + else + { + result = result + WString::FromChar(buffer); + } + } + return result; + } + else + { + AString buffer; + for (;;) + { + char c = 0; + DWORD count = 0; + if (!ReadFile(inHandle, &c, 1, &count, 0) || count == 0) + { + if (buffer.Length() == 0) + { + return {}; + } + break; + } + + if (c == '\n') + { + break; + } + else + { + buffer = buffer + AString::FromChar(c); + } + } + + if (buffer.Length() > 0 && buffer[buffer.Length() - 1] == '\r') + { + buffer = buffer.Left(buffer.Length() - 1); + } + int codePage = GetConsoleCP(); + if (codePage == 0) + { + codePage = CP_THREAD_ACP; + } + auto charCount = MultiByteToWideChar(codePage, 0, buffer.Buffer(), (int)buffer.Length(), nullptr, 0); + auto wbuffer = new wchar_t[charCount + 1]; + MultiByteToWideChar(codePage, 0, buffer.Buffer(), (int)buffer.Length(), wbuffer, charCount); + wbuffer[charCount] = 0; + return WString::TakeOver(wbuffer, charCount); + } + } + WString Console::Read() { - WString result; - DWORD count; - for (;;) - { - wchar_t buffer; - ReadConsole(GetStdHandle(STD_INPUT_HANDLE), &buffer, 1, &count, 0); - if (buffer == L'\r') - { - ReadConsole(GetStdHandle(STD_INPUT_HANDLE), &buffer, 1, &count, 0); - break; - } - else if (buffer == L'\n') - { - break; - } - else - { - result = result + WString::FromChar(buffer); - } - } - return result; + auto result = TryRead(); + return result ? result.Value() : WString::Empty; } void Console::SetColor(bool red, bool green, bool blue, bool light) diff --git a/Import/Vlpp.h b/Import/Vlpp.h index ec764c72..8eef7eb8 100644 --- a/Import/Vlpp.h +++ b/Import/Vlpp.h @@ -8733,8 +8733,12 @@ namespace vl /// Content to write. static void WriteLine(const WString& string); - /// Read a string from the command-line window. - /// The whole line read from the command-line window. + /// Try to read a string from the command-line window or redirected input. + /// The whole line read from the command-line window or redirected input. Returns null if no line is available. + static Nullable TryRead(); + + /// Read a string from the command-line window or redirected input. + /// The whole line read from the command-line window or redirected input. Returns an empty string if no line is available. static WString Read(); static void SetColor(bool red, bool green, bool blue, bool light); @@ -8745,6 +8749,7 @@ namespace vl #endif + /*********************************************************************** .\EXCEPTION.H ***********************************************************************/ diff --git a/Import/VlppOS.Windows.cpp b/Import/VlppOS.Windows.cpp index cd47c421..217c3445 100644 --- a/Import/VlppOS.Windows.cpp +++ b/Import/VlppOS.Windows.cpp @@ -3807,7 +3807,15 @@ RESTART_LOOP: } DWORD error = GetLastError(); - if (error == ERROR_BROKEN_PIPE || error == ERROR_INVALID_HANDLE) + if (error == ERROR_BROKEN_PIPE || error == ERROR_NO_DATA) + { + if (!stopped) + { + OnDisconnected(); + } + return; + } + if (error == ERROR_INVALID_HANDLE) { if (!stopped) { @@ -3866,7 +3874,16 @@ RESTART_LOOP: else { DWORD error = GetLastError(); - if (error == ERROR_OPERATION_ABORTED || error == ERROR_INVALID_HANDLE || error == ERROR_BROKEN_PIPE || error == ERROR_NO_DATA) + if (error == ERROR_BROKEN_PIPE || error == ERROR_NO_DATA) + { + if (!self->stopped) + { + self->OnDisconnected(); + } + finalize(); + return; + } + if (error == ERROR_OPERATION_ABORTED || error == ERROR_INVALID_HANDLE) { if (!self->stopped) { @@ -4061,6 +4078,14 @@ void NamedPipeConnection::InstallCallback(INetworkProtocolCallback* _callback) void NamedPipeConnection::Stop() { stopped = 1; + SPIN_LOCK(lockWrite) + { + if (hPipe != INVALID_HANDLE_VALUE) + { + CancelIoEx(hPipe, NULL); + } + } + ReadWaitContext* context = readWaitContext.exchange(nullptr); if (context) { @@ -4081,7 +4106,6 @@ void NamedPipeConnection::Stop() { if (hPipe != INVALID_HANDLE_VALUE) { - CancelIoEx(hPipe, NULL); CloseHandle(hPipe); hPipe = INVALID_HANDLE_VALUE; } diff --git a/Import/VlppOS.Windows.h b/Import/VlppOS.Windows.h index dbd81602..5ae73629 100644 --- a/Import/VlppOS.Windows.h +++ b/Import/VlppOS.Windows.h @@ -60,6 +60,7 @@ namespace vl::inter_process #endif + /*********************************************************************** .\HTTPCLIENTAPI.WINDOWS.H ***********************************************************************/ diff --git a/Import/VlppOS.cpp b/Import/VlppOS.cpp index 454fa131..d2395347 100644 --- a/Import/VlppOS.cpp +++ b/Import/VlppOS.cpp @@ -1297,6 +1297,71 @@ ThreadLocalStorage delete temp; } } + +/*********************************************************************** +TaskQueue +***********************************************************************/ + + TaskQueue::TaskQueue() + { + CHECK_ERROR(semaphoreTasks.Create(0, 65536), L"vl::TaskQueue::TaskQueue()#Failed to create the task semaphore."); + } + + TaskQueue::~TaskQueue() + { + } + + void TaskQueue::QueueTask(Func task) + { + SPIN_LOCK(lockTasks) + { + tasks.Add(task); + } + semaphoreTasks.Release(); + } + + void TaskQueue::QueueExitTask() + { + SPIN_LOCK(lockTasks) + { + exitTaskQueued = true; + } + semaphoreTasks.Release(); + } + + void TaskQueue::RunTaskQueue() + { + while (true) + { + Func task; + bool hasTask = false; + bool shouldExit = false; + SPIN_LOCK(lockTasks) + { + if (tasks.Count() > 0) + { + task = tasks[0]; + tasks.RemoveAt(0); + hasTask = true; + } + else + { + shouldExit = exitTaskQueued; + } + } + + if (shouldExit) + { + break; + } + if (!hasTask) + { + semaphoreTasks.Wait(); + continue; + } + task(); + } + } } @@ -2833,13 +2898,104 @@ Unicode General (extern templates) /*********************************************************************** -.\INTERPROCESS\TEXTNETWORKPROTOCOL.CPP +.\INTERPROCESS\CHANNELIMPLS\CHANNELPACKAGE.CPP ***********************************************************************/ namespace vl::inter_process { + NetworkPackage NetworkPackage::Create(Nullable _clientId, const WString& _channelName, const WString& _messageBody) + { + NetworkPackage package; + package.clientId = std::move(_clientId); + package.channelName = _channelName; + package.messageBody = _messageBody; + return package; + } + + NetworkPackage NetworkPackage::Create(Nullable _clientId, const ClientIdList& _extraClientIds, const WString& _channelName, const WString& _messageBody) + { + auto package = Create(std::move(_clientId), _channelName, _messageBody); + if (_extraClientIds.Count() > 0) + { + ClientIdList extraClientIds; + for (auto clientId : _extraClientIds) + { + extraClientIds.Add(clientId); + } + package.extraClientIds = std::move(extraClientIds); + } + return package; + } + + WString NetworkPackage::ToString(const NetworkPackage& package) + { + auto clientIds = package.clientId ? itow(package.clientId.Value()) : WString::Empty; + if (package.extraClientIds) + { + for (auto clientId : package.extraClientIds.Value()) + { + clientIds += L","; + clientIds += itow(clientId); + } + } + + return clientIds + + L";" + package.channelName + + L";" + package.messageBody + ; + } + + void NetworkPackage::Parse(const WString& str, NetworkPackage& package) + { +#define ERROR_MESSAGE_PREFIX L"vl::inter_process::NetworkPackage::Parse(const WString&, NetworkPackage&)#" + const wchar_t* reading = str.Buffer(); + + const wchar_t* afterClientId = wcschr(reading, L';'); + CHECK_ERROR(afterClientId != nullptr, ERROR_MESSAGE_PREFIX L"Invalid package format."); + package.clientId.Reset(); + package.extraClientIds.Reset(); + const wchar_t* firstExtraClientId = wcschr(reading, L','); + if (firstExtraClientId && firstExtraClientId > afterClientId) + { + firstExtraClientId = nullptr; + } + auto clientIdLength = firstExtraClientId ? (vint)(firstExtraClientId - reading) : (vint)(afterClientId - reading); + if (clientIdLength > 0) + { + package.clientId = wtoi(str.Left(clientIdLength)); + } + + if (firstExtraClientId) + { + ClientIdList extraClientIds; + auto readingExtraClientId = firstExtraClientId + 1; + while (readingExtraClientId < afterClientId) + { + auto delimiter = wcschr(readingExtraClientId, L','); + if (delimiter && delimiter > afterClientId) + { + delimiter = nullptr; + } + auto endingExtraClientId = delimiter ? delimiter : afterClientId; + CHECK_ERROR(endingExtraClientId > readingExtraClientId, ERROR_MESSAGE_PREFIX L"Invalid extra client id format."); + extraClientIds.Add(wtoi(WString::CopyFrom(readingExtraClientId, (vint)(endingExtraClientId - readingExtraClientId)))); + readingExtraClientId = endingExtraClientId + (delimiter ? 1 : 0); + } + if (extraClientIds.Count() > 0) + { + package.extraClientIds = std::move(extraClientIds); + } + } + + const wchar_t* afterChannelName = wcschr(afterClientId + 1, L';'); + CHECK_ERROR(afterChannelName != nullptr, ERROR_MESSAGE_PREFIX L"Invalid package format."); + package.channelName = str.Sub((vint)(afterClientId - reading + 1), (vint)(afterChannelName - afterClientId - 1)); + package.messageBody = str.Right(str.Length() - (vint)(afterChannelName - reading + 1)); +#undef ERROR_MESSAGE_PREFIX + } } + /*********************************************************************** .\STREAM\ACCESSOR.CPP ***********************************************************************/ diff --git a/Import/VlppOS.h b/Import/VlppOS.h index 5cf13409..4b8c8c2d 100644 --- a/Import/VlppOS.h +++ b/Import/VlppOS.h @@ -1108,6 +1108,32 @@ RepeatingTaskExecutor } } }; + +/*********************************************************************** +TaskQueue +***********************************************************************/ + + /// A single-threaded blocking task queue. + class TaskQueue : public Object + { + private: + SpinLock lockTasks; + Semaphore semaphoreTasks; + collections::List> tasks; + bool exitTaskQueued = false; + + public: + TaskQueue(); + ~TaskQueue(); + + /// Queue a task to be executed by . + /// The task to execute. + void QueueTask(Func task); + /// Request to return after all queued tasks are executed. + void QueueExitTask(); + /// Run queued tasks in the current thread until is called. + void RunTaskQueue(); + }; } #endif @@ -1189,18 +1215,24 @@ IGuiRemoteProtocolChannel /// Queue a message to send to a client using the same channel. /// If the remote client doesn't have this channel, the message will be discarded. /// - /// The sender client id. /// The receiver client id. /// The message to send. - virtual void SendToClient(vint senderClientId, vint receiverClientId, const TPackage& package) = 0; + virtual void SendToClient(vint receiverClientId, const TPackage& package) = 0; /// /// Queue a message to broadcast to all other clients using the same channel. /// If the remote client doesn't have this channel, the message will be discarded. /// - /// The sender client id. /// The message to broadcast. - virtual void BroadcastFromClient(vint senderClientId, const TPackage& package) = 0; + virtual void BroadcastFromClient(const TPackage& package) = 0; + + /// + /// Queue a message to broadcast to all other clients using the same channel. + /// If the remote client doesn't have this channel, the message will be discarded. + /// + /// The message to broadcast. + /// The receiver client ids blocked from this broadcast. + virtual void BroadcastFromClient(const TPackage& package, const collections::List& blockedReceivers) = 0; /// /// Send all queued messages. @@ -1332,8 +1364,9 @@ IChannelServer /// /// The client id. /// The available channels. + /// The local client. It is null for network clients. /// Returns "Reject" to disconnect the client immediatelly. - virtual WaitForClientResult OnClientConnected(vint clientId, const IChannelClient::ChannelNameList& availableChannels) = 0; + virtual WaitForClientResult OnClientConnected(vint clientId, const IChannelClient::ChannelNameList& availableChannels, IChannelClient* localClient) = 0; /// /// Start the server. @@ -1503,18 +1536,24 @@ struct { } - void SendToClient(vint senderClientId, vint receiverClientId, const typename TSerialization::SourceType& package) override + void SendToClient(vint receiverClientId, const typename TSerialization::SourceType& package) override { typename TSerialization::DestType serialized; TSerialization::Serialize(context, package, serialized); - this->channel->SendToClient(senderClientId, receiverClientId, serialized); + this->channel->SendToClient(receiverClientId, serialized); } - void BroadcastFromClient(vint senderClientId, const typename TSerialization::SourceType& package) override + void BroadcastFromClient(const typename TSerialization::SourceType& package) override + { + collections::List blockedReceivers; + BroadcastFromClient(package, blockedReceivers); + } + + void BroadcastFromClient(const typename TSerialization::SourceType& package, const collections::List& blockedReceivers) override { typename TSerialization::DestType serialized; TSerialization::Serialize(context, package, serialized); - this->channel->BroadcastFromClient(senderClientId, serialized); + this->channel->BroadcastFromClient(serialized, blockedReceivers); } }; @@ -1548,70 +1587,21 @@ String Transformation /*********************************************************************** -.\INTERPROCESS\TEXTNETWORKPROTOCOL.H +.\INTERPROCESS\NETWORKPROTOCOL.H ***********************************************************************/ /*********************************************************************** Vczh Library++ 3.0 Developer: Zihan Chen(vczh) Interfaces: - INetworkProtocol - ***********************************************************************/ -#ifndef VCZH_INTERPROCESS_TEXTNETWORKPROTOCOL -#define VCZH_INTERPROCESS_TEXTNETWORKPROTOCOL +#ifndef VCZH_INTERPROCESS_NETWORKPROTOCOL +#define VCZH_INTERPROCESS_NETWORKPROTOCOL namespace vl::inter_process { - struct NetworkPackage - { - Nullable clientId; - WString channelName; - WString messageBody; - - static inline NetworkPackage Create(Nullable _clientId, const WString& _channelName, const WString& _messageBody) - { - NetworkPackage package; - package.clientId = std::move(_clientId); - package.channelName = _channelName; - package.messageBody = _messageBody; - return package; - } - - static inline WString ToString(const NetworkPackage& package) - { - return (package.clientId ? itow(package.clientId.Value()) : WString::Empty) - + L";" + package.channelName - + L";" + package.messageBody - ; - } - - static inline void Parse(const WString& str, NetworkPackage& package) - { -#define ERROR_MESSAGE_PREFIX L"vl::inter_process::NetworkPackage::Parse(const WString&, NetworkPackage&)#" - const wchar_t* reading = str.Buffer(); - - const wchar_t* afterClientId = wcschr(reading, L';'); - CHECK_ERROR(afterClientId != nullptr, ERROR_MESSAGE_PREFIX L"Invalid package format."); - if (afterClientId == reading) - { - package.clientId.Reset(); - } - else - { - package.clientId = wtoi(str.Left((vint)(afterClientId - reading))); - } - - const wchar_t* afterChannelName = wcschr(afterClientId + 1, L';'); - CHECK_ERROR(afterChannelName != nullptr, ERROR_MESSAGE_PREFIX L"Invalid package format."); - package.channelName = str.Sub((vint)(afterClientId - reading + 1), (vint)(afterChannelName - afterClientId - 1)); - package.messageBody = str.Right(str.Length() - (vint)(afterChannelName - reading + 1)); -#undef ERROR_MESSAGE_PREFIX - } - }; - /*********************************************************************** INetworkProtocolServer ***********************************************************************/ @@ -1635,25 +1625,25 @@ INetworkProtocolServer /// Called when an error message is received from the other side of the connection. /// /// The error message. - virtual void OnReadError(const WString& error) = 0; + virtual void OnReadError(const WString&) {} /// /// Called when a local transport error occurs. /// /// The error message. /// Indicates whether the connection should be disconnected after this callback. - virtual void OnLocalError(const WString& error, bool fatal) = 0; + virtual void OnLocalError(const WString&, bool) {} /// /// Called when the connection becomes available. /// This function might not be called if is called after the connection is already established. /// - virtual void OnConnected() = 0; + virtual void OnConnected() {} /// /// Called when the connection is lost. /// - virtual void OnDisconnected() = 0; + virtual void OnDisconnected() {} /// /// Called when the callback is installed to a connection. @@ -1756,34 +1746,27 @@ INetworkProtocolServer virtual bool IsStopped() = 0; }; +} + +#endif + + /*********************************************************************** -Hooking IChannelServer/IChannelClient to INetworkProtocolServer/INetworkProtocolClient +.\INTERPROCESS\CHANNELIMPLS\CHANNELIMPL.H +***********************************************************************/ +/*********************************************************************** +Vczh Library++ 3.0 +Developer: Zihan Chen(vczh) -The serialization contract is the same to the one described in ChannelSerialization.h -SourceType will be List -DestType will be WString - -NetworkPackage will be used as text message parsing and formatting for INetworkProtocolConnection. -BatchWrite belongs to IChannel, meaning each channel sends its own batch messages in one NetworkPackage. -channelName will be either a system channel or a user defined channel. -messageBody represents a list of TPackage. -When sending from client to server, clientId means the target client. - Empty means broadcasting. -When sending from server to client, clientId means the source client. - Channel messages delivered by the server always carry a source client id. - -When a client establishes a connection to the server, channel names will be sent to the server: - clientId will be empty, it does not mean broadcasting. - channelName will be empty. - messageBody will be all available channel names joined by "!", as "!" cannot be part of the channel name anyway. -After the server receives the first message from a client, an client id will be sent to the client: - clientId is the assigned client id, starting from 1. - channelName will be empty. - messageBody will be empty. - -Later +Interfaces: ***********************************************************************/ +#ifndef VCZH_INTERPROCESS_CHANNELIMPLS_CHANNELIMPL +#define VCZH_INTERPROCESS_CHANNELIMPLS_CHANNELIMPL + + +namespace vl::inter_process +{ /*********************************************************************** NetworkProtocolChannel ***********************************************************************/ @@ -1795,6 +1778,7 @@ NetworkProtocolChannel static_assert(std::is_same_v); protected: using PackageList = typename TSerialization::SourceType; + using ClientIdList = collections::List; using ChannelMap = collections::Dictionary*>; using ChannelNameList = typename ChannelMap::KeyContainer; @@ -1806,8 +1790,8 @@ NetworkProtocolChannel struct QueuedPackage { - vint senderClientId = -1; Nullable receiverClientId; + ClientIdList blockedReceivers; TPackage package; }; @@ -1822,8 +1806,8 @@ NetworkProtocolChannel SpinLock lockQueuedPackages; collections::List queuedPackages; - virtual void ValidatePackage(vint senderClientId, Nullable receiverClientId) = 0; - virtual bool WriteBatch(vint senderClientId, Nullable receiverClientId, const PackageList& batch) = 0; + virtual void ValidatePackage(Nullable receiverClientId, const ClientIdList& blockedReceivers) = 0; + virtual bool WriteBatch(Nullable receiverClientId, const ClientIdList& blockedReceivers, const PackageList& batch) = 0; public: NetworkProtocolChannel(const WString& _channelName) @@ -1900,16 +1884,23 @@ NetworkProtocolChannel } } - void SendToClient(vint senderClientId, vint receiverClientId, const TPackage& package) override + void SendToClient(vint receiverClientId, const TPackage& package) override { - ValidatePackage(senderClientId, receiverClientId); - QueuePackage(senderClientId, receiverClientId, package); + ClientIdList blockedReceivers; + ValidatePackage(receiverClientId, blockedReceivers); + QueuePackage(receiverClientId, blockedReceivers, package); } - void BroadcastFromClient(vint senderClientId, const TPackage& package) override + void BroadcastFromClient(const TPackage& package) override { - ValidatePackage(senderClientId, {}); - QueuePackage(senderClientId, {}, package); + ClientIdList blockedReceivers; + BroadcastFromClient(package, blockedReceivers); + } + + void BroadcastFromClient(const TPackage& package, const ClientIdList& blockedReceivers) override + { + ValidatePackage({}, blockedReceivers); + QueuePackage({}, blockedReceivers, package); } void BatchWrite(bool& disconnected) override @@ -1923,12 +1914,14 @@ NetworkProtocolChannel while (packages.Count() > 0) { PackageList batch; - auto senderClientId = packages[0].senderClientId; auto receiverClientId = packages[0].receiverClientId; + ClientIdList blockedReceivers = std::move(packages[0].blockedReceivers); + batch.Add(packages[0].package); + packages.RemoveAt(0); for (vint i = 0; i < packages.Count();) { auto&& package = packages[i]; - if (package.senderClientId == senderClientId && package.receiverClientId == receiverClientId) + if (package.receiverClientId == receiverClientId && ClientIdsEqual(package.blockedReceivers, blockedReceivers)) { batch.Add(package.package); packages.RemoveAt(i); @@ -1939,7 +1932,7 @@ NetworkProtocolChannel } } - if (WriteBatch(senderClientId, receiverClientId, batch)) + if (WriteBatch(receiverClientId, blockedReceivers, batch)) { disconnected = true; return; @@ -1970,19 +1963,94 @@ NetworkProtocolChannel } protected: - void QueuePackage(vint senderClientId, Nullable receiverClientId, const TPackage& package) + static bool ClientIdsEqual(const ClientIdList& a, const ClientIdList& b) + { + if (a.Count() != b.Count()) + { + return false; + } + for (vint i = 0; i < a.Count(); i++) + { + if (a[i] != b[i]) + { + return false; + } + } + return true; + } + + void QueuePackage(Nullable receiverClientId, const ClientIdList& blockedReceivers, const TPackage& package) { SPIN_LOCK(lockQueuedPackages) { QueuedPackage queuedPackage; - queuedPackage.senderClientId = senderClientId; queuedPackage.receiverClientId = receiverClientId; + for (auto blockedReceiver : blockedReceivers) + { + queuedPackage.blockedReceivers.Add(blockedReceiver); + } queuedPackage.package = package; - queuedPackages.Add(queuedPackage); + queuedPackages.Add(std::move(queuedPackage)); } } }; +} + +#endif + + +/*********************************************************************** +.\INTERPROCESS\CHANNELIMPLS\CHANNELPACKAGE.H +***********************************************************************/ +/*********************************************************************** +Vczh Library++ 3.0 +Developer: Zihan Chen(vczh) + +Interfaces: +***********************************************************************/ + +#ifndef VCZH_INTERPROCESS_CHANNELIMPLS_CHANNELPACKAGE +#define VCZH_INTERPROCESS_CHANNELIMPLS_CHANNELPACKAGE + + +namespace vl::inter_process +{ + struct NetworkPackage + { + using ClientIdList = collections::List; + + Nullable clientId; + Nullable extraClientIds; + WString channelName; + WString messageBody; + + static NetworkPackage Create(Nullable _clientId, const WString& _channelName, const WString& _messageBody); + static NetworkPackage Create(Nullable _clientId, const ClientIdList& _extraClientIds, const WString& _channelName, const WString& _messageBody); + static WString ToString(const NetworkPackage& package); + static void Parse(const WString& str, NetworkPackage& package); + }; +} + +#endif + + +/*********************************************************************** +.\INTERPROCESS\CHANNELIMPLS\CHANNELCLIENTBASEIMPL.H +***********************************************************************/ +/*********************************************************************** +Vczh Library++ 3.0 +Developer: Zihan Chen(vczh) + +Interfaces: +***********************************************************************/ + +#ifndef VCZH_INTERPROCESS_CHANNELIMPLS_CHANNELCLIENTBASEIMPL +#define VCZH_INTERPROCESS_CHANNELIMPLS_CHANNELCLIENTBASEIMPL + + +namespace vl::inter_process +{ /*********************************************************************** NetworkProtocolChannelClientBase ***********************************************************************/ @@ -2000,24 +2068,29 @@ NetworkProtocolChannelClientBase class Channel : public NetworkProtocolChannel { using Base = NetworkProtocolChannel; + using ClientIdList = typename Base::ClientIdList; private: NetworkProtocolChannelClientBase* client = nullptr; - void ValidatePackage(vint senderClientId, Nullable receiverClientId) override + void ValidatePackage(Nullable receiverClientId, const ClientIdList& blockedReceivers) override { auto currentClientId = client->GetClientId(); CHECK_ERROR(currentClientId != -1, L"NetworkProtocolChannelClient::Channel needs to be connected before sending."); - CHECK_ERROR(senderClientId == currentClientId, L"NetworkProtocolChannelClient::Channel needs senderClientId to match the current client id."); if (receiverClientId) { CHECK_ERROR(receiverClientId.Value() > 0, L"NetworkProtocolChannelClient::Channel needs a valid receiverClientId."); + CHECK_ERROR(blockedReceivers.Count() == 0, L"NetworkProtocolChannelClient::Channel cannot block receivers when sending to a specified client."); + } + for (auto blockedReceiver : blockedReceivers) + { + CHECK_ERROR(blockedReceiver > 0 && blockedReceiver != currentClientId, L"NetworkProtocolChannelClient::Channel needs valid blockedReceivers."); } } - bool WriteBatch(vint, Nullable receiverClientId, const PackageList& batch) override + bool WriteBatch(Nullable receiverClientId, const ClientIdList& blockedReceivers, const PackageList& batch) override { - return client->SendBatch(receiverClientId, this->channelName, batch); + return client->SendBatch(receiverClientId, blockedReceivers, this->channelName, batch); } public: @@ -2080,7 +2153,7 @@ NetworkProtocolChannelClientBase return connected; } - virtual bool SendBatch(Nullable receiverClientId, const WString& channelName, const PackageList& batch) = 0; + virtual bool SendBatch(Nullable receiverClientId, const NetworkPackage::ClientIdList& blockedReceivers, const WString& channelName, const PackageList& batch) = 0; void ReceiveBatch(const WString& channelName, vint senderClientId, const WString& messageBody) { @@ -2214,6 +2287,27 @@ NetworkProtocolChannelClientBase } }; +} + +#endif + + +/*********************************************************************** +.\INTERPROCESS\CHANNELIMPLS\CHANNELCLIENTIMPL.H +***********************************************************************/ +/*********************************************************************** +Vczh Library++ 3.0 +Developer: Zihan Chen(vczh) + +Interfaces: +***********************************************************************/ + +#ifndef VCZH_INTERPROCESS_CHANNELIMPLS_CHANNELCLIENTIMPL +#define VCZH_INTERPROCESS_CHANNELIMPLS_CHANNELCLIENTIMPL + + +namespace vl::inter_process +{ /*********************************************************************** NetworkProtocolChannelClient ***********************************************************************/ @@ -2279,7 +2373,7 @@ NetworkProtocolChannelClient collections::List queuedPackagesBeforeConnected; protected: - bool SendBatch(Nullable receiverClientId, const WString& channelName, const PackageList& batch) override + bool SendBatch(Nullable receiverClientId, const NetworkPackage::ClientIdList& blockedReceivers, const WString& channelName, const PackageList& batch) override { if (this->GetStatus() != ClientStatus::Connected) { @@ -2289,7 +2383,7 @@ NetworkProtocolChannelClient CHECK_ERROR(npClient, L"NetworkProtocolChannelClient::SendBatch needs an established network connection."); WString messageBody; TSerialization::Serialize(this->context, batch, messageBody); - npClient->GetConnection()->SendString(NetworkPackage::ToString(NetworkPackage::Create(std::move(receiverClientId), channelName, messageBody))); + npClient->GetConnection()->SendString(NetworkPackage::ToString(NetworkPackage::Create(std::move(receiverClientId), blockedReceivers, channelName, messageBody))); return false; } @@ -2378,7 +2472,7 @@ NetworkProtocolChannelClient ~NetworkProtocolChannelClient() { - if (npClient) + if (npClient && this->GetStatus() != ClientStatus::Disconnected) { npClient->GetConnection()->Stop(); } @@ -2424,25 +2518,58 @@ NetworkProtocolChannelClient } }; +} + +#endif + + +/*********************************************************************** +.\INTERPROCESS\CHANNELIMPLS\LOCALCHANNELCLIENTIMPL.H +***********************************************************************/ +/*********************************************************************** +Vczh Library++ 3.0 +Developer: Zihan Chen(vczh) + +Interfaces: +***********************************************************************/ + +#ifndef VCZH_INTERPROCESS_CHANNELIMPLS_LOCALCHANNELCLIENTIMPL +#define VCZH_INTERPROCESS_CHANNELIMPLS_LOCALCHANNELCLIENTIMPL + + +namespace vl::inter_process +{ /*********************************************************************** NetworkProtocolLocalChannelClient ***********************************************************************/ template + class INetworkProtocolLocalChannelServer : public virtual Interface + { + protected: + using PackageList = typename TSerialization::SourceType; + + public: + virtual bool SendFromLocalClient(Nullable receiverClientId, const NetworkPackage::ClientIdList& blockedReceivers, vint senderClientId, const WString& channelName, const PackageList& batch) = 0; + virtual void BroadcastError(const WString& errorMessage) = 0; + }; + + template class NetworkProtocolChannelServer; template class NetworkProtocolLocalChannelClient : public NetworkProtocolChannelClientBase { - friend class NetworkProtocolChannelServer; + template + friend class NetworkProtocolChannelServer; private: using Base = NetworkProtocolChannelClientBase; using PackageList = typename TSerialization::SourceType; - NetworkProtocolChannelServer* localServer = nullptr; + INetworkProtocolLocalChannelServer* localServer = nullptr; - bool ConnectLocalServer(NetworkProtocolChannelServer* server, vint assignedClientId) + bool ConnectLocalServer(INetworkProtocolLocalChannelServer* server, vint assignedClientId) { CHECK_ERROR(server, L"NetworkProtocolLocalChannelClient::ConnectLocalServer needs a valid server."); localServer = server; @@ -2461,7 +2588,7 @@ NetworkProtocolLocalChannelClient } protected: - bool SendBatch(Nullable receiverClientId, const WString& channelName, const PackageList& batch) override + bool SendBatch(Nullable receiverClientId, const NetworkPackage::ClientIdList& blockedReceivers, const WString& channelName, const PackageList& batch) override { if (this->GetStatus() != ClientStatus::Connected) { @@ -2470,7 +2597,7 @@ NetworkProtocolLocalChannelClient if (localServer) { - return localServer->SendFromLocalClient(receiverClientId, this->GetClientId(), channelName, batch); + return localServer->SendFromLocalClient(receiverClientId, blockedReceivers, this->GetClientId(), channelName, batch); } return true; } @@ -2503,12 +2630,37 @@ NetworkProtocolLocalChannelClient } }; +} + +#endif + + +/*********************************************************************** +.\INTERPROCESS\CHANNELIMPLS\CHANNELSERVERIMPL.H +***********************************************************************/ +/*********************************************************************** +Vczh Library++ 3.0 +Developer: Zihan Chen(vczh) + +Interfaces: +***********************************************************************/ + +#ifndef VCZH_INTERPROCESS_CHANNELIMPLS_CHANNELSERVERIMPL +#define VCZH_INTERPROCESS_CHANNELIMPLS_CHANNELSERVERIMPL + +#include + +namespace vl::inter_process +{ /*********************************************************************** NetworkProtocolChannelServer ***********************************************************************/ - template - class NetworkProtocolChannelServer : public Object, public virtual IChannelServer, public virtual INetworkProtocolServer + template + class NetworkProtocolChannelServer + : public TServerBase + , public virtual IChannelServer + , public virtual INetworkProtocolLocalChannelServer { friend class NetworkProtocolLocalChannelClient; @@ -2629,7 +2781,7 @@ NetworkProtocolChannelServer } } - if (OnClientConnected(assignedClientId, availableChannels.Keys()) == WaitForClientResult::Accept) + if (OnClientConnected(assignedClientId, availableChannels.Keys(), nullptr) == WaitForClientResult::Accept) { bool accepted = false; { @@ -2665,15 +2817,24 @@ NetworkProtocolChannelServer } CHECK_ERROR(ClientHasChannel(connection->clientId, package.channelName), L"NetworkProtocolChannelServer received a message from a client without the specified channel."); + NetworkPackage::ClientIdList noBlockedReceivers; + auto blockedReceivers = package.extraClientIds ? &package.extraClientIds.Value() : &noBlockedReceivers; if (package.clientId) { auto receiverClientId = package.clientId.Value(); CHECK_ERROR(receiverClientId > 0 && ClientHasChannel(receiverClientId, package.channelName), L"NetworkProtocolChannelServer received a message to a client without the specified channel."); } + else + { + for (auto blockedReceiver : *blockedReceivers) + { + CHECK_ERROR(blockedReceiver > 0 && ClientHasChannel(blockedReceiver, package.channelName), L"NetworkProtocolChannelServer received a message blocking a client without the specified channel."); + } + } PackageList batch; TSerialization::Deserialize(context, package.messageBody, batch); - SendBatch(package.clientId, connection->clientId, connection->clientId, package.channelName, batch); + SendBatch(package.clientId, *blockedReceivers, connection->clientId, connection->clientId, package.channelName, batch); } void OnConnectionDisconnected(Connection* connection) @@ -2730,7 +2891,7 @@ NetworkProtocolChannelServer } } - bool SendBatch(Nullable receiverClientId, vint senderClientId, vint excludedClientId, const WString& channelName, const PackageList& batch) + bool SendBatch(Nullable receiverClientId, const NetworkPackage::ClientIdList& blockedReceivers, vint senderClientId, vint excludedClientId, const WString& channelName, const PackageList& batch) { if (IsStopped()) { @@ -2774,14 +2935,14 @@ NetworkProtocolChannelServer { for (auto&& connection : connections.Values()) { - if (connection->clientId != excludedClientId && clientChannels.Contains(connection->clientId, channelName)) + if (connection->clientId != excludedClientId && !blockedReceivers.Contains(connection->clientId) && clientChannels.Contains(connection->clientId, channelName)) { targetConnections.Add(connection); } } for (auto&& clientId : localClients.Keys()) { - if (clientId != excludedClientId && clientChannels.Contains(clientId, channelName)) + if (clientId != excludedClientId && !blockedReceivers.Contains(clientId) && clientChannels.Contains(clientId, channelName)) { targetLocalClients.Add(localClients[clientId]); } @@ -2800,14 +2961,21 @@ NetworkProtocolChannelServer return false; } - bool SendFromLocalClient(Nullable receiverClientId, vint senderClientId, const WString& channelName, const PackageList& batch) + bool SendFromLocalClient(Nullable receiverClientId, const NetworkPackage::ClientIdList& blockedReceivers, vint senderClientId, const WString& channelName, const PackageList& batch) override { CHECK_ERROR(senderClientId > 0 && ClientHasChannel(senderClientId, channelName), L"NetworkProtocolChannelServer received a message from a local client without the specified channel."); if (receiverClientId) { CHECK_ERROR(receiverClientId.Value() > 0 && ClientHasChannel(receiverClientId.Value(), channelName), L"NetworkProtocolChannelServer received a message from a local client to a client without the specified channel."); } - return SendBatch(receiverClientId, senderClientId, senderClientId, channelName, batch); + else + { + for (auto blockedReceiver : blockedReceivers) + { + CHECK_ERROR(blockedReceiver > 0 && ClientHasChannel(blockedReceiver, channelName), L"NetworkProtocolChannelServer received a message from a local client blocking a client without the specified channel."); + } + } + return SendBatch(receiverClientId, blockedReceivers, senderClientId, senderClientId, channelName, batch); } public: @@ -2840,9 +3008,10 @@ NetworkProtocolChannelServer CHECK_ERROR(!stopped, L"NetworkProtocolChannelServer has stopped."); started = true; } + TServerBase::Start(); } - WaitForClientResult OnClientConnected(vint clientId, const typename IChannelClient::ChannelNameList& availableChannels) override + WaitForClientResult OnClientConnected(vint clientId, const typename IChannelClient::ChannelNameList& availableChannels, IChannelClient* localClient) override { // default implementation allows all clients to connect return WaitForClientResult::Accept; @@ -2853,10 +3022,24 @@ NetworkProtocolChannelServer // default implementation does nothing } - NetworkProtocolChannelServer( - const typename TSerialization::ContextType& _context = {} - ) - : context(_context) + NetworkProtocolChannelServer() + : TServerBase() + , context() + { + } + + template + requires (!std::is_constructible_v) + NetworkProtocolChannelServer(TFirst&& first, TArgs&&... args) + : TServerBase(std::forward(first), std::forward(args)...) + , context() + { + } + + template + NetworkProtocolChannelServer(const typename TSerialization::ContextType& _context, TArgs&&... args) + : TServerBase(std::forward(args)...) + , context(_context) { } @@ -2900,7 +3083,7 @@ NetworkProtocolChannelServer } } - if (OnClientConnected(assignedClientId, channels.Keys()) == WaitForClientResult::Reject) + if (OnClientConnected(assignedClientId, channels.Keys(), localClient.Obj()) == WaitForClientResult::Reject) { return -1; } @@ -3060,21 +3243,23 @@ NetworkProtocolChannelServer if (shouldStop) { - for (auto&& connection : stoppingPendingConnections) - { - connection->connection->Stop(); - } - for (auto&& connection : stoppingConnections) - { - connection->connection->Stop(); - OnClientDisconnected(connection->clientId); - } for (vint i = 0; i < stoppingLocalClients.Count(); i++) { NotifyLocalClientDisconnected(stoppingLocalClients[i]); OnClientDisconnected(stoppingLocalClientIds[i]); } } + + TServerBase::Stop(); + + if (shouldStop) + { + // Network connections are owned and stopped by TServerBase. + for (auto&& connection : stoppingConnections) + { + OnClientDisconnected(connection->clientId); + } + } } bool IsStopped() override @@ -3084,7 +3269,7 @@ NetworkProtocolChannelServer { result = stopped; } - return result; + return result || TServerBase::IsStopped(); } }; } @@ -3092,6 +3277,54 @@ NetworkProtocolChannelServer #endif +/*********************************************************************** +.\INTERPROCESS\NETWORKPROTOCOLCHANNEL.H +***********************************************************************/ +/*********************************************************************** +Vczh Library++ 3.0 +Developer: Zihan Chen(vczh) + +Interfaces: +***********************************************************************/ + +#ifndef VCZH_INTERPROCESS_NETWORKPROTOCOLCHANNEL +#define VCZH_INTERPROCESS_NETWORKPROTOCOLCHANNEL + + +/*********************************************************************** +Hooking IChannelServer/IChannelClient to INetworkProtocolServer/INetworkProtocolClient + +The serialization contract is the same to the one described in ChannelSerialization.h +SourceType will be List +DestType will be WString + +NetworkPackage will be used as text message parsing and formatting for INetworkProtocolConnection. +BatchWrite belongs to IChannel, meaning each channel sends its own batch messages in one NetworkPackage. +channelName will be either a system channel or a user defined channel. +messageBody represents a list of TPackage. +The first section is "clientId,extraClientId1,extraClientId2,...". + Empty clientId means no direct client id. + Empty or missing extra client ids are equivalent. +When sending from client to server, clientId means the target client. + Empty means broadcasting. + When clientId is empty, extraClientIds means blocked receiver client ids. + When clientId is not empty, extraClientIds are ignored. +When sending from server to client, clientId means the source client. + Channel messages delivered by the server always carry a source client id. + +When a client establishes a connection to the server, channel names will be sent to the server: + clientId will be empty, it does not mean broadcasting. + channelName will be empty. + messageBody will be all available channel names joined by "!", as "!" cannot be part of the channel name anyway. +After the server receives the first message from a client, an client id will be sent to the client: + clientId is the assigned client id, starting from 1. + channelName will be empty. + messageBody will be empty. +***********************************************************************/ + +#endif + + /*********************************************************************** .\STREAM\INTERFACES.H ***********************************************************************/ diff --git a/Import/VlppWorkflowCompiler.cpp b/Import/VlppWorkflowCompiler.cpp index 0b7ba03f..85042d4e 100644 --- a/Import/VlppWorkflowCompiler.cpp +++ b/Import/VlppWorkflowCompiler.cpp @@ -10127,6 +10127,15 @@ namespace vl return testing; } + Ptr CreateIsType(Ptr expression, Ptr type) + { + auto testing = Ptr(new WfTypeTestingExpression); + testing->test = WfTypeTesting::IsType; + testing->expression = expression; + testing->type = type; + return testing; + } + Ptr CreateBool(bool value) { auto expression = Ptr(new WfLiteralExpression); @@ -11077,44 +11086,6 @@ namespace vl return functionDecl; } - Ptr BuildRegisterService() - { - auto block = CreateBlock(); - auto registerBranch = CreateBlock(); - AddStatement( - registerBranch, - CreateExpressionStatement( - CreateCall( - CreateMember(CreateMember(CreateReference(L"_lc"), L"Dispatcher"), L"RegisterService"), - CreateReference(L"typeId"), - CreateCall(CreateMember(CreateReference(L"_lc"), L"PtrToRef"), CreateReference(L"service")) - ) - ) - ); - auto nonCtorBranch = CreateBlock(); - AddStatement(nonCtorBranch, CreateRaise(L"RPC service type id is not an @rpc:Ctor interface.")); - auto invalidTypeIdBranch = CreateBlock(); - AddStatement(invalidTypeIdBranch, CreateRaise(L"RPC service type id does not exist.")); - auto invalidRegisterBranch = CreateBlock(); - AddStatement( - invalidRegisterBranch, - CreateIf( - CreateCall(CreateReference(L"rpcwrapper_IsInterfaceTypeId"), CreateReference(L"typeId")), - nonCtorBranch, - invalidTypeIdBranch - ) - ); - AddStatement( - block, - CreateIf( - CreateCall(CreateReference(L"rpcwrapper_IsCtorInterfaceTypeId"), CreateReference(L"typeId")), - registerBranch, - invalidRegisterBranch - ) - ); - return block; - } - Ptr GenerateObjectOpsFactory(const List& interfaces) { auto functionDecl = CreateFunctionDeclaration(L"rpcops_IRpcObjectOps", CreateTypeFromCpp>(), WfFunctionKind::Normal); @@ -11161,15 +11132,6 @@ namespace vl newOps->declarations.Add(objectHold); } - { - auto registerService = CreateFunctionDeclaration(L"RegisterService", CreatePredefinedType(WfPredefinedTypeName::Void), WfFunctionKind::Override); - registerService->arguments.Add(CreateFunctionArgument(L"typeId", CreatePredefinedType(WfPredefinedTypeName::Int))); - registerService->arguments.Add(CreateFunctionArgument(L"service", CreateTypeFromCpp>())); - auto block = registerService->statement.Cast(); - AddStatement(block, BuildRegisterService()); - newOps->declarations.Add(registerService); - } - AddStatement(functionDecl->statement.Cast(), CreateReturn(newOps)); return functionDecl; } @@ -11246,6 +11208,42 @@ namespace vl return nullptr; } + void SortInterfaceModelsLeafFirst(const List& interfaces, List& sortedInterfaces) + { + sortedInterfaces.Clear(); + + List typeFullNames; + Group dependencyGroup; + for (auto&& interfaceModel : interfaces) + { + typeFullNames.Add(interfaceModel.fullName); + } + for (auto&& interfaceModel : interfaces) + { + for (auto&& baseFullName : interfaceModel.baseFullNames) + { + if (typeFullNames.Contains(baseFullName)) + { + dependencyGroup.Add(baseFullName, interfaceModel.fullName); + } + } + } + + PartialOrderingProcessor pop; + pop.InitWithGroup(typeFullNames, dependencyGroup); + pop.Sort(); + + for (auto&& component : pop.components) + { + for (vint i = 0; i < component.nodeCount; i++) + { + auto interfaceModel = FindInterfaceModel(interfaces, typeFullNames[component.firstNode[i]]); + CHECK_ERROR(interfaceModel, L"SortInterfaceModelsLeafFirst: Invalid RPC interface name."); + sortedInterfaces.Add(interfaceModel); + } + } + } + bool ContainsEventModel(const List& events, const WString& fullName) { for (auto eventModel : events) @@ -11928,6 +11926,28 @@ namespace vl AddStatement(block, switchStat); return functionDecl; } + + Ptr GenerateWrapperGetTypeId(const List& interfaces) + { + auto functionDecl = CreateFunctionDeclaration(L"rpcwrapper_GetTypeId", CreatePredefinedType(WfPredefinedTypeName::Int), WfFunctionKind::Normal); + functionDecl->arguments.Add(CreateFunctionArgument(L"obj", CreatePredefinedType(WfPredefinedTypeName::Object))); + auto block = functionDecl->statement.Cast(); + + List sortedInterfaces; + SortInterfaceModelsLeafFirst(interfaces, sortedInterfaces); + for (auto interfaceModel : sortedInterfaces) + { + AddStatement( + block, + CreateIf( + CreateIsType(CreateReference(L"obj"), CreateRawType(interfaceModel->fullName)), + CreateReturn(CreateRpcConstantReference(L"rpctype_", interfaceModel->fullName)) + )); + } + + AddStatement(block, CreateReturn(CreateInt(rpc_controller::RpcTypeId_NotFound))); + return functionDecl; + } } Ptr GenerateModuleRpc(WfLexicalScopeManager* manager, WString assemblyName) @@ -12016,6 +12036,7 @@ namespace vl } module->declarations.Add(GenerateWrapperDispatcher(interfaces, opsInterfaceName)); + module->declarations.Add(GenerateWrapperGetTypeId(interfaces)); return module; } @@ -12727,7 +12748,6 @@ namespace vl void CollectMangledNames(WfLexicalScopeManager* manager); List BuildInterfaceModels(WfLexicalScopeManager* manager); bool HasRpcEvents(const List& interfaces); - Ptr BuildRegisterService(); WString GetRpcOpsInterfaceName(const WString& assemblyName); WString GetRpcOpsInvokeMethodName(const RpcMethodModel& methodModel); WString GetRpcOpsInvokeEventName(const RpcEventModel& eventModel); @@ -13940,15 +13960,6 @@ namespace vl newOps->declarations.Add(objectHold); } - { - auto registerService = CreateFunctionDeclaration(L"RegisterService", CreatePredefinedType(WfPredefinedTypeName::Void), WfFunctionKind::Override); - registerService->arguments.Add(CreateFunctionArgument(L"typeId", CreatePredefinedType(WfPredefinedTypeName::Int))); - registerService->arguments.Add(CreateFunctionArgument(L"service", CreateTypeFromCpp>())); - auto block = registerService->statement.Cast(); - AddStatement(block, BuildRegisterService()); - newOps->declarations.Add(registerService); - } - AddStatement(functionDecl->statement.Cast(), CreateReturn(newOps)); return functionDecl; } diff --git a/Import/VlppWorkflowLibrary.cpp b/Import/VlppWorkflowLibrary.cpp index bc892f77..5e93dd5e 100644 --- a/Import/VlppWorkflowLibrary.cpp +++ b/Import/VlppWorkflowLibrary.cpp @@ -1037,6 +1037,7 @@ TypeName IMPL_TYPE_INFO_RENAME(vl::rpc_controller::RpcException, system::RpcException) IMPL_TYPE_INFO_RENAME(vl::rpc_controller::RpcByvalReturnValue, system::RpcByvalReturnValue) IMPL_TYPE_INFO_RENAME(vl::rpc_controller::IRpcSerializer, system::IRpcSerializer) + IMPL_TYPE_INFO_RENAME(vl::rpc_controller::IRpcJsonMessageDispatcher::RequestType, system::IRpcJsonMessageDispatcher::RequestType) IMPL_TYPE_INFO_RENAME(vl::rpc_controller::IRpcJsonMessageDispatcher, system::IRpcJsonMessageDispatcher) IMPL_TYPE_INFO_RENAME(vl::rpc_controller::IRpcListOps, system::IRpcListOps) IMPL_TYPE_INFO_RENAME(vl::rpc_controller::IRpcListEventOps, system::IRpcListEventOps) @@ -1142,9 +1143,15 @@ WfLoadLibraryTypes CLASS_MEMBER_METHOD(Deserialize, { L"value" }) END_INTERFACE_MEMBER(vl::rpc_controller::IRpcSerializer) + BEGIN_ENUM_ITEM(vl::rpc_controller::IRpcJsonMessageDispatcher::RequestType) + ENUM_CLASS_ITEM(Direct) + ENUM_CLASS_ITEM(Broadcast) + ENUM_CLASS_ITEM(BroadcastAndDrop) + END_ENUM_ITEM(vl::rpc_controller::IRpcJsonMessageDispatcher::RequestType) + BEGIN_INTERFACE_MEMBER(vl::rpc_controller::IRpcJsonMessageDispatcher) CLASS_MEMBER_METHOD(AllocateRequestId, NO_PARAMETER) - CLASS_MEMBER_METHOD(OnJsonRequest, { L"message" }) + CLASS_MEMBER_METHOD(OnJsonRequest, { L"message" _ L"requestType" }) END_INTERFACE_MEMBER(vl::rpc_controller::IRpcJsonMessageDispatcher) BEGIN_INTERFACE_MEMBER(vl::rpc_controller::IRpcListOps) @@ -1181,7 +1188,6 @@ WfLoadLibraryTypes CLASS_MEMBER_METHOD(InvokeMethod, { L"ref" _ L"methodId" _ L"arguments" }) CLASS_MEMBER_METHOD(EndInvokeMethod, { L"slot" }) CLASS_MEMBER_METHOD(ObjectHold, { L"ref" _ L"remoteClientId" _ L"hold" }) - CLASS_MEMBER_METHOD(RegisterService, { L"typeId" _ L"service" }) END_INTERFACE_MEMBER(vl::rpc_controller::IRpcObjectOps) BEGIN_INTERFACE_MEMBER(vl::rpc_controller::IRpcObjectEventOps) @@ -1189,17 +1195,14 @@ WfLoadLibraryTypes END_INTERFACE_MEMBER(vl::rpc_controller::IRpcObjectEventOps) BEGIN_INTERFACE_MEMBER_NOPROXY(vl::rpc_controller::IRpcOperations) - CLASS_MEMBER_PROPERTY_READONLY_FAST(ListOps) CLASS_MEMBER_PROPERTY_READONLY_FAST(ObjectOps) - CLASS_MEMBER_PROPERTY_READONLY_FAST(ListEventOps) CLASS_MEMBER_PROPERTY_READONLY_FAST(ObjectEventOps) END_INTERFACE_MEMBER(vl::rpc_controller::IRpcOperations) BEGIN_INTERFACE_MEMBER_NOPROXY(vl::rpc_controller::IRpcDispatcher) CLASS_MEMBER_METHOD(Finalize, NO_PARAMETER) - CLASS_MEMBER_METHOD(IsRegisteredService, { L"ref" }) - CLASS_MEMBER_METHOD(RegisterService, { L"typeId" _ L"ref" }) - CLASS_MEMBER_METHOD(RequestService, { L"typeId" }) + CLASS_MEMBER_METHOD(Initialize, NO_PARAMETER) + CLASS_MEMBER_METHOD(DeclareLocalService, { L"ref" }) CLASS_MEMBER_METHOD(BroadcastFromClient_ObjectEventOps, { L"selfClientId" }) CLASS_MEMBER_METHOD(SendToClient_ObjectOps, { L"targetClientId" }) END_INTERFACE_MEMBER(vl::rpc_controller::IRpcDispatcher) @@ -1215,6 +1218,7 @@ WfLoadLibraryTypes BEGIN_INTERFACE_MEMBER_NOPROXY(vl::rpc_controller::IRpcLifecycle) CLASS_MEMBER_METHOD(Finalize, NO_PARAMETER) + CLASS_MEMBER_METHOD(Initialize, NO_PARAMETER) CLASS_MEMBER_PROPERTY_READONLY_FAST(ClientId) CLASS_MEMBER_PROPERTY_READONLY_FAST(Dispatcher) CLASS_MEMBER_PROPERTY_READONLY_FAST(Controller) @@ -1223,8 +1227,10 @@ WfLoadLibraryTypes CLASS_MEMBER_METHOD(PtrToRef, { L"obj" }) CLASS_MEMBER_METHOD(LocalObjectHold, { L"ref" _ L"remoteClientId" }) CLASS_MEMBER_METHOD(LocalObjectUnhold, { L"ref" _ L"remoteClientId" }) - CLASS_MEMBER_METHOD(RegisterService, { L"fullName" _ L"service" }) - CLASS_MEMBER_METHOD(RequestService, { L"fullName" }) + CLASS_MEMBER_METHOD(RegisterLocalService, { L"typeId" _ L"service" }) + CLASS_MEMBER_METHOD(DeclareRemoteService, { L"ref" }) + CLASS_MEMBER_METHOD(GetTypeIdFromName, { L"typeName" }) + CLASS_MEMBER_METHOD(RequestService, { L"typeName" }) CLASS_MEMBER_STATIC_EXTERNALMETHOD(RpcBoxByref, { L"trivial" _ L"lc" }, vl::rpc_controller::RpcObjectReference(*)(vl::Ptr, vl::rpc_controller::IRpcLifecycle*), vl::rpc_controller::RpcBoxByref) CLASS_MEMBER_STATIC_EXTERNALMETHOD(RpcUnboxByref, { L"serializable" _ L"lc" }, vl::Ptr(*)(vl::rpc_controller::RpcObjectReference, vl::rpc_controller::IRpcLifecycle*), vl::rpc_controller::RpcUnboxByref) CLASS_MEMBER_STATIC_EXTERNALMETHOD(RpcCopyByval, { L"trivial" _ L"lc" }, Value(*)(const Value&, vl::rpc_controller::IRpcLifecycle*), vl::rpc_controller::RpcCopyByval) @@ -1916,17 +1922,10 @@ namespace vl { } - void RpcControllerDefault::Register(Ptr _objectCallback, Ptr _eventCallback, Ptr _listCallback, Ptr _listEventCallback) + void RpcControllerDefault::Register(Ptr _objectCallback, Ptr _eventCallback) { objectCallback = _objectCallback; eventCallback = _eventCallback; - listCallback = _listCallback; - listEventCallback = _listEventCallback; - } - - IRpcListOps* RpcControllerDefault::GetListOps() - { - return listCallback.Obj(); } IRpcObjectOps* RpcControllerDefault::GetObjectOps() @@ -1934,11 +1933,6 @@ namespace vl return objectCallback.Obj(); } - IRpcListEventOps* RpcControllerDefault::GetListEventOps() - { - return listEventCallback.Obj(); - } - IRpcObjectEventOps* RpcControllerDefault::GetObjectEventOps() { return eventCallback.Obj(); @@ -2400,6 +2394,11 @@ namespace vl return GetJsonInt(GetJsonObjectField(object, WString::Unmanaged(L"targetClientId"))); } + bool IsRpcMethodPrefix(const WString& rpcMethod, const WString& prefix) + { + return rpcMethod.Length() >= prefix.Length() && rpcMethod.Left(prefix.Length()) == prefix; + } + Ptr MethodResultToJsonResponse(const Value& result) { if (result.GetValueType() == Value::SharedPtr) @@ -2645,6 +2644,42 @@ namespace vl CHECK_FAIL(L"Unsupported RPC JSON node."); } +/*********************************************************************** +* IRpcJsonMessageDispatcher +***********************************************************************/ + + Ptr IRpcJsonMessageDispatcher::DefaultTranslate( + Ptr message, + RequestType requestType, + IRpcObjectOps* objectOps, + IRpcObjectEventOps* objectEventOps, + IRpcDispatcher* dispatcher, + IRpcLifecycle* lifecycle + ) + { +#define ERROR_MESSAGE_PREFIX L"vl::rpc_controller::IRpcJsonMessageDispatcher::DefaultTranslate(Ptr, RequestType, IRpcObjectOps*, IRpcObjectEventOps*, IRpcDispatcher*, IRpcLifecycle*)#" + auto request = GetJsonObject(message); + auto rpcMethod = GetJsonString(GetJsonObjectField(request, WString::Unmanaged(L"rpcMethod"))); + + if (requestType == RequestType::Direct && IsRpcMethodPrefix(rpcMethod, WString::Unmanaged(L"Request:IObjectOps_"))) + { + return RpcJsonObjectOps::Translate(message, objectOps, lifecycle); + } + else if (requestType == RequestType::Broadcast && IsRpcMethodPrefix(rpcMethod, WString::Unmanaged(L"Request:IObjectEventOps_"))) + { + return RpcJsonObjectEventOps::Translate(message, objectEventOps, lifecycle); + } + else if (requestType == RequestType::BroadcastAndDrop && rpcMethod == WString::Unmanaged(L"Request:IRpcDispatcher_DeclareRemoteService")) + { + RpcJsonDispatcher::Translate(message, dispatcher, lifecycle); + return nullptr; + } + + CHECK_FAIL(ERROR_MESSAGE_PREFIX L"Unknown JSON RPC method."); + return nullptr; +#undef ERROR_MESSAGE_PREFIX + } + /*********************************************************************** * RpcJsonObjectOps ***********************************************************************/ @@ -2657,11 +2692,10 @@ namespace vl #undef ERROR_MESSAGE_PREFIX } - RpcJsonObjectOps::RpcJsonObjectOps(vint _sourceClientId, vint _targetClientId, IRpcJsonMessageDispatcher* _dispatcher, IRpcLifecycle* _lifecycle) + RpcJsonObjectOps::RpcJsonObjectOps(vint _sourceClientId, vint _targetClientId, IRpcJsonMessageDispatcher* _dispatcher) : sourceClientId(_sourceClientId) , targetClientId(_targetClientId) , dispatcher(_dispatcher) - , lifecycle(_lifecycle) { #define ERROR_MESSAGE_PREFIX L"vl::rpc_controller::RpcJsonObjectOps::RpcJsonObjectOps(...)#" CHECK_ERROR(dispatcher, ERROR_MESSAGE_PREFIX L"Dispatcher is required."); @@ -2676,14 +2710,14 @@ namespace vl { #define ERROR_MESSAGE_PREFIX L"vl::rpc_controller::RpcJsonObjectOps::InvokeMethod(RpcObjectReference, vint, Ptr)#" CHECK_ERROR(dispatcher, ERROR_MESSAGE_PREFIX L"Dispatcher is required."); - auto request = CreateRpcMessage(WString::Unmanaged(L"IObjectOps_InvokeMethod"), dispatcher->AllocateRequestId(), sourceClientId); + auto request = CreateRpcMessage(WString::Unmanaged(L"Request:IObjectOps_InvokeMethod"), dispatcher->AllocateRequestId(), sourceClientId); AddJsonObjectField(request, WString::Unmanaged(L"targetClientId"), CreateJsonNumber(targetClientId == RpcClientId_Invalid ? ref.clientId : targetClientId)); AddJsonObjectField(request, WString::Unmanaged(L"ref"), CreateRpcObjectReferenceJson(ref)); AddJsonObjectField(request, WString::Unmanaged(L"methodId"), CreateJsonNumber(methodId)); AddJsonObjectField(request, WString::Unmanaged(L"arguments"), ValueArrayToJsonArray(arguments)); - auto response = GetJsonObject(dispatcher->OnJsonRequest(request)); - CHECK_ERROR(GetJsonString(GetJsonObjectField(response, WString::Unmanaged(L"rpcMethod"))) == WString::Unmanaged(L"IObjectOps_InvokeMethod"), ERROR_MESSAGE_PREFIX L"Unexpected response method."); + auto response = GetJsonObject(dispatcher->OnJsonRequest(request, IRpcJsonMessageDispatcher::RequestType::Direct)); + CHECK_ERROR(GetJsonString(GetJsonObjectField(response, WString::Unmanaged(L"rpcMethod"))) == WString::Unmanaged(L"Response:IObjectOps_InvokeMethod"), ERROR_MESSAGE_PREFIX L"Unexpected response method."); CHECK_ERROR(ReadRequestId(response) == ReadRequestId(request), ERROR_MESSAGE_PREFIX L"Unexpected response request id."); return JsonResponseToMethodResult(GetJsonObjectField(response, WString::Unmanaged(L"response"))); #undef ERROR_MESSAGE_PREFIX @@ -2693,12 +2727,12 @@ namespace vl { #define ERROR_MESSAGE_PREFIX L"vl::rpc_controller::RpcJsonObjectOps::EndInvokeMethod(vint)#" CHECK_ERROR(dispatcher, ERROR_MESSAGE_PREFIX L"Dispatcher is required."); - auto request = CreateRpcMessage(WString::Unmanaged(L"IObjectOps_EndInvokeMethod"), dispatcher->AllocateRequestId(), sourceClientId); + auto request = CreateRpcMessage(WString::Unmanaged(L"Request:IObjectOps_EndInvokeMethod"), dispatcher->AllocateRequestId(), sourceClientId); AddJsonObjectField(request, WString::Unmanaged(L"targetClientId"), CreateJsonNumber(targetClientId)); AddJsonObjectField(request, WString::Unmanaged(L"slot"), CreateJsonNumber(slot)); - auto response = GetJsonObject(dispatcher->OnJsonRequest(request)); - CHECK_ERROR(GetJsonString(GetJsonObjectField(response, WString::Unmanaged(L"rpcMethod"))) == WString::Unmanaged(L"IObjectOps_EndInvokeMethod"), ERROR_MESSAGE_PREFIX L"Unexpected response method."); + auto response = GetJsonObject(dispatcher->OnJsonRequest(request, IRpcJsonMessageDispatcher::RequestType::Direct)); + CHECK_ERROR(GetJsonString(GetJsonObjectField(response, WString::Unmanaged(L"rpcMethod"))) == WString::Unmanaged(L"Response:IObjectOps_EndInvokeMethod"), ERROR_MESSAGE_PREFIX L"Unexpected response method."); CHECK_ERROR(ReadRequestId(response) == ReadRequestId(request), ERROR_MESSAGE_PREFIX L"Unexpected response request id."); #undef ERROR_MESSAGE_PREFIX } @@ -2707,31 +2741,14 @@ namespace vl { #define ERROR_MESSAGE_PREFIX L"vl::rpc_controller::RpcJsonObjectOps::ObjectHold(RpcObjectReference, vint, bool)#" CHECK_ERROR(dispatcher, ERROR_MESSAGE_PREFIX L"Dispatcher is required."); - auto request = CreateRpcMessage(WString::Unmanaged(L"IObjectOps_ObjectHold"), dispatcher->AllocateRequestId(), sourceClientId == RpcClientId_Invalid ? remoteClientId : sourceClientId); + auto request = CreateRpcMessage(WString::Unmanaged(L"Request:IObjectOps_ObjectHold"), dispatcher->AllocateRequestId(), sourceClientId == RpcClientId_Invalid ? remoteClientId : sourceClientId); AddJsonObjectField(request, WString::Unmanaged(L"targetClientId"), CreateJsonNumber(targetClientId == RpcClientId_Invalid ? ref.clientId : targetClientId)); AddJsonObjectField(request, WString::Unmanaged(L"ref"), CreateRpcObjectReferenceJson(ref)); AddJsonObjectField(request, WString::Unmanaged(L"remoteClientId"), CreateJsonNumber(remoteClientId)); AddJsonObjectField(request, WString::Unmanaged(L"hold"), CreateJsonBool(hold)); - auto response = GetJsonObject(dispatcher->OnJsonRequest(request)); - CHECK_ERROR(GetJsonString(GetJsonObjectField(response, WString::Unmanaged(L"rpcMethod"))) == WString::Unmanaged(L"IObjectOps_ObjectHold"), ERROR_MESSAGE_PREFIX L"Unexpected response method."); - CHECK_ERROR(ReadRequestId(response) == ReadRequestId(request), ERROR_MESSAGE_PREFIX L"Unexpected response request id."); -#undef ERROR_MESSAGE_PREFIX - } - - void RpcJsonObjectOps::RegisterService(vint typeId, Ptr service) - { -#define ERROR_MESSAGE_PREFIX L"vl::rpc_controller::RpcJsonObjectOps::RegisterService(vint, Ptr)#" - CHECK_ERROR(dispatcher, ERROR_MESSAGE_PREFIX L"Dispatcher is required."); - CHECK_ERROR(lifecycle, ERROR_MESSAGE_PREFIX L"Lifecycle is required."); - auto serviceRef = lifecycle->PtrToRef(service); - auto request = CreateRpcMessage(WString::Unmanaged(L"IObjectOps_RegisterService"), dispatcher->AllocateRequestId(), sourceClientId == RpcClientId_Invalid ? serviceRef.clientId : sourceClientId); - AddJsonObjectField(request, WString::Unmanaged(L"targetClientId"), CreateJsonNumber(targetClientId == RpcClientId_Invalid ? serviceRef.clientId : targetClientId)); - AddJsonObjectField(request, WString::Unmanaged(L"typeId"), CreateJsonNumber(typeId)); - AddJsonObjectField(request, WString::Unmanaged(L"service"), CreateRpcObjectReferenceJson(serviceRef)); - - auto response = GetJsonObject(dispatcher->OnJsonRequest(request)); - CHECK_ERROR(GetJsonString(GetJsonObjectField(response, WString::Unmanaged(L"rpcMethod"))) == WString::Unmanaged(L"IObjectOps_RegisterService"), ERROR_MESSAGE_PREFIX L"Unexpected response method."); + auto response = GetJsonObject(dispatcher->OnJsonRequest(request, IRpcJsonMessageDispatcher::RequestType::Direct)); + CHECK_ERROR(GetJsonString(GetJsonObjectField(response, WString::Unmanaged(L"rpcMethod"))) == WString::Unmanaged(L"Response:IObjectOps_ObjectHold"), ERROR_MESSAGE_PREFIX L"Unexpected response method."); CHECK_ERROR(ReadRequestId(response) == ReadRequestId(request), ERROR_MESSAGE_PREFIX L"Unexpected response request id."); #undef ERROR_MESSAGE_PREFIX } @@ -2739,16 +2756,23 @@ namespace vl Ptr RpcJsonObjectOps::Translate(Ptr message, IRpcObjectOps* ops, IRpcLifecycle* lifecycle) { #define ERROR_MESSAGE_PREFIX L"vl::rpc_controller::RpcJsonObjectOps::Translate(Ptr, IRpcObjectOps*)#" + (void)lifecycle; CHECK_ERROR(ops, ERROR_MESSAGE_PREFIX L"Object ops is required."); auto request = GetJsonObject(message); auto rpcMethod = GetJsonString(GetJsonObjectField(request, WString::Unmanaged(L"rpcMethod"))); auto requestId = ReadRequestId(request); auto sourceClientId = ReadSourceClientId(request); auto targetClientId = ReadTargetClientId(request); - auto response = CreateRpcMessage(rpcMethod, requestId, targetClientId); + auto responseMethod = + rpcMethod == WString::Unmanaged(L"Request:IObjectOps_InvokeMethod") ? WString::Unmanaged(L"Response:IObjectOps_InvokeMethod") : + rpcMethod == WString::Unmanaged(L"Request:IObjectOps_EndInvokeMethod") ? WString::Unmanaged(L"Response:IObjectOps_EndInvokeMethod") : + rpcMethod == WString::Unmanaged(L"Request:IObjectOps_ObjectHold") ? WString::Unmanaged(L"Response:IObjectOps_ObjectHold") : + WString::Empty; + CHECK_ERROR(responseMethod != WString::Empty, ERROR_MESSAGE_PREFIX L"Unexpected RPC method."); + auto response = CreateRpcMessage(responseMethod, requestId, lifecycle ? lifecycle->GetClientId() : targetClientId); AddJsonObjectField(response, WString::Unmanaged(L"targetClientId"), CreateJsonNumber(sourceClientId)); - if (rpcMethod == WString::Unmanaged(L"IObjectOps_InvokeMethod")) + if (rpcMethod == WString::Unmanaged(L"Request:IObjectOps_InvokeMethod")) { auto result = ops->InvokeMethod( GetRpcObjectReferenceFromJson(GetJsonObjectField(request, WString::Unmanaged(L"ref"))), @@ -2757,11 +2781,11 @@ namespace vl ); AddJsonObjectField(response, WString::Unmanaged(L"response"), MethodResultToJsonResponse(result)); } - else if (rpcMethod == WString::Unmanaged(L"IObjectOps_EndInvokeMethod")) + else if (rpcMethod == WString::Unmanaged(L"Request:IObjectOps_EndInvokeMethod")) { ops->EndInvokeMethod(GetJsonInt(GetJsonObjectField(request, WString::Unmanaged(L"slot")))); } - else if (rpcMethod == WString::Unmanaged(L"IObjectOps_ObjectHold")) + else if (rpcMethod == WString::Unmanaged(L"Request:IObjectOps_ObjectHold")) { ops->ObjectHold( GetRpcObjectReferenceFromJson(GetJsonObjectField(request, WString::Unmanaged(L"ref"))), @@ -2769,14 +2793,6 @@ namespace vl GetJsonBool(GetJsonObjectField(request, WString::Unmanaged(L"hold"))) ); } - else if (rpcMethod == WString::Unmanaged(L"IObjectOps_RegisterService")) - { - CHECK_ERROR(lifecycle, ERROR_MESSAGE_PREFIX L"Lifecycle is required to translate RegisterService."); - ops->RegisterService( - GetJsonInt(GetJsonObjectField(request, WString::Unmanaged(L"typeId"))), - lifecycle->RefToPtr(GetRpcObjectReferenceFromJson(GetJsonObjectField(request, WString::Unmanaged(L"service")))) - ); - } else { CHECK_FAIL(ERROR_MESSAGE_PREFIX L"Unexpected RPC method."); @@ -2814,28 +2830,28 @@ namespace vl { #define ERROR_MESSAGE_PREFIX L"vl::rpc_controller::RpcJsonObjectEventOps::InvokeEvent(RpcObjectReference, vint, Ptr)#" CHECK_ERROR(dispatcher, ERROR_MESSAGE_PREFIX L"Dispatcher is required."); - auto request = CreateRpcMessage(WString::Unmanaged(L"IObjectEventOps_InvokeEvent"), dispatcher->AllocateRequestId(), sourceClientId); + auto request = CreateRpcMessage(WString::Unmanaged(L"Request:IObjectEventOps_InvokeEvent"), dispatcher->AllocateRequestId(), sourceClientId); AddJsonObjectField(request, WString::Unmanaged(L"ref"), CreateRpcObjectReferenceJson(ref)); AddJsonObjectField(request, WString::Unmanaged(L"eventId"), CreateJsonNumber(eventId)); AddJsonObjectField(request, WString::Unmanaged(L"arguments"), ValueArrayToJsonArray(arguments)); - auto response = GetJsonObject(dispatcher->OnJsonRequest(request)); - CHECK_ERROR(GetJsonString(GetJsonObjectField(response, WString::Unmanaged(L"rpcMethod"))) == WString::Unmanaged(L"IObjectEventOps_InvokeEvent"), ERROR_MESSAGE_PREFIX L"Unexpected response method."); + auto response = GetJsonObject(dispatcher->OnJsonRequest(request, IRpcJsonMessageDispatcher::RequestType::Broadcast)); + CHECK_ERROR(GetJsonString(GetJsonObjectField(response, WString::Unmanaged(L"rpcMethod"))) == WString::Unmanaged(L"Response:Broadcast_Response"), ERROR_MESSAGE_PREFIX L"Unexpected response method."); CHECK_ERROR(ReadRequestId(response) == ReadRequestId(request), ERROR_MESSAGE_PREFIX L"Unexpected response request id."); return BoxValue(CreateSerializedEventExceptionMap(GetJsonObjectField(response, WString::Unmanaged(L"response")))); #undef ERROR_MESSAGE_PREFIX } - Ptr RpcJsonObjectEventOps::Translate(Ptr message, IRpcObjectEventOps* ops) + Ptr RpcJsonObjectEventOps::Translate(Ptr message, IRpcObjectEventOps* ops, IRpcLifecycle* lifecycle) { #define ERROR_MESSAGE_PREFIX L"vl::rpc_controller::RpcJsonObjectEventOps::Translate(Ptr, IRpcObjectEventOps*)#" CHECK_ERROR(ops, ERROR_MESSAGE_PREFIX L"Object event ops is required."); auto request = GetJsonObject(message); auto rpcMethod = GetJsonString(GetJsonObjectField(request, WString::Unmanaged(L"rpcMethod"))); - CHECK_ERROR(rpcMethod == WString::Unmanaged(L"IObjectEventOps_InvokeEvent"), ERROR_MESSAGE_PREFIX L"Unexpected RPC method."); + CHECK_ERROR(rpcMethod == WString::Unmanaged(L"Request:IObjectEventOps_InvokeEvent"), ERROR_MESSAGE_PREFIX L"Unexpected RPC method."); auto requestId = ReadRequestId(request); auto sourceClientId = ReadSourceClientId(request); - auto response = CreateRpcMessage(rpcMethod, requestId, RpcClientId_Invalid); + auto response = CreateRpcMessage(WString::Unmanaged(L"Response:Broadcast_Response"), requestId, lifecycle ? lifecycle->GetClientId() : RpcClientId_Invalid); AddJsonObjectField(response, WString::Unmanaged(L"targetClientId"), CreateJsonNumber(sourceClientId)); auto result = ops->InvokeEvent( @@ -2848,6 +2864,138 @@ namespace vl #undef ERROR_MESSAGE_PREFIX } +/*********************************************************************** +* RpcJsonDispatcher +***********************************************************************/ + + RpcJsonDispatcher::RpcJsonDispatcher(vint _sourceClientId, IRpcJsonMessageDispatcher* _dispatcher) + : sourceClientId(_sourceClientId) + , dispatcher(_dispatcher) + { +#define ERROR_MESSAGE_PREFIX L"vl::rpc_controller::RpcJsonDispatcher::RpcJsonDispatcher(vint, IRpcJsonMessageDispatcher*)#" + CHECK_ERROR(sourceClientId != RpcClientId_Invalid, ERROR_MESSAGE_PREFIX L"Source client id is required."); + CHECK_ERROR(dispatcher, ERROR_MESSAGE_PREFIX L"Dispatcher is required."); +#undef ERROR_MESSAGE_PREFIX + } + + void RpcJsonDispatcher::Finalize() + { + } + + void RpcJsonDispatcher::Initialize() + { + } + + void RpcJsonDispatcher::DeclareLocalService(RpcObjectReference ref) + { +#define ERROR_MESSAGE_PREFIX L"vl::rpc_controller::RpcJsonDispatcher::DeclareLocalService(RpcObjectReference)#" + CHECK_ERROR(dispatcher, ERROR_MESSAGE_PREFIX L"Dispatcher is required."); + CHECK_ERROR(ref.clientId == sourceClientId, ERROR_MESSAGE_PREFIX L"Client id does not match this dispatcher."); + auto request = CreateRpcMessage(WString::Unmanaged(L"Request:IRpcDispatcher_DeclareRemoteService"), dispatcher->AllocateRequestId(), sourceClientId); + AddJsonObjectField(request, WString::Unmanaged(L"ref"), CreateRpcObjectReferenceJson(ref)); + + auto response = dispatcher->OnJsonRequest(request, IRpcJsonMessageDispatcher::RequestType::BroadcastAndDrop); + CHECK_ERROR(!response, ERROR_MESSAGE_PREFIX L"DeclareLocalService should not receive a response."); +#undef ERROR_MESSAGE_PREFIX + } + + IRpcObjectEventOps* RpcJsonDispatcher::BroadcastFromClient_ObjectEventOps(vint selfClientId) + { +#define ERROR_MESSAGE_PREFIX L"vl::rpc_controller::RpcJsonDispatcher::BroadcastFromClient_ObjectEventOps(vint)#" + CHECK_ERROR(selfClientId == sourceClientId, ERROR_MESSAGE_PREFIX L"Client id does not match this dispatcher."); + if (!objectEventOps) + { + objectEventOps = Ptr(new RpcJsonObjectEventOps(sourceClientId, dispatcher)); + } + return objectEventOps.Obj(); +#undef ERROR_MESSAGE_PREFIX + } + + IRpcObjectOps* RpcJsonDispatcher::SendToClient_ObjectOps(vint targetClientId) + { + if (auto index = objectOps.Keys().IndexOf(targetClientId); index != -1) + { + return objectOps.Values()[index].Obj(); + } + + auto ops = Ptr(new RpcJsonObjectOps(sourceClientId, targetClientId, dispatcher)); + objectOps.Set(targetClientId, ops); + return ops.Obj(); + } + + Ptr RpcJsonDispatcher::Translate(Ptr message, IRpcDispatcher* dispatcher, IRpcLifecycle* lifecycle) + { +#define ERROR_MESSAGE_PREFIX L"vl::rpc_controller::RpcJsonDispatcher::Translate(Ptr, IRpcDispatcher*, IRpcLifecycle*)#" + (void)dispatcher; + CHECK_ERROR(lifecycle, ERROR_MESSAGE_PREFIX L"Lifecycle is required."); + auto request = GetJsonObject(message); + auto rpcMethod = GetJsonString(GetJsonObjectField(request, WString::Unmanaged(L"rpcMethod"))); + CHECK_ERROR(rpcMethod == WString::Unmanaged(L"Request:IRpcDispatcher_DeclareRemoteService"), ERROR_MESSAGE_PREFIX L"Unexpected RPC method."); + auto sourceClientId = ReadSourceClientId(request); + auto ref = GetRpcObjectReferenceFromJson(GetJsonObjectField(request, WString::Unmanaged(L"ref"))); + CHECK_ERROR(ref.clientId == sourceClientId, ERROR_MESSAGE_PREFIX L"Client id does not match the message source."); + + lifecycle->DeclareRemoteService(ref); + return nullptr; +#undef ERROR_MESSAGE_PREFIX + } + +/*********************************************************************** +* RpcJsonLifecycle +***********************************************************************/ + + RpcJsonLifecycle::RpcJsonLifecycle(vint _clientId, RpcJsonDispatcher* _dispatcher) + : RpcLifecycleBase(_clientId) + , dispatcher(_dispatcher) + { +#define ERROR_MESSAGE_PREFIX L"vl::rpc_controller::RpcJsonLifecycle::RpcJsonLifecycle(vint, RpcJsonDispatcher*)#" + CHECK_ERROR(dispatcher, ERROR_MESSAGE_PREFIX L"Dispatcher is required."); +#undef ERROR_MESSAGE_PREFIX + } + + void RpcJsonLifecycle::Register( + Ptr _serializer, + Ptr _objectOps, + Ptr _objectEventOps, + Func _getTypeId, + Func _eventAttacher + ) + { + serializer = _serializer; + getTypeId = _getTypeId; + eventAttacher = _eventAttacher; + listOps = Ptr(new RpcCalleeListOps(this, serializer.Obj())); + listEventOps = Ptr(new RpcCalleeListEventOps(this, serializer.Obj())); + objectOpsForList = Ptr(new RpcCalleeObjectOpsForList(listOps, _objectOps, serializer.Obj())); + objectEventOpsForList = Ptr(new RpcCalleeObjectEventOpsForList(listEventOps, _objectEventOps, serializer.Obj())); + GetController()->Register(objectOpsForList, objectEventOpsForList); + } + + vint RpcJsonLifecycle::DecideTypeId(IDescriptable* obj)const + { + auto result = RpcLifecycleBase::DecideTypeId(obj); + if (result != RpcTypeId_NotFound) return result; + return getTypeId ? getTypeId(obj) : RpcTypeId_NotFound; + } + + IRpcSerializer* RpcJsonLifecycle::GetSerializer() + { + return serializer.Obj(); + } + + IRpcDispatcher* RpcJsonLifecycle::GetDispatcher() + { + return dispatcher; + } + + void RpcJsonLifecycle::AttachLocalObjectEvents(RpcObjectReference ref, IDescriptable* obj) + { + if (eventAttacher) + { + eventAttacher(ref, obj); + } + } + /*********************************************************************** * Request Id ***********************************************************************/ @@ -3064,6 +3212,24 @@ namespace vl #undef ERROR_MESSAGE_PREFIX } + RpcObjectReference RpcLifecycleBase::CreateLocalObject(Ptr obj, RpcObjectReference ref) + { +#define ERROR_MESSAGE_PREFIX L"vl::rpc_controller::RpcLifecycleBase::CreateLocalObject(Ptr, RpcObjectReference)#" + CHECK_ERROR(ref.clientId == clientId, ERROR_MESSAGE_PREFIX L"Ref is not local."); + CHECK_ERROR(!localObjectProperties.Keys().Contains(ref.objectId), ERROR_MESSAGE_PREFIX L"Object ID already registered."); + auto props = Ptr(new RpcLocalObjectProperties); + props->ref = ref; + props->ownedPtr = obj; + localObjectProperties.Set(ref.objectId, props); + TrackLocalObject(ref, obj.Obj()); + if (nextObjectId < ref.objectId) + { + nextObjectId = ref.objectId; + } + return ref; +#undef ERROR_MESSAGE_PREFIX + } + void RpcLifecycleBase::UntrackLocalObject(RpcObjectReference ref, bool clearInternalProperty) { #define ERROR_MESSAGE_PREFIX L"vl::rpc_controller::RpcLifecycleBase::UntrackLocalObject(RpcObjectReference, bool)#" @@ -3233,27 +3399,35 @@ namespace vl auto props = localObjectProperties.Values().Get(localObjectProperties.Count() - 1); RemoveLocalObject(props->ref, true); } + registeredLocalServices.Clear(); + registeredRemoteServices.Clear(); controller.Finalize(); } + void RpcLifecycleBase::Initialize() + { + if (!initialized) + { + GetDispatcher()->Initialize(); + initialized = true; + } + } + vint RpcLifecycleBase::GetClientId() { return clientId; } - IRpcDispatcher* RpcLifecycleBase::GetDispatcher() - { -#define ERROR_MESSAGE_PREFIX L"vl::rpc_controller::RpcLifecycleBase::GetDispatcher()#" - CHECK_ERROR(dispatcher, ERROR_MESSAGE_PREFIX L"No dispatcher registered."); - return dispatcher; -#undef ERROR_MESSAGE_PREFIX - } - RpcControllerDefault* RpcLifecycleBase::GetController() { return &controller; } + const RpcLocalServiceMap& RpcLifecycleBase::GetRegisteredLocalServices() + { + return registeredLocalServices; + } + void RpcLifecycleBase::LocalObjectHold(RpcObjectReference ref, vint remoteClientId) { #define ERROR_MESSAGE_PREFIX L"vl::rpc_controller::RpcLifecycleBase::LocalObjectHold(RpcObjectReference, vint)#" @@ -3284,23 +3458,57 @@ namespace vl #undef ERROR_MESSAGE_PREFIX } - void RpcLifecycleBase::RegisterService(const WString& fullName, Ptr service) + void RpcLifecycleBase::RegisterLocalService(vint typeId, Ptr service) { - auto index = idMap.Keys().IndexOf(fullName); - if (index == -1) +#define ERROR_MESSAGE_PREFIX L"vl::rpc_controller::RpcLifecycleBase::RegisterLocalService(vint, Ptr)#" + CHECK_ERROR(service, ERROR_MESSAGE_PREFIX L"Service is required."); + if (initialized) { - CHECK_FAIL(L"Unknown RPC type id."); + throw Exception(ERROR_MESSAGE_PREFIX L"RegisterLocalService cannot be called after Initialize."); } - controller.GetObjectOps()->RegisterService(idMap.Values()[index], service); + if (registeredLocalServices.Keys().Contains(typeId)) + { + throw Exception(ERROR_MESSAGE_PREFIX L"Service is already registered."); + } + + auto ref = CreateLocalObject(service, RpcObjectReference{ clientId, typeId, typeId }); + LocalObjectHold(ref, clientId); + registeredLocalServices.Set(typeId, service); + GetDispatcher()->DeclareLocalService(ref); +#undef ERROR_MESSAGE_PREFIX } - Ptr RpcLifecycleBase::RequestService(const WString& fullName) + void RpcLifecycleBase::DeclareRemoteService(RpcObjectReference ref) { - if (idMap.Keys().Contains(fullName)) + registeredRemoteServices.Set(ref.typeId, ref); + } + + vint RpcLifecycleBase::GetTypeIdFromName(WString typeName) + { + auto typeIndex = idMap.Keys().IndexOf(typeName); + if (typeIndex == -1) { - auto typeId = idMap.Get(fullName); - auto ref = GetDispatcher()->RequestService(typeId); - return RefToPtr(ref); + return RpcTypeId_NotFound; + } + return idMap.Values()[typeIndex]; + } + + Ptr RpcLifecycleBase::RequestService(WString typeName) + { + auto typeId = GetTypeIdFromName(typeName); + if (typeId == RpcTypeId_NotFound) + { + return nullptr; + } + + if (auto index = registeredLocalServices.Keys().IndexOf(typeId); index != -1) + { + return registeredLocalServices.Values()[index]; + } + + if (auto index = registeredRemoteServices.Keys().IndexOf(typeId); index != -1) + { + return RefToPtr(registeredRemoteServices.Values()[index]); } return nullptr; @@ -3376,15 +3584,7 @@ namespace vl auto typeId = DecideTypeId(obj.Obj()); CHECK_ERROR(typeId != RpcTypeId_NotFound, ERROR_MESSAGE_PREFIX L"DecideTypeId returned RpcTypeId_NotFound (unknown type)."); - auto ref = RpcObjectReference{ clientId, ++nextObjectId, typeId }; - CHECK_ERROR(!localObjectProperties.Keys().Contains(ref.objectId), ERROR_MESSAGE_PREFIX L"Object ID already registered."); - auto props = Ptr(new RpcLocalObjectProperties); - props->ref = ref; - props->ownedPtr = obj; - localObjectProperties.Set(ref.objectId, props); - - TrackLocalObject(ref, obj.Obj()); - return ref; + return CreateLocalObject(obj, RpcObjectReference{ clientId, ++nextObjectId, typeId }); #undef ERROR_MESSAGE_PREFIX } } @@ -4298,11 +4498,6 @@ namespace vl objectOps->ObjectHold(ref, remoteClientId, hold); } - void RpcCalleeObjectOpsForList::RegisterService(vint typeId, Ptr service) - { - objectOps->RegisterService(typeId, service); - } - /*********************************************************************** * RpcCalleeObjectEventOpsForList ***********************************************************************/ diff --git a/Import/VlppWorkflowLibrary.h b/Import/VlppWorkflowLibrary.h index 6272ac8b..df6464f0 100644 --- a/Import/VlppWorkflowLibrary.h +++ b/Import/VlppWorkflowLibrary.h @@ -961,6 +961,7 @@ namespace vl }; using RpcEventExceptionMap = Ptr; + using RpcLocalServiceMap = collections::Dictionary>; extern void MergeRpcEventExceptionMap(RpcEventExceptionMap target, RpcEventExceptionMap source); @@ -1060,7 +1061,6 @@ namespace vl virtual reflection::description::Value InvokeMethod(RpcObjectReference ref, vint methodId, Ptr arguments) = 0; virtual void EndInvokeMethod(vint slot) = 0; virtual void ObjectHold(RpcObjectReference ref, vint remoteClientId, bool hold) = 0; - virtual void RegisterService(vint typeId, Ptr service) = 0; }; class IRpcListEventOps @@ -1088,9 +1088,7 @@ namespace vl , public reflection::Description { public: - virtual IRpcListOps* GetListOps() = 0; virtual IRpcObjectOps* GetObjectOps() = 0; - virtual IRpcListEventOps* GetListEventOps() = 0; virtual IRpcObjectEventOps* GetObjectEventOps() = 0; }; @@ -1100,9 +1098,8 @@ namespace vl { public: virtual void Finalize() = 0; - virtual bool IsRegisteredService(RpcObjectReference ref) = 0; - virtual void RegisterService(vint typeId, RpcObjectReference ref) = 0; - virtual RpcObjectReference RequestService(vint typeId) = 0; + virtual void Initialize() = 0; + virtual void DeclareLocalService(RpcObjectReference ref) = 0; virtual IRpcObjectEventOps* BroadcastFromClient_ObjectEventOps(vint selfClientId) = 0; virtual IRpcObjectOps* SendToClient_ObjectOps(vint targetClientId) = 0; @@ -1120,22 +1117,86 @@ namespace vl virtual bool GetItemChangedSuppressedFlag(RpcObjectReference ref) = 0; }; + /* + * [Configuration] + * + * RunRpcTestCase_JsonRequest configures one RpcJsonDispatcher and one RpcJsonLifecycle for each client. + * messageDispatcher = shared IRpcJsonMessageDispatcher + * dispatcher = RpcJsonDispatcher(clientId, messageDispatcher) + * lifecycle = RpcJsonLifecycle(clientId, dispatcher) + * serializer = rpcops_IRpcSerializer() + * getTypeId = [rpcwrapper_GetTypeId(BoxValue(obj))] + * lifecycle->Register(serializer, rpcops_IRpcObjectOpsJson(lifecycle), rpcops_IRpcObjectEventOpsJson(lifecycle), getTypeId, eventAttacher) + * + * Triggering RpcLifecycleBase::AttachLocalObjectEvents + * call rpclistener_Attach(ref.typeId, this, ref, obj, (cached)rpcops_IOps_CreateJson(this)) + * + * [Call graph for JSON based RPC] + * + * Calling Method of Remote Object: + * -> IMyInterface::Method (generated Workflow code) + * -> rpcops_IOps_::InvokeMethod_IMyInterface_Method (generated Workflow code) + * { + * -> IRpcLifecycle->GetDispatcher()->SendToClient_ObjectOps()->InvokeMethod + * -> RpcJsonObjectOps::InvokeMethod + * -> IRpcJsonMessageDispatcher::OnJsonRequest + * ---- NETWORK PROTOCOL (request) ---- + * -> RpcJsonObjectOps::Translate + * -> IRpcLifecycle->GetController()->GetObjectOps()->InvokeMethod + * -> RpcCalleeObjectOpsForList::InvokeMethod + * -> rpcops_IRpcObjectOpsJson()->InvokeMethod (generated Workflow code) + * -> IMyInterface::Method (actual) + * ---- NETWORK PROTOCOL (response) ---- + * } + * { optional EndInvokeMethod when @rpc:Byval on return value } + * + * Triggering Event of Remote Object (events are automatically hooked when creating a wrapper for a remote object): + * -> IMyInterface::SomethingHappened + * -> rpcops_IOps_::InvokeMethod_IMyInterface_SomethingHappened (generated Workflow code) + * { + * -> IRpcLifecycle->GetDispatcher()->BroadcastFromClient_ObjectEventOps()->InvokeEvent + * -> RpcJsonEventObjectOps::InvokeEvent + * -> IRpcJsonMessageDispatcher::OnJsonRequest + * ---- NETWORK PROTOCOL (broadcast) ---- + * -> RpcJsonEventObjectOps::Translate + * -> IRpcLifecycle->GetController()->GetObjectOps()->InvokeEvent + * -> RpcCalleeObjectEventOpsForList::InvokeEvent + * -> rpcops_IRpcObjectEventOpsJson()->InvokeEvent (generated Workflow code) + * -> IMyInterface::SomethingHappened + * ---- NETWORK PROTOCOL (response) ---- + * } + * + * Triggering Event of Local Object (when a local object is tracked, RpcLifecycleBase::AttachLocalObjectEvents will be called) + * The same to remote object. + * + * Registering Service: + * -> IRpcLifecycle->RegisterLocalService + * { + * -> IRpcLifecycle->GetDispatcher()->DeclareLocalService(ref) + * ---- NETWORK PROTOCOL (broadcast) ---- + * -> IRpcLifecycle::DeclareRemoteService(ref) + * } + */ class IRpcLifecycle : public virtual reflection::IDescriptable , public reflection::Description { public: virtual void Finalize() = 0; + virtual void Initialize() = 0; virtual vint GetClientId() = 0; virtual IRpcDispatcher* GetDispatcher() = 0; virtual IRpcController* GetController() = 0; virtual IRpcSerializer* GetSerializer() = 0; + virtual const RpcLocalServiceMap& GetRegisteredLocalServices() = 0; virtual Ptr RefToPtr(RpcObjectReference ref) = 0; virtual RpcObjectReference PtrToRef(Ptr obj) = 0; virtual void LocalObjectHold(RpcObjectReference ref, vint remoteClientId) = 0; virtual void LocalObjectUnhold(RpcObjectReference ref, vint remoteClientId) = 0; - virtual void RegisterService(const WString& fullName, Ptr service) = 0; - virtual Ptr RequestService(const WString& fullName) = 0; + virtual void RegisterLocalService(vint typeId, Ptr service) = 0; + virtual void DeclareRemoteService(RpcObjectReference ref) = 0; + virtual vint GetTypeIdFromName(WString typeName) = 0; + virtual Ptr RequestService(WString typeName) = 0; }; class IRpcWrapperBase @@ -1196,8 +1257,6 @@ namespace vl protected: Ptr objectCallback; Ptr eventCallback; - Ptr listCallback; - Ptr listEventCallback; collections::Dictionary eventSuppressedFlags; collections::Dictionary itemChangedSuppressedFlags; @@ -1212,13 +1271,11 @@ namespace vl RpcControllerDefault(); ~RpcControllerDefault(); - void Register(Ptr objectCallback, Ptr eventCallback, Ptr listCallback, Ptr listEventCallback); + void Register(Ptr objectCallback, Ptr eventCallback); // IRpcController - IRpcListOps* GetListOps()override; IRpcObjectOps* GetObjectOps()override; - IRpcListEventOps* GetListEventOps()override; IRpcObjectEventOps* GetObjectEventOps()override; void Finalize()override; @@ -1233,85 +1290,6 @@ namespace vl #endif -/*********************************************************************** -.\RPC\WFLIBRARYRPCJSON.H -***********************************************************************/ -/*********************************************************************** -Vczh Library++ 3.0 -Developer: Zihan Chen(vczh) -Framework::RPC - -JSON Helpers: -***********************************************************************/ - -#ifndef VCZH_WORKFLOW_LIBRARY_RPC_JSON -#define VCZH_WORKFLOW_LIBRARY_RPC_JSON - - -namespace vl -{ - namespace rpc_controller - { - using RpcJsonSerializeCallback = Func(const reflection::description::Value&)>; - using RpcJsonDeserializeCallback = Func)>; - - extern Ptr JsonSerializePredefinedTypes(const reflection::description::Value& value, const RpcJsonSerializeCallback& rpcjson_Serialize); - extern reflection::description::Value JsonDeserializePredefinedTypes(const reflection::description::Value& value, const RpcJsonDeserializeCallback& rpcjson_Deserialize); - - class IRpcJsonMessageDispatcher - : public virtual reflection::IDescriptable - , public reflection::Description - { - public: - virtual vint AllocateRequestId() = 0; - virtual Ptr OnJsonRequest(Ptr message) = 0; - }; - - class RpcJsonObjectOps : public Object, public IRpcObjectOps - { - private: - vint sourceClientId = RpcClientId_Invalid; - vint targetClientId = RpcClientId_Invalid; - IRpcJsonMessageDispatcher* dispatcher = nullptr; - IRpcLifecycle* lifecycle = nullptr; - - public: - RpcJsonObjectOps(IRpcJsonMessageDispatcher* _dispatcher); - RpcJsonObjectOps(vint _sourceClientId, vint _targetClientId, IRpcJsonMessageDispatcher* _dispatcher, IRpcLifecycle* _lifecycle = nullptr); - ~RpcJsonObjectOps(); - - reflection::description::Value InvokeMethod(RpcObjectReference ref, vint methodId, Ptr arguments)override; - void EndInvokeMethod(vint slot)override; - void ObjectHold(RpcObjectReference ref, vint remoteClientId, bool hold)override; - void RegisterService(vint typeId, Ptr service)override; - - static Ptr Translate(Ptr message, IRpcObjectOps* ops, IRpcLifecycle* lifecycle = nullptr); - }; - - class RpcJsonObjectEventOps : public Object, public IRpcObjectEventOps - { - private: - vint sourceClientId = RpcClientId_Invalid; - IRpcJsonMessageDispatcher* dispatcher = nullptr; - - public: - RpcJsonObjectEventOps(IRpcJsonMessageDispatcher* _dispatcher); - RpcJsonObjectEventOps(vint _sourceClientId, IRpcJsonMessageDispatcher* _dispatcher); - ~RpcJsonObjectEventOps(); - - reflection::description::Value InvokeEvent(RpcObjectReference ref, vint eventId, Ptr arguments)override; - - static Ptr Translate(Ptr message, IRpcObjectEventOps* ops); - }; - - extern vint ReadRequestId(Ptr message); - extern void WriteRequestId(Ptr message, vint requestId); - } -} - -#endif - - /*********************************************************************** .\RPC\WFLIBRARYRPCLIFECYCLE.H ***********************************************************************/ @@ -1392,7 +1370,10 @@ namespace vl RpcControllerDefault controller; vint clientId = RpcClientId_Invalid; vint nextObjectId = RpcObjectId_Invalid; + bool initialized = false; LocalProperties localObjectProperties; + RpcLocalServiceMap registeredLocalServices; + collections::Dictionary registeredRemoteServices; static WString InternalProperty_LocalObjectTracker; static WString InternalProperty_WrapperTracker; UniversalWrapperFactory universalWrapperFactory; @@ -1403,13 +1384,13 @@ namespace vl bool TryGetTrackedWrapperRef(reflection::DescriptableObject* obj, RpcObjectReference& ref)const; IRpcWrapperBase* GetTrackedWrapper(RpcObjectReference ref)const; void TrackLocalObject(RpcObjectReference ref, reflection::IDescriptable* obj); + RpcObjectReference CreateLocalObject(Ptr obj, RpcObjectReference ref); void UntrackLocalObject(RpcObjectReference ref, bool clearInternalProperty); void RemoveLocalObject(RpcObjectReference ref, bool clearInternalProperty); bool IsTracked(vint objectId)const; Ptr CreateCallerProxy(RpcObjectReference ref, IRpcSerializer* serializer); void DisconnectWrappersForFinalize(); protected: - IRpcDispatcher* dispatcher = nullptr; collections::Dictionary idMap; Ptr serializer; @@ -1427,13 +1408,16 @@ namespace vl // IRpcLifecycle void Finalize()override; + void Initialize()override; vint GetClientId()override; - IRpcDispatcher* GetDispatcher()override; RpcControllerDefault* GetController()override; + const RpcLocalServiceMap& GetRegisteredLocalServices()override; void LocalObjectHold(RpcObjectReference ref, vint remoteClientId)override; void LocalObjectUnhold(RpcObjectReference ref, vint remoteClientId)override; - void RegisterService(const WString& fullName, Ptr service)override; - Ptr RequestService(const WString& fullName)override; + void RegisterLocalService(vint typeId, Ptr service)override; + void DeclareRemoteService(RpcObjectReference ref)override; + vint GetTypeIdFromName(WString typeName)override; + Ptr RequestService(WString typeName)override; Ptr RefToPtr(RpcObjectReference ref)override; Ptr RefToPtr(RpcObjectReference ref, IRpcSerializer* serializer); RpcObjectReference PtrToRef(Ptr obj)override; @@ -1444,6 +1428,153 @@ namespace vl #endif +/*********************************************************************** +.\RPC\WFLIBRARYRPCJSON.H +***********************************************************************/ +/*********************************************************************** +Vczh Library++ 3.0 +Developer: Zihan Chen(vczh) +Framework::RPC + +JSON Helpers: +***********************************************************************/ + +#ifndef VCZH_WORKFLOW_LIBRARY_RPC_JSON +#define VCZH_WORKFLOW_LIBRARY_RPC_JSON + + +namespace vl +{ + namespace rpc_controller + { + class RpcCalleeListOps; + class RpcCalleeListEventOps; + class RpcCalleeObjectOpsForList; + class RpcCalleeObjectEventOpsForList; + + using RpcJsonSerializeCallback = Func(const reflection::description::Value&)>; + using RpcJsonDeserializeCallback = Func)>; + + extern Ptr JsonSerializePredefinedTypes(const reflection::description::Value& value, const RpcJsonSerializeCallback& rpcjson_Serialize); + extern reflection::description::Value JsonDeserializePredefinedTypes(const reflection::description::Value& value, const RpcJsonDeserializeCallback& rpcjson_Deserialize); + + class IRpcJsonMessageDispatcher + : public virtual reflection::IDescriptable + , public reflection::Description + { + public: + enum class RequestType + { + Direct, + Broadcast, + BroadcastAndDrop, + }; + + virtual vint AllocateRequestId() = 0; + virtual Ptr OnJsonRequest(Ptr message, RequestType requestType) = 0; + + static Ptr DefaultTranslate( + Ptr message, + RequestType requestType, + IRpcObjectOps* objectOps, + IRpcObjectEventOps* objectEventOps, + IRpcDispatcher* dispatcher, + IRpcLifecycle* lifecycle + ); + }; + + class RpcJsonObjectOps : public Object, public IRpcObjectOps + { + private: + vint sourceClientId = RpcClientId_Invalid; + vint targetClientId = RpcClientId_Invalid; + IRpcJsonMessageDispatcher* dispatcher = nullptr; + + public: + RpcJsonObjectOps(IRpcJsonMessageDispatcher* _dispatcher); + RpcJsonObjectOps(vint _sourceClientId, vint _targetClientId, IRpcJsonMessageDispatcher* _dispatcher); + ~RpcJsonObjectOps(); + + reflection::description::Value InvokeMethod(RpcObjectReference ref, vint methodId, Ptr arguments)override; + void EndInvokeMethod(vint slot)override; + void ObjectHold(RpcObjectReference ref, vint remoteClientId, bool hold)override; + + static Ptr Translate(Ptr message, IRpcObjectOps* ops, IRpcLifecycle* lifecycle = nullptr); + }; + + class RpcJsonObjectEventOps : public Object, public IRpcObjectEventOps + { + private: + vint sourceClientId = RpcClientId_Invalid; + IRpcJsonMessageDispatcher* dispatcher = nullptr; + + public: + RpcJsonObjectEventOps(IRpcJsonMessageDispatcher* _dispatcher); + RpcJsonObjectEventOps(vint _sourceClientId, IRpcJsonMessageDispatcher* _dispatcher); + ~RpcJsonObjectEventOps(); + + reflection::description::Value InvokeEvent(RpcObjectReference ref, vint eventId, Ptr arguments)override; + + static Ptr Translate(Ptr message, IRpcObjectEventOps* ops, IRpcLifecycle* lifecycle = nullptr); + }; + + class RpcJsonDispatcher : public Object, public IRpcDispatcher + { + private: + vint sourceClientId = RpcClientId_Invalid; + IRpcJsonMessageDispatcher* dispatcher = nullptr; + Ptr objectEventOps; + collections::Dictionary> objectOps; + + public: + RpcJsonDispatcher(vint _sourceClientId, IRpcJsonMessageDispatcher* _dispatcher); + + void Finalize()override; + void Initialize()override; + void DeclareLocalService(RpcObjectReference ref)override; + IRpcObjectEventOps* BroadcastFromClient_ObjectEventOps(vint selfClientId)override; + IRpcObjectOps* SendToClient_ObjectOps(vint targetClientId)override; + + static Ptr Translate(Ptr message, IRpcDispatcher* dispatcher, IRpcLifecycle* lifecycle); + }; + + class RpcJsonLifecycle : public RpcLifecycleBase + { + private: + RpcJsonDispatcher* dispatcher = nullptr; + Func getTypeId; + Func eventAttacher; + Ptr listOps; + Ptr listEventOps; + Ptr objectOpsForList; + Ptr objectEventOpsForList; + + protected: + vint DecideTypeId(reflection::IDescriptable* obj)const override; + void AttachLocalObjectEvents(RpcObjectReference ref, reflection::IDescriptable* obj)override; + + public: + RpcJsonLifecycle(vint _clientId, RpcJsonDispatcher* _dispatcher); + + void Register( + Ptr _serializer, + Ptr _objectOps, + Ptr _objectEventOps, + Func _getTypeId, + Func _eventAttacher + ); + IRpcSerializer* GetSerializer()override; + IRpcDispatcher* GetDispatcher()override; + }; + + extern vint ReadRequestId(Ptr message); + extern void WriteRequestId(Ptr message, vint requestId); + } +} + +#endif + + /*********************************************************************** .\RPC\WFLIBRARYRPCWRAPPERS.H ***********************************************************************/ @@ -1663,7 +1794,6 @@ namespace vl reflection::description::Value InvokeMethod(RpcObjectReference ref, vint methodId, Ptr arguments)override; void EndInvokeMethod(vint slot)override; void ObjectHold(RpcObjectReference ref, vint remoteClientId, bool hold)override; - void RegisterService(vint typeId, Ptr service)override; }; class RpcCalleeObjectEventOpsForList : public Object, public IRpcObjectEventOps @@ -1796,6 +1926,7 @@ Predefined Types F(vl::rpc_controller::RpcException)\ F(vl::rpc_controller::RpcByvalReturnValue)\ F(vl::rpc_controller::IRpcSerializer)\ + F(vl::rpc_controller::IRpcJsonMessageDispatcher::RequestType)\ F(vl::rpc_controller::IRpcJsonMessageDispatcher)\ F(vl::rpc_controller::IRpcListOps)\ F(vl::rpc_controller::IRpcListEventOps)\ @@ -1846,9 +1977,9 @@ Interface Implementation Proxy (Implement) INVOKEGET_INTERFACE_PROXY_NOPARAMS(AllocateRequestId); } - vl::Ptr OnJsonRequest(vl::Ptr message)override + vl::Ptr OnJsonRequest(vl::Ptr message, vl::rpc_controller::IRpcJsonMessageDispatcher::RequestType requestType)override { - INVOKEGET_INTERFACE_PROXY(OnJsonRequest, message); + INVOKEGET_INTERFACE_PROXY(OnJsonRequest, message, requestType); } END_INTERFACE_PROXY(vl::rpc_controller::IRpcJsonMessageDispatcher) @@ -1975,10 +2106,6 @@ Interface Implementation Proxy (Implement) INVOKE_INTERFACE_PROXY(ObjectHold, ref, remoteClientId, hold); } - void RegisterService(vl::vint typeId, vl::Ptr service)override - { - INVOKE_INTERFACE_PROXY(RegisterService, typeId, service); - } END_INTERFACE_PROXY(vl::rpc_controller::IRpcObjectOps) BEGIN_INTERFACE_PROXY_NOPARENT_SHAREDPTR(vl::rpc_controller::IRpcListEventOps) diff --git a/Tools/Reflection32.bin b/Tools/Reflection32.bin index e7c21dd1..f75c1ef3 100644 Binary files a/Tools/Reflection32.bin and b/Tools/Reflection32.bin differ diff --git a/Tools/Reflection64.bin b/Tools/Reflection64.bin index 434de469..2ae955ae 100644 Binary files a/Tools/Reflection64.bin and b/Tools/Reflection64.bin differ