jsonapi.cc

Go to the documentation of this file.

 // This Source Code Form is licensed MPL-2.0: http://mozilla.org/MPL/2.0
#include "jsonapi.hh"
#include "server.hh"
#include "main.hh"
#include "internal.hh"
 
#define GCDEBUG(...)      Ase::debug ("gc", __VA_ARGS__)
#define GCDEBUG_ENABLED() Ase::debug_key_enabled ("gc")
 
namespace Ase {
 
static String subprotocol_authentication;
 
void
jsonapi_set_subprotocol (const String &subprotocol)
{
  subprotocol_authentication = subprotocol;
}
 
// == JsonapiConnection ==
class JsonapiConnection;
using JsonapiConnectionP = std::shared_ptr<JsonapiConnection>;
using JsonapiConnectionW = std::weak_ptr<JsonapiConnection>;
static JsonapiConnectionP current_message_conection;
 
static bool
is_localhost (const String &url, int port)
{
  return Re::search ("^https?://(localhost|127\\.0\\.0\\.1)(:[0-9]+)?/", url, Re::I) >= 0;
}
 
class JsonapiConnection : public WebSocketConnection, public CustomDataContainer {
  Jsonipc::InstanceMap imap_, gcmap_;
  void
  log (const String &message) override
  {
    printerr ("%s: %s\n", nickname(), message);
  }
  int
  validate() override
  {
    using namespace AnsiColors;
    const auto C1 = color (BOLD), C0 = color (BOLD_OFF);
    const Info info = get_info();
    const String origin = info.header ("Origin") + "/";
    const bool localhost_origin = is_localhost (origin, info.lport);
    const bool subproto_ok = (info.subs.size() == 0 && subprotocol_authentication.empty()) ||
                             (info.subs.size() == 1 && subprotocol_authentication == info.subs[0]);
    if (localhost_origin && subproto_ok)
      return 0; // OK
    // log rejection
    String why;
    if (!localhost_origin)      why = "Bad Origin";
    else if (!subproto_ok)      why = "Bad Subprotocol";
    const String ua = info.header ("User-Agent");
    if (logflags_ & 2)
      log (string_format ("%sREJECT:%s  %s:%d/ (%s) - %s", C1, C0, info.remote, info.rport, why, ua));
    return -1; // reject
  }
  void
  opened() override
  {
    using namespace AnsiColors;
    const auto C1 = color (BOLD), C0 = color (BOLD_OFF);
    const Info info = get_info();
    const String ua = info.header ("User-Agent");
    if (logflags_ & 4)
      log (string_format ("%sACCEPT:%s  %s:%d/ - %s", C1, C0, info.remote, info.rport, ua));
  }
  void
  closed() override
  {
    using namespace AnsiColors;
    const auto C1 = color (BOLD), C0 = color (BOLD_OFF);
    if (logflags_ & 4)
      log (string_format ("%sCLOSED%s", C1, C0));
    trigger_destroy_hooks();
  }
  void
  message (const String &message) override
  {
    JsonapiConnectionP conp = std::dynamic_pointer_cast<JsonapiConnection> (shared_from_this());
    assert_return (conp);
    struct MessageData { // keep arguments alive, even after __func__ returns
      ScopedSemaphore sem;
      String message, reply;
      JsonapiConnectionP conp;
      MessageData (JsonapiConnectionP c, const String &m) : message (m), conp (c) {}
    };
    auto data = std::make_shared<MessageData> (conp, message);
    main_jobs += [data, this] () {
      if (main_loop->has_quit())
        return;
      current_message_conection = data->conp;
      data->reply = this->handle_jsonipc (data->message);
      current_message_conection = nullptr;
      data->sem.post();
    };
    nickname(); // cache socket nickname for use during errors
    // wait with timeout, checking if main loop has quit to avoid deadlock on shutdown
    const uint64_t timeout_us = 50 * 1000;
    do {
      const int ret = data->sem.wait_for (timeout_us);
      if (ret == 0)
        break; // semaphore was posted, job completed successfully
      if (data->reply.empty() && main_loop->has_quit())
        data->reply = "{id:0,error:{code:-32601,message:\"Method not found: endpoint shutting down\"}}\n";
    } while (data->reply.empty());
    // when queueing asynchronously, we have to use WebSocketConnectionP
    if (!data->reply.empty())
      send_text (data->reply);
  }
  String handle_jsonipc (const std::string &message);
  std::vector<JsTrigger> triggers_; // HINT: use unordered_map if this becomes slow
public:
  explicit JsonapiConnection (WebSocketConnection::Internals &internals, int logflags) :
    WebSocketConnection (internals, logflags)
  {}
  ~JsonapiConnection()
  {
    trigger_destroy_hooks();
  }
  bool
  renew_gc ()
  {
    const bool starting_gc = imap_.mark_unused();
    GCDEBUG ("%s: imap_=%d%s\n", __func__, imap_.size(), starting_gc ? " (duplicate)" : "");
    return starting_gc; // false: duplicate request, waiting for report_gc
  }
  bool
  report_gc (const std::vector<size_t> &ids)
  {
    const size_t preerved = imap_.purge_unused (ids);
    GCDEBUG ("%s: considered=%d retained=%d purged=%d active=%d\n", __func__,
             ids.size(), preerved, ids.size() - preerved, imap_.size());
    return imap_.size();
  }
  JsTrigger
  trigger_lookup (const String &id)
  {
    for (auto it = triggers_.begin(); it != triggers_.end(); it++)
      if (id == it->id())
        return *it;
    return {};
  }
  void
  trigger_remove (const String &id)
  {
    trigger_lookup (id).destroy();
  }
  void
  trigger_create (const String &id)
  {
    using namespace Jsonipc;
    JsonapiConnectionP jsonapi_connection_p = std::dynamic_pointer_cast<JsonapiConnection> (shared_from_this());
    assert_return (jsonapi_connection_p);
    std::weak_ptr<JsonapiConnection> selfw = jsonapi_connection_p;
    const int logflags = logflags_;
    // marshal remote trigger
    auto trigger_remote = [selfw, id, logflags] (ValueS &&args)    // weak_ref avoids cycles
    {
      JsonapiConnectionP selfp = selfw.lock();
      return_unless (selfp);
      const String msg = jsonobject_to_string ("method", id /*"Jsonapi/Trigger/_%%%"*/, "params", args);
      if (logflags & 8)
        selfp->log (string_format ("⬰ %s", msg));
      selfp->send_text (msg);
    };
    JsTrigger trigger = JsTrigger::create (id, trigger_remote);
    triggers_.push_back (trigger);
    // marshall remote destroy notification and erase triggers_ entry
    auto erase_trigger = [selfw, id, logflags] ()               // weak_ref avoids cycles
    {
      std::shared_ptr<JsonapiConnection> selfp = selfw.lock();
      return_unless (selfp);
      if (selfp->is_open())
        {
          ValueS args { id };
          const String msg = jsonobject_to_string ("method", "Jsonapi/Trigger/killed", "params", args);
          if (logflags & 8)
            selfp->log (string_format ("↚ %s", msg));
          selfp->send_text (msg);
        }
      Aux::erase_first (selfp->triggers_, [id] (auto &t) { return id == t.id(); });
    };
    trigger.ondestroy (erase_trigger);
  }
  void
  trigger_destroy_hooks()
  {
    std::vector<JsTrigger> old;
    old.swap (triggers_); // speed up erase_trigger() searches
    for (auto &trigger : old)
      trigger.destroy();
    custom_data_destroy();
  }
};
 
WebSocketConnectionP
jsonapi_make_connection (WebSocketConnection::Internals &internals, int logflags)
{
  return std::make_shared<JsonapiConnection> (internals, logflags);
}
 
#define ERROR500(WHAT)                                          \
  Jsonipc::bad_invocation (-32500,                              \
                           __FILE__ ":"                         \
                           ASE_CPP_STRINGIFY (__LINE__) ": "    \
                           "Internal Server Error: "            \
                           WHAT)
#define assert_500(c) (__builtin_expect (static_cast<bool> (c), 1) ? (void) 0 : throw ERROR500 (#c) )
 
static Jsonipc::IpcDispatcher*
make_dispatcher()
{
  using namespace Jsonipc;
  static IpcDispatcher *dispatcher = [] () {
    dispatcher = new IpcDispatcher();
    dispatcher->add_method ("Jsonapi/renew-gc",
                            [] (CallbackInfo &cbi)
                            {
                              assert_500 (current_message_conection);
                              if (cbi.n_args() > 0)
                                throw Jsonipc::bad_invocation (-32602, "Invalid params");
                              const auto ret = current_message_conection->renew_gc ();
                              cbi.set_result (to_json (ret, cbi.allocator()).Move());
                            });
    dispatcher->add_method ("Jsonapi/report-gc",
                            [] (CallbackInfo &cbi)
                            {
                              assert_500 (current_message_conection);
                              if (cbi.n_args() != 1)
                                throw Jsonipc::bad_invocation (-32602, "Invalid params");
                              const auto ids = from_json<std::vector<size_t>> (cbi.ntharg (0));
                              const auto ret = current_message_conection->report_gc (ids);
                              cbi.set_result (to_json (ret, cbi.allocator()).Move());
                            });
    dispatcher->add_method ("Jsonapi/initialize",
                            [] (CallbackInfo &cbi)
                            {
                              assert_500 (current_message_conection);
                              Server &server = ASE_SERVER;
                              std::shared_ptr<Server> serverp = shared_ptr_cast<Server> (&server);
                              cbi.set_result (to_json (serverp, cbi.allocator()).Move());
                            });
    dispatcher->add_method ("Jsonapi/Trigger/create",
                            [] (CallbackInfo &cbi)
                            {
                              assert_500 (current_message_conection);
                              const String triggerid = cbi.n_args() == 1 ? from_json<String> (cbi.ntharg (0)) : "";
                              if (triggerid.compare (0, 17, "Jsonapi/Trigger/_") != 0)
                                throw Jsonipc::bad_invocation (-32602, "Invalid params");
                              current_message_conection->trigger_create (triggerid);
                            });
    dispatcher->add_method ("Jsonapi/Trigger/remove",
                            [] (CallbackInfo &cbi)
                            {
                              assert_500 (current_message_conection);
                              const String triggerid = cbi.n_args() == 1 ? from_json<String> (cbi.ntharg (0)) : "";
                              if (triggerid.compare (0, 17, "Jsonapi/Trigger/_") != 0)
                                throw Jsonipc::bad_invocation (-32602, "Invalid params");
                              current_message_conection->trigger_remove (triggerid);
                            });
    return dispatcher;
  } ();
  return dispatcher;
}
 
String
JsonapiConnection::handle_jsonipc (const std::string &message)
{
  if (logflags_ & 8)
    log (string_format ("→ %s", message.size() > 1024 ? message.substr (0, 1020) + "..." + message.back() : message));
  Jsonipc::Scope message_scope (imap_);
  String reply;
  { // enfore notifies *before* reply (and the corresponding log() messages)
    CoalesceNotifies coalesce_notifies; // coalesce multiple "notify:detail" emissions
    reply = make_dispatcher()->dispatch_message (message);
  } // coalesced notifications occour *here*
  if (logflags_ & 8)
    {
      const char *errorat = strstr (reply.c_str(), "\"error\":{");
      if (errorat && errorat > reply.c_str() && (errorat[-1] == ',' || errorat[-1] == '{'))
        {
          using namespace AnsiColors;
          auto R1 = color (BOLD) + color (FG_RED), R0 = color (FG_DEFAULT) + color (BOLD_OFF);
          log (string_format ("%s←%s %s", R1, R0, reply));
        }
      else
        log (string_format ("← %s", reply.size() > 1024 ? reply.substr (0, 1020) + "..." + reply.back() : reply));
    }
  return reply;
}
 
// == JsTrigger ==
class JsTrigger::Impl {
  using Func = std::function<void (ValueS)>;
  const String id;
  Func func;
  using VoidFunc = std::function<void()>;
  std::vector<VoidFunc> destroyhooks;
  friend class JsTrigger;
  /*ctor*/ Impl   () = delete;
  /*copy*/ Impl   (const Impl&) = delete;
  Impl& operator= (const Impl&) = delete;
public:
  ~Impl ()
  {
    destroy();
  }
  Impl (const Func &f, const String &_id) :
    id (_id), func (f)
  {}
  void
  destroy ()
  {
    func = nullptr;
    while (!destroyhooks.empty())
      {
        VoidFunc destroyhook = destroyhooks.back();
        destroyhooks.pop_back();
        destroyhook();
      }
  }
};
 
void
JsTrigger::ondestroy (const VoidFunc &vf)
{
  assert_return (p_);
  if (vf)
    p_->destroyhooks.push_back (vf);
}
 
void
JsTrigger::call (ValueS &&args) const
{
  assert_return (p_);
  if (p_->func)
    p_->func (std::move (args));
}
 
JsTrigger
JsTrigger::create (const String &triggerid, const JsTrigger::Impl::Func &f)
{
  JsTrigger trigger;
  trigger.p_ = std::make_shared<JsTrigger::Impl> (f, triggerid);
  assert_return (f != nullptr, trigger);
  return trigger;
}
 
String
JsTrigger::id () const
{
  return p_ ? p_->id : "";
}
 
void
JsTrigger::destroy ()
{
  if (p_)
    p_->destroy();
}
 
JsTrigger::operator bool () const noexcept
{
  return p_ && p_->func;
}
 
JsTrigger
ConvertJsTrigger::lookup (const String &triggerid)
{
  if (current_message_conection)
    return current_message_conection->trigger_lookup (triggerid);
  assert_return (current_message_conection, {});
  return {};
}
 
CustomDataContainer*
jsonapi_connection_data ()
{
  if (current_message_conection)
    return current_message_conection.get();
  return nullptr;
}
 
JsonapiBinarySender
jsonapi_connection_sender ()
{
  return_unless (current_message_conection, {});
  JsonapiConnectionW conw = current_message_conection;
  return [conw] (const String &blob) {
    JsonapiConnectionP conp = conw.lock();
    return conp ? conp->send_binary (blob) : false;
  };
}
 
} // Ase
 
// Build generated bindings
#include "ase/gen/api-jsonipc.g.cc"
 
#include "testing.hh"
 
namespace { // Anon
 
TEST_INTEGRITY (jsonapi_tests);
static void
jsonapi_tests()
{
  using namespace Ase;
  using IdStringMap = std::map<size_t,std::string>;
  IdStringMap tmap;
  for (size_t i = 1; i <= 99; i++)
    tmap[1000 - i] = string_format ("%d", i);
  TASSERT (tmap.size() == 99);
  for (auto it = tmap.begin(), next = it; it != tmap.end() ? ++next, 1 : 0; it = next) // keep next ahead of it, but avoid ++end
    tmap.erase (it);
  TASSERT (tmap.size() == 0);
}
 
} // Anon