Switched from Websockets to POST requests and SSE for streams

src/generators/cpp_s.rs

@@ -1,6 +1,21 @@
 use itertools::Itertools;
 use crate::data::RPC;
 
+pub const JSON_INNER_IMPLS: &[(&str, &str)] = &[
+    ("std::string", "String"),
+    ("std::int8_t", "Int"),
+    ("std::int16_t", "Int"),
+    ("std::int32_t", "Int"),
+    ("std::int64_t", "Int64"),
+    ("std::uint8_t", "Uint"),
+    ("std::uint16_t", "Uint"),
+    ("std::uint32_t", "Uint"),
+    ("std::uint64_t", "Uint64"),
+    ("bool", "Bool"),
+    ("std::float_t", "Double"),
+    ("std::double_t", "Double")
+];
+
 pub fn ty_to_str(ty: &crate::data::Types) -> String {
     use crate::data::Types;
     match &ty {
@@ -22,11 +37,10 @@ pub fn ty_to_str(ty: &crate::data::Types) -> String {
     }
 }
 
-
 pub fn method_args(method: &crate::data::MethodTy) -> String {
     method.args.iter()
         .map(|arg| format!("{} &&{}", ty_to_str(&arg.ty), arg.name))
-        .chain(method.ret_stream.then(|| format!("std::shared_ptr<MRPCStream<{}>>&&", ty_to_str(method.ret.as_ref().unwrap()))))
+        .chain(method.ret_stream.then(|| format!("MRPCStream<{}>&&", ty_to_str(method.ret.as_ref().unwrap()))))
         .join(", ")
 }
 
@@ -45,6 +59,49 @@ pub fn call_args(method: &crate::data::MethodTy) -> String {
         .join(", ")
 }
 
+pub fn json_write(ty: &crate::data::FieldTy) -> String {
+    use crate::data::Types;
+    match &ty.ty {
+        Types::String => format!("__w.String({});", ty.name),
+        Types::Bool => format!("__w.Bool({});", ty.name),
+        Types::F32 | Types::F64 => format!("__w.Double({});", ty.name),
+        Types::I8 | Types::I16 | Types::I32 | Types::I64 => format!("__w.Int64({});", ty.name),
+        Types::U8 | Types::U16 | Types::U32 | Types::U64 => format!("__w.Uint64({});", ty.name),
+        Types::Named(_) => format!("{} >> __w;", ty.name),
+        Types::Optional(inner) => {
+            let inner = crate::data::FieldTy { name: format!("({}.value())", ty.name), ty: (**inner).clone() };
+            let inner = json_write(&inner);
+            format!(
+"if ({}.has_value()) {{
+        {}
+    }} else __w.Null();", ty.name, inner)
+        },
+        Types::Array(inner) => {
+            let inner_var_name = format!("__{}__entry", ty.name);
+            let inner = crate::data::FieldTy { name: inner_var_name.clone(), ty: (**inner).clone() };
+            let inner = json_write(&inner);
+            format!(
+"__w.StartArray();
+    for (const auto &{} : {}) {{
+        {}
+    }}
+    __w.EndArray();", inner_var_name, ty.name, inner)
+        }
+    }
+}
+
+pub fn streams_required(rpc: &RPC) -> Vec<String> {
+    let mut streams = std::collections::HashSet::new();
+    for s in &rpc.services {
+        for m in &s.methods {
+            if m.ret_stream {
+                streams.insert(ty_to_str(m.ret.as_ref().unwrap()));
+            }
+        }
+    }
+    streams.into_iter().collect()
+}
+
 pub fn gen(file_base_name: &std::path::PathBuf, rpc: &RPC) {
     let header_name = file_base_name.with_extension("h");
     let header_name = header_name.file_name().unwrap().to_str().unwrap();

src/generators/ts_c.rs

@@ -19,7 +19,7 @@ pub fn ty_to_str(ty: &crate::data::Types) -> String {
 pub fn method_args(method: &crate::data::MethodTy) -> String {
     method.args.iter()
         .map(|arg| format!("{}: {}", arg.name, ty_to_str(&arg.ty)))
-        .chain(method.ret_stream.then(|| format!("__cbk: (v: {}) => void", ty_to_str(method.ret.as_ref().unwrap()))))
+        .chain(method.ret_stream.then(|| format!("__cbk: (v: {}|null) => void", ty_to_str(method.ret.as_ref().unwrap()))))
         .join(", ")
 }
 

templates/cpp_server.rs.cpp

@@ -1,96 +1,96 @@
 @use crate::data::RPC;
 @use crate::generators::cpp_s::*;
+@use super::cpp_server_json_cpp;
 
 @(header_name: &str, rpc: &RPC)
 #include "@header_name"
-using json = nlohmann::json;
+#include <corvusoft/restbed/session.hpp>
+#include <corvusoft/restbed/resource.hpp>
+#include <corvusoft/restbed/request.hpp>
+
+@:cpp_server_json_cpp(rpc)
 
-namespace nlohmann @{
 template<typename T>
-    struct adl_serializer<std::optional<T>> @{
+void send_msg(const std::shared_ptr<restbed::Session> &c, const T &v) @{
-        static void to_json(json &j, const std::optional<T> &v) @{
+    if (c->is_closed())
-            if (v.has_value())
+        return;
-                j = v.value();
+    rapidjson::StringBuffer s;
-            else
+    mrpc::MRPCJWriter writer@{s@};
-                j = nullptr;
+    v >> writer;
+    const auto body_ptr = s.GetString();
+    const auto body = restbed::Bytes@{body_ptr, body_ptr+s.GetLength()@};
+    c->yield(
+        200,
+        body,
+        std::multimap<std::string, std::string>@{
+            @{"Content-Type", "application/json"@},
+            @{"Content-Length", std::to_string(body.size())@}
+        @}
+    );
 @}
 
-        static void from_json(const json &j, std::optional<T> &v) @{
+template<typename T>
-            if (j.is_null())
+void send_sse_msg(const std::shared_ptr<restbed::Session> &c, const T &v) @{
-                v.reset();
+    if (c->is_closed())
-            else
+        return;
-                v = j.get<T>();
+    rapidjson::StringBuffer s;
-        @}
+    std::memcpy(s.Push(5), "data:", 5);
-    @};
+    mrpc::MRPCJWriter writer@{s@};
+    v >> writer;
+    std::memcpy(s.Push(2), "\n\n", 2);
+    const auto body_ptr = s.GetString();
+    const auto body = restbed::Bytes@{body_ptr, body_ptr+s.GetLength()@};
+    c->yield(body);
 @}
 
-namespace mrpc @{
+mrpc::MRPCStreamImpl::MRPCStreamImpl(const std::shared_ptr<restbed::Session> &conn) : conn(conn) @{
-@for s in &rpc.structs {
+    conn->yield(
-void to_json(nlohmann::json &j, const @s.name &v) @{
+        200,
-@for f in &s.fields {    j["@f.name"] = v.@f.name;
+        std::multimap<std::string, std::string>@{
-}
+		    @{"Cache-Control", "no-cache"@},
+		    @{"Content-Type", "text/event-stream"@}
         @}
-void from_json(const nlohmann::json &j, @s.name &v) @{
+    );
-@for f in &s.fields {    j.at("@f.name").get_to(v.@f.name);
-}
 @}
+
+void mrpc::MRPCStreamImpl::close() const noexcept @{ conn->close("data:null\n\n"); @}
+bool mrpc::MRPCStreamImpl::is_open() const noexcept @{ return conn->is_open(); @}
+@for s in streams_required(rpc) {template<> void MRPCStream<mrpc::@s>::send(const mrpc::@s &v) const noexcept @{ send_sse_msg(conn, v); @}
 }
 
-template<class T>
+mrpc::MRPCServer::MRPCServer(std::shared_ptr<restbed::Resource> &r) @{
-void send_msg(crow::websocket::connection &c, uint64_t id, const T &v) @{
+    r->set_method_handler("POST", [this](const std::shared_ptr<restbed::Session>& s) @{
-    c.send_text(json@{@{"id", id@},@{"data", v@}@}.dump());
+        const auto req = s->get_request();
-@}
+        const auto body_len = req->get_header("Content-Length", 0);
-
+        s->fetch(body_len, [this](const std::shared_ptr<restbed::Session> &s, auto &&body) @{
-void mrpc::MRPCStreamImpl::close() noexcept @{
+            try @{ msg_handler(s, body); @}
-    if (conn != nullptr) @{
+            catch (const std::exception &_) @{ s->close(400); @}
-        send_msg(*conn, id, nullptr);
+        @});
-        conn = nullptr;
-    @}
-@}
-void mrpc::MRPCStreamImpl::abort() noexcept @{ conn = nullptr; @}
-bool mrpc::MRPCStreamImpl::is_open() noexcept @{ return conn != nullptr; @}
-
-void mrpc::MRPCServer::install(crow::SimpleApp &app, std::string &&route) @{
-    app.route_dynamic(std::move(route))
-        .websocket()
-        .onclose([&](crow::websocket::connection &c, const std::string&)@{
-            std::lock_guard guard@{__streams_mutex@};
-            auto range = __streams.equal_range(&c);
-            for (auto it = range.first; it != range.second; ++it)
-                it->second->abort();
-            __streams.erase(&c);
-        @})
-        .onmessage([this](auto &&a, auto &&b, auto &&c) @{
-            try @{ msg_handler(a, b, c); @}
-            catch (const std::exception &_) @{@}
     @});
 @}
-void mrpc::MRPCServer::msg_handler(crow::websocket::connection &__c, const std::string &__msg, bool) @{
+
-    json __j = json::parse(__msg);
+void mrpc::MRPCServer::msg_handler(const std::shared_ptr<restbed::Session> __c, const restbed::Bytes &__msg) @{
-    std::uint64_t __id = __j.at("id");
+    rapidjson::Document __j;
-    std::string __service = __j.at("service"), __method = __j.at("method");
+    __j.Parse((const char*)__msg.data(), __msg.size());
-    try @{
+    if (__j.HasParseError())
-        json __data = __j.at("data");
+        throw std::exception@{@};
-@for (si, s) in rpc.services.iter().enumerate() {
+    std::string __service, __method;
-        @if si > 0 {else }if (__service == "@s.name") @{
+    json_get(__j, "service", __service); json_get(__j, "method", __method);
-@for (mi, m) in s.methods.iter().enumerate() {
+    auto __data_member = __j.FindMember("data");
-            @if mi > 0 {else }if (__method == "@m.name") @{
+    if (__data_member == __j.MemberEnd() || !__data_member->value.IsObject())
-                @if m.ret_stream {
+        throw std::exception@{@};
-                auto __stream = std::make_shared<MRPCStream<@ty_to_str(m.ret.as_ref().unwrap())>>(&__c, __id);
+    auto &__data = __data_member->value;
-                @{ std::lock_guard guard@{__streams_mutex@}; __streams.emplace(&__c, __stream); @}
+@for (si, s) in rpc.services.iter().enumerate() {@if si > 0 { else }else{    }if (__service == "@s.name") @{
+@for (mi, m) in s.methods.iter().enumerate() {@if mi > 0 { else }else{        }if (__method == "@m.name") @{
+            @if m.ret_stream {auto __stream = MRPCStream<@ty_to_str(m.ret.as_ref().unwrap())>@{__c@};
             }
-@for (name, ty) in m.args.iter().map(|a| (&a.name, ty_to_str(&a.ty))) {                @ty @name = __data.at("@name");
+@for (name, ty) in m.args.iter().map(|a| (&a.name, ty_to_str(&a.ty))) {            @ty @name; json_get<@ty>(__data, "@name", @name);
             }
             @if m.ret_stream || m.ret.is_none() {@(s.name)_@(m.name)(@call_args(m));}
-                else {send_msg(__c, __id, @(s.name)_@(m.name)(@call_args(m)));}
+            else {send_msg(__c, @(s.name)_@(m.name)(@call_args(m)));}
-            @}
+        @}}
-}
+        else @{ throw std::exception@{@}; @}
+    @}}
     else @{ throw std::exception@{@}; @}
 @}
-}
-        else @{ throw std::exception@{@}; @}
-    @} catch (const std::exception &_) @{
-        std::cerr << "Got invalid request " << __id << " for " << __service << "::" << __method << std::endl;
-    @}
-@}
 @}

templates/cpp_server.rs.h

@@ -7,67 +7,62 @@
 #ifndef MRPC_GEN_H
 #define MRPC_GEN_H
 
-#include <unordered_map>
 #include <memory>
-#include <mutex>
-#include <iosfwd>
 #include <string>
+#include <vector>
+#include <optional>
 #include <cstdint>
-#include <crow.h>
+#include <cmath>
-#include <json.hpp>
+#include <corvusoft/restbed/byte.hpp>
+#define RAPIDJSON_HAS_STDSTRING 1
+#include <rapidjson/stringbuffer.h>
+#include <rapidjson/writer.h>
+#include <rapidjson/document.h>
+
+namespace restbed @{
+    class Resource;
+    class Session;
+@}
 
 namespace mrpc @{
+using MRPCJWriter = rapidjson::Writer<rapidjson::StringBuffer>;
 @for e in &rpc.enums {
 enum struct @e.name : std::uint64_t @{
     @e.values.iter().map(|(k,v)| format!("{k} = {v}")).join(",\n    ")
 @};
 }
-@for s in &rpc.structs {
+@for s in &rpc.structs {struct @s.name;
-struct @s.name;
-void to_json(nlohmann::json&, const @s.name&);
-void from_json(const nlohmann::json&, @s.name&);
 }
 @for s in &rpc.structs {
 struct @s.name @{
 @for f in &s.fields {    @ty_to_str(&f.ty) @f.name;
 }
-@};
+    MRPCJWriter& operator >>(MRPCJWriter&) const;
-}
+    @(s.name)& operator <<(const rapidjson::Value&);
-
+@};}
+@if streams_required(rpc).len() > 0 {
 struct MRPCStreamImpl @{
-    virtual void close() noexcept final;
+    void close() const noexcept;
-    virtual void abort() noexcept final;
+    bool is_open() const noexcept;
-    virtual bool is_open() noexcept final;
 protected:
-    MRPCStreamImpl(crow::websocket::connection *conn, uint64_t id) : conn(conn), id(id) @{@}
+    explicit MRPCStreamImpl(const std::shared_ptr<restbed::Session> &conn);
-    crow::websocket::connection* conn;
+    std::shared_ptr<restbed::Session> conn;
-    std::uint64_t id;
 @};
 
-template<class T>
+template<typename T>
 struct MRPCStream final : MRPCStreamImpl @{
-    MRPCStream(crow::websocket::connection *conn, uint64_t id) : MRPCStreamImpl(conn, id) @{@}
+    explicit MRPCStream(const std::shared_ptr<restbed::Session> &conn) : MRPCStreamImpl(conn) @{@}
-    bool send(const T &v) noexcept @{
+    void send(const T &v) const noexcept;
-        if (!conn) return false;
-        try @{
-            conn->send_text(nlohmann::json@{@{"id", id@},@{"data", v@}@}.dump());
-        @} catch (const std::exception &_) @{
-            abort();
-            return false;
-        @}
-        return true;
-    @}
 @};
 
+@for s in streams_required(rpc) {template struct MRPCStream<@(s)>;
+}}
 struct MRPCServer @{
-    virtual void install(crow::SimpleApp &app, std::string &&route) final;
+    explicit MRPCServer(std::shared_ptr<restbed::Resource>&);
 private:
 @for s in &rpc.services {@for m in &s.methods {    virtual @method_ret(m) @(s.name)_@(m.name)(@method_args(m)) = 0;
 }}
-    virtual void msg_handler(crow::websocket::connection&, const std::string&, bool) final;
+    virtual void msg_handler(std::shared_ptr<restbed::Session>, const restbed::Bytes&) final;
-
-    std::mutex __streams_mutex;
-    std::unordered_multimap<crow::websocket::connection*, std::shared_ptr<MRPCStreamImpl>> __streams;
 @};
 @}
 

templates/cpp_server_json.rs.cpp

@@ -0,0 +1,73 @@
+@use crate::data::RPC;
+@use crate::generators::cpp_s::*;
+
+@(rpc: &RPC)
+template<typename T>
+void json_get(const rapidjson::Value &j, const char *key, T &v);
+template<typename T>
+void json_get_inner(const rapidjson::Value&, T &v) = delete;
+@for (ty, jty) in JSON_INNER_IMPLS {
+template<> inline void json_get_inner(const rapidjson::Value &member, @ty &v) @{
+    if (!member.Is@(jty)())
+        throw std::exception@{@};
+    v = member.Get@(jty)();
+@}}
+
+template<typename T>
+inline void json_get_inner(const rapidjson::Value &member, std::optional<T> &v) @{
+    if (member.IsNull())
+        v = std::nullopt;
+    else @{
+        T t;
+        json_get_inner<T>(member, t);
+        v = std::move(t);
+    @}
+@}
+
+template<typename T>
+inline void json_get_inner(const rapidjson::Value &member, std::vector<T> &v) @{
+    if (!member.IsArray())
+        throw std::exception@{@};
+    for (const auto &j : member.GetArray()) @{
+        T t;
+        json_get_inner<T>(j, t);
+        v.push_back(std::move(t));
+    @}
+@}
+
+@for s in &rpc.structs {
+template<> inline void json_get_inner(const rapidjson::Value &__j, mrpc::@s.name &v) @{
+    using namespace mrpc;
+@for f in &s.fields {    json_get<@ty_to_str(&f.ty)>(__j, "@f.name", v.@f.name);
+}@}
+}
+
+@for e in &rpc.enums {
+template<> inline void json_get_inner(const rapidjson::Value &j, mrpc::@e.name &v) @{
+    json_get_inner<std::uint64_t>(j, (std::uint64_t&)v);
+@}
+mrpc::MRPCJWriter& operator >>(const mrpc::@e.name &v, mrpc::MRPCJWriter &w) @{
+    w.Uint64((std::uint64_t)v);
+    return w;
+@}
+}
+
+template<typename T>
+inline void json_get(const rapidjson::Value &j, const char *key, T &v) @{
+    auto member = j.FindMember(key);
+    if (member == j.MemberEnd())
+        throw std::exception@{@};
+    json_get_inner(member->value, v);
+@}
+
+namespace mrpc @{
+@for s in &rpc.structs {
+MRPCJWriter& @(s.name)::operator >>(MRPCJWriter &__w) const @{
+    __w.StartObject();
+@for f in &s.fields {    __w.Key("@f.name", @f.name.len());
+    @json_write(&f)
+}    __w.EndObject();
+    return __w;
+@}
+@(s.name)& @(s.name)::operator <<(const rapidjson::Value &__j) @{ json_get_inner<@(s.name)>(__j, *this); return *this; @}
+}

templates/typescript_client.rs.ts

@@ -3,79 +3,41 @@
 @use crate::generators::ts_c::*;
 
 @(rpc: &RPC)
+import @{ fetchEventSource @} from '@@microsoft/fetch-event-source';
 @for e in &rpc.enums {
 export enum @e.name @{
 @for (k,v) in &e.values {    @k = @v,
-}
+}@}
-@}
 }
 @for s in &rpc.structs {
 export interface @s.name @{
 @for f in &s.fields {    @f.name: @ty_to_str(&f.ty);
+}@}
 }
-@}
-}
-
-interface _WSResponse @{
-    id: number;
-    data: any;
-@}
-
-interface _WSWaitingEntry @{
-    ok: (v: any) => void;
-    err: (reason?: any) => void;
-@}
 
 export class MRPCConnector @{
     url: string;
-    socket: WebSocket;
-    nmi: number;
-    waiting: @{ [id: number]: _WSWaitingEntry @};
-    streams: @{ [id: number]: (v: any) => void @};
-
-    private open() @{
-        this.socket = new WebSocket(this.url);
-        this.socket.onmessage = ev => @{
-            const data = JSON.parse(ev.data) as _WSResponse;
-            if (data.id in this.streams) @{
-                this.streams[data.id](data.data);
-                if (data.data == null)
-                    delete this.streams[data.id];
-            @} else if (data.id in this.waiting) @{
-                this.waiting[data.id].ok(data.data);
-                delete this.waiting[data.id];
-            @} else @{
-                console.log(`Got unexpected message: $@{data@}`);
-            @}
-        @}
-        this.socket.onerror = () => setTimeout(() => @{this.open();@}, 2500);
-        this.socket.onclose = () => setTimeout(() => @{this.open();@}, 2500);
-    @}
-
-    private get_prom<T>(id: number): Promise<T> @{
-        return new Promise<T>((ok, err) => @{ this.waiting[id] = @{ok, err@}; @});
-    @}
 
     public constructor(url: string) @{
         this.url = url;
-        this.nmi = 0;
-        this.waiting = @{@};
-        this.streams = @{@};
-        this.open();
     @}
 
 @for s in &rpc.services { @for m in &s.methods {
     public @(s.name)_@(m.name)(@method_args(m))@method_ret(m) @{
         const __msg = @{
-            id: this.nmi++,
             service: '@s.name',
             method: '@m.name',
             data: @{@m.args.iter().map(|a| &a.name).join(",")@}
         @};
-        @if m.ret.is_some() && !m.ret_stream {const __p = this.get_prom<@ty_to_str(m.ret.as_ref().unwrap())>(__msg.id);}
+        @if m.ret.is_some() && !m.ret_stream {return fetch(this.url, @{
-        else if m.ret_stream {this.streams[__msg.id] = __cbk;}
+            method: 'POST',
-        this.socket.send(JSON.stringify(__msg));
+            body: JSON.stringify(__msg)
-        @if m.ret.is_some() && !m.ret_stream {return __p;}
+        @}).then((__r) => __r.json());}
+        else if m.ret_stream {fetchEventSource(this.url, @{
+            method: 'POST',
+            body: JSON.stringify(__msg),
+            onmessage: __e => __cbk(JSON.parse(__e.data))
+        @});} else {fetch(this.url, @{method: 'POST', body: JSON.stringify(__msg)@});}
     @}
 }}
 @}