// Matrix Construct // // Copyright (C) Matrix Construct Developers, Authors & Contributors // Copyright (C) 2016-2018 Jason Volk // // Permission to use, copy, modify, and/or distribute this software for any // purpose with or without fee is hereby granted, provided that the above // copyright notice and this permission notice is present in all copies. The // full license for this software is available in the LICENSE file. namespace ircd::m::feds { struct request_base; template struct request; using request_list = std::list>; template using create_closure = std::function &, const string_view &origin)>; template static request_list for_one(const string_view &origin, const opts &, const closure &, const create_closure &); template static request_list for_each_in_room(const opts &, const closure &, const create_closure &); static bool call_user(const closure &closure, const result &result); static bool handler(request_list &, const milliseconds &, const closure &); static request_list head(const opts &, const closure &); static request_list auth(const opts &, const closure &); static request_list event(const opts &, const closure &); static request_list state(const opts &, const closure &); static request_list backfill(const opts &, const closure &); static request_list version(const opts &, const closure &); static request_list keys(const opts &, const closure &); static request_list send(const opts &, const closure &); } // // request_base // /// Polymorphic non-template base struct ircd::m::feds::request_base { const feds::opts *opts {nullptr}; char origin[event::ORIGIN_MAX_SIZE]; request_base(const feds::opts &opts) :opts{&opts} {} request_base() = default; virtual ~request_base() noexcept; }; ircd::m::feds::request_base::~request_base() noexcept { } // // request // template struct ircd::m::feds::request :request_base ,T { char buf[8_KiB]; request(const feds::opts &opts, const std::function &closure) :request_base{opts} ,T(closure(*this)) {} request(request &&) = delete; request(const request &) = delete; ~request() noexcept; }; template ircd::m::feds::request::~request() noexcept { if(this->valid()) { server::cancel(*this); this->wait(); } } /////////////////////////////////////////////////////////////////////////////// // // m/feds.h // ircd::m::feds::execute::execute(const vector_view &optsv, const closure &closure) :returns{true} { request_list list; for(const auto &opts : optsv) switch(opts.op) { case op::head: list.splice(list.end(), head(opts, closure)); continue; case op::auth: list.splice(list.end(), auth(opts, closure)); continue; case op::event: list.splice(list.end(), event(opts, closure)); continue; case op::state: list.splice(list.end(), state(opts, closure)); continue; case op::backfill: list.splice(list.end(), backfill(opts, closure)); continue; case op::version: list.splice(list.end(), version(opts, closure)); continue; case op::keys: list.splice(list.end(), keys(opts, closure)); continue; case op::send: list.splice(list.end(), send(opts, closure)); continue; case op::noop: continue; } milliseconds timeout {0}; for(const auto &opts : optsv) timeout = std::max(opts.timeout, timeout); ret = handler(list, timeout, closure); } ircd::m::feds::request_list ircd::m::feds::send(const opts &opts, const closure &closure) { const auto make_request{[&opts] (auto &request, const auto &origin) { m::fed::send::opts v1opts; v1opts.remote = string_view { strlcpy{request.origin, origin} }; return m::fed::send { opts.arg[0], opts.arg[1], request.buf, std::move(v1opts) }; }}; return for_each_in_room(opts, closure, make_request); } ircd::m::feds::request_list ircd::m::feds::keys(const opts &opts, const closure &closure) { const auto make_request{[&opts] (auto &request, const auto &origin) { m::fed::key::query::opts v1opts; v1opts.dynamic = false; v1opts.remote = string_view { strlcpy{request.origin, origin} }; const m::fed::key::server_key server_key { opts.arg[0], opts.arg[1] }; return m::fed::key::query { {&server_key, 1}, request.buf, std::move(v1opts) }; }}; return opts.room_id? for_each_in_room(opts, closure, make_request): for_one(opts.arg[0], opts, closure, make_request); } ircd::m::feds::request_list ircd::m::feds::version(const opts &opts, const closure &closure) { static const auto make_request{[] (auto &request, const auto &origin) { m::fed::version::opts opts; opts.dynamic = false; opts.remote = string_view { strlcpy{request.origin, origin} }; return m::fed::version { request.buf, std::move(opts) }; }}; return for_each_in_room(opts, closure, make_request); } ircd::m::feds::request_list ircd::m::feds::backfill(const opts &opts, const closure &closure) { const auto make_request{[&opts] (auto &request, const auto &origin) { m::fed::backfill::opts v1opts; v1opts.event_id = opts.event_id; v1opts.limit = opts.argi[0]; v1opts.remote = string_view { strlcpy{request.origin, origin} }; return m::fed::backfill { opts.room_id, request.buf, std::move(v1opts) }; }}; return for_each_in_room(opts, closure, make_request); } ircd::m::feds::request_list ircd::m::feds::state(const opts &opts, const closure &closure) { const auto make_request{[&opts] (auto &request, const auto &origin) { m::fed::state::opts v1opts; v1opts.ids_only = opts.arg[0] == "ids"; v1opts.event_id = opts.event_id; v1opts.remote = string_view { strlcpy{request.origin, origin} }; return m::fed::state { opts.room_id, request.buf, std::move(v1opts) }; }}; return for_each_in_room(opts, closure, make_request); } ircd::m::feds::request_list ircd::m::feds::event(const opts &opts, const closure &closure) { const auto make_request{[&opts] (auto &request, const auto &origin) { m::fed::event::opts v1opts; v1opts.remote = string_view { strlcpy{request.origin, origin} }; return m::fed::event { opts.event_id, request.buf, std::move(v1opts) }; }}; return for_each_in_room(opts, closure, make_request); } ircd::m::feds::request_list ircd::m::feds::auth(const opts &opts, const closure &closure) { const auto make_request{[&opts] (auto &request, const auto &origin) { m::fed::event_auth::opts v1opts; v1opts.remote = string_view { strlcpy{request.origin, origin} }; return m::fed::event_auth { opts.room_id, opts.event_id, request.buf, std::move(v1opts) }; }}; return for_each_in_room(opts, closure, make_request); } ircd::m::feds::request_list ircd::m::feds::head(const opts &opts, const closure &closure) { const auto make_request{[&opts] (auto &request, const auto &origin) { m::fed::make_join::opts v1opts; v1opts.remote = string_view { strlcpy{request.origin, origin} }; return m::fed::make_join { opts.room_id, opts.user_id, request.buf, std::move(v1opts) }; }}; return for_each_in_room(opts, closure, make_request); } /////////////////////////////////////////////////////////////////////////////// // // (internal) // bool ircd::m::feds::handler(request_list &reqs, const milliseconds &timeout, const closure &closure) { const auto when { now() + timeout }; while(!reqs.empty()) { static const auto dereferencer{[] (auto &iterator) -> server::request & { return dynamic_cast(**iterator); }}; auto next { ctx::when_any(begin(reqs), end(reqs), dereferencer) }; if(!next.wait_until(when, std::nothrow)) break; const auto it { next.get() }; assert(it != end(reqs)); const unwind remove{[&reqs, &it] { reqs.erase(it); }}; request_base &req(**it); server::request &sreq(dynamic_cast(req)); try { const auto code{sreq.get()}; const json::array &array{sreq.in.content}; const json::object &object{sreq.in.content}; const result result { req.opts, req.origin, {}, object, array }; if(!call_user(closure, result)) return false; } catch(const std::exception &) { if(!req.opts->closure_errors && !req.opts->nothrow_closure_retval) return false; if(!req.opts->closure_errors) continue; const ctx::exception_handler eptr; const std::exception_ptr &eptr_(eptr); const result result { req.opts, req.origin, eptr_ }; if(!call_user(closure, result)) return false; } } return true; } bool ircd::m::feds::call_user(const closure &closure, const result &result) try { return closure(result); } catch(const std::exception &) { assert(result.request); if(result.request->nothrow_closure) return result.request->nothrow_closure_retval; throw; } template ircd::m::feds::request_list ircd::m::feds::for_each_in_room(const opts &opts, const feds::closure &closure, const std::function &, const string_view &origin)> &create_closure) { request_list ret; if(!opts.room_id) return ret; const m::room::origins origins { opts.room_id }; // Prelink loop if(opts.prelink) origins.for_each([&opts] (const string_view &origin) { if(opts.exclude_myself && my_host(origin)) return; fed::prelink(origin); }); // Request loop origins.for_each([&opts, &ret, &closure, &create_closure] (const string_view &origin) { if(opts.exclude_myself && my_host(origin)) return; const auto errant { fed::errant(origin) }; if(opts.closure_cached_errors || !errant) try { ret.emplace_back(std::make_unique>(opts, [&create_closure, &origin] (auto &request) { return create_closure(request, origin); })); } catch(const std::exception &) { if(!opts.closure_cached_errors) return; feds::result result; result.request = &opts; result.origin = origin; result.eptr = std::current_exception(); const ctx::exception_handler eh; m::feds::call_user(closure, result); } }); return ret; } template ircd::m::feds::request_list ircd::m::feds::for_one(const string_view &origin, const opts &opts, const feds::closure &closure, const std::function &, const string_view &origin)> &create_closure) { request_list ret; if(opts.exclude_myself && my_host(origin)) return ret; const auto errant { fed::errant(origin) }; if(opts.closure_cached_errors || !errant) try { ret.emplace_back(std::make_unique>(opts, [&create_closure, &origin] (auto &request) { return create_closure(request, origin); })); } catch(const std::exception &e) { if(!opts.closure_cached_errors) return ret; feds::result result; result.request = &opts; result.origin = origin; result.eptr = std::current_exception(); const ctx::exception_handler eh; m::feds::call_user(closure, result); } return ret; }