1#include "rpc_connection.h"
  2#include "serialization.h"
  3
  4#include <atomic>
  5
  6static const int RpcVersion = 1;
  7static RpcConnection Instance;
  8
  9/*static*/ RpcConnection* RpcConnection::Create(const char* applicationId)
 10{
 11    Instance.connection = BaseConnection::Create();
 12    StringCopy(Instance.appId, applicationId);
 13    return &Instance;
 14}
 15
 16/*static*/ void RpcConnection::Destroy(RpcConnection*& c)
 17{
 18    c->Close();
 19    BaseConnection::Destroy(c->connection);
 20    c = nullptr;
 21}
 22
 23void RpcConnection::Open()
 24{
 25    if (state == State::Connected) {
 26        return;
 27    }
 28
 29    if (state == State::Disconnected && !connection->Open()) {
 30        return;
 31    }
 32
 33    if (state == State::SentHandshake) {
 34        JsonDocument message;
 35        if (Read(message)) {
 36            auto cmd = GetStrMember(&message, "cmd");
 37            auto evt = GetStrMember(&message, "evt");
 38            if (cmd && evt && !strcmp(cmd, "DISPATCH") && !strcmp(evt, "READY")) {
 39                state = State::Connected;
 40                if (onConnect) {
 41                    onConnect(message);
 42                }
 43            }
 44        }
 45    }
 46    else {
 47        sendFrame.opcode = Opcode::Handshake;
 48        sendFrame.length = (uint32_t)JsonWriteHandshakeObj(
 49          sendFrame.message, sizeof(sendFrame.message), RpcVersion, appId);
 50
 51        if (connection->Write(&sendFrame, sizeof(MessageFrameHeader) + sendFrame.length)) {
 52            state = State::SentHandshake;
 53        }
 54        else {
 55            Close();
 56        }
 57    }
 58}
 59
 60void RpcConnection::Close()
 61{
 62    if (onDisconnect && (state == State::Connected || state == State::SentHandshake)) {
 63        onDisconnect(lastErrorCode, lastErrorMessage);
 64    }
 65    connection->Close();
 66    state = State::Disconnected;
 67}
 68
 69bool RpcConnection::Write(const void* data, size_t length)
 70{
 71    sendFrame.opcode = Opcode::Frame;
 72    memcpy(sendFrame.message, data, length);
 73    sendFrame.length = (uint32_t)length;
 74    if (!connection->Write(&sendFrame, sizeof(MessageFrameHeader) + length)) {
 75        Close();
 76        return false;
 77    }
 78    return true;
 79}
 80
 81bool RpcConnection::Read(JsonDocument& message)
 82{
 83    if (state != State::Connected && state != State::SentHandshake) {
 84        return false;
 85    }
 86    MessageFrame readFrame;
 87    for (;;) {
 88        bool didRead = connection->Read(&readFrame, sizeof(MessageFrameHeader));
 89        if (!didRead) {
 90            if (!connection->isOpen) {
 91                lastErrorCode = (int)ErrorCode::PipeClosed;
 92                StringCopy(lastErrorMessage, "Pipe closed");
 93                Close();
 94            }
 95            return false;
 96        }
 97
 98        if (readFrame.length > 0) {
 99            didRead = connection->Read(readFrame.message, readFrame.length);
100            if (!didRead) {
101                lastErrorCode = (int)ErrorCode::ReadCorrupt;
102                StringCopy(lastErrorMessage, "Partial data in frame");
103                Close();
104                return false;
105            }
106            readFrame.message[readFrame.length] = 0;
107        }
108
109        switch (readFrame.opcode) {
110        case Opcode::Close: {
111            message.ParseInsitu(readFrame.message);
112            lastErrorCode = GetIntMember(&message, "code");
113            StringCopy(lastErrorMessage, GetStrMember(&message, "message", ""));
114            Close();
115            return false;
116        }
117        case Opcode::Frame:
118            message.ParseInsitu(readFrame.message);
119            return true;
120        case Opcode::Ping:
121            readFrame.opcode = Opcode::Pong;
122            if (!connection->Write(&readFrame, sizeof(MessageFrameHeader) + readFrame.length)) {
123                Close();
124            }
125            break;
126        case Opcode::Pong:
127            break;
128        case Opcode::Handshake:
129        default:
130            // something bad happened
131            lastErrorCode = (int)ErrorCode::ReadCorrupt;
132            StringCopy(lastErrorMessage, "Bad ipc frame");
133            Close();
134            return false;
135        }
136    }
137}