MirrorHub/construct
0
0
Fork 0
mirror of https://github.com/matrix-construct/construct synced 2024-11-25 08:12:37 +01:00
Code Issues Releases Wiki Activity

modules/client/sync: Additional sync items reorganization.

Browse source
This commit is contained in:
Jason Volk 2019-01-04 14:47:01 -08:00
parent 136d3a61a3
commit 61e88e75bc
10 changed files with 364 additions and 272 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
include/ircd/m/sync.h
View file

@ -34,19 +34,16 @@ struct ircd::m::sync::item
handle _polylog;
handle _linear;
handle _longpoll;
public:
string_view name() const;
bool polylog(data &);
bool linear(data &, const m::event &);
bool longpoll(data &, const m::event &);
bool polylog(data &);
item(std::string name,
handle polylog = {},
handle linear = {},
handle longpoll = {});
handle linear = {});
item(item &&) = delete;
item(const item &) = delete;
@ -86,6 +83,8 @@ struct ircd::m::sync::data
const m::event *event {nullptr};
const m::room *room {nullptr};
string_view membership;
window_buffer ret;
std::array<string_view, 16> path;
// unsorted / misc
uint64_t state_at {0};

48
ircd/m/m.cc
View file

@ -580,8 +580,7 @@ ircd::m::sync::item::instance_multimap::map
ircd::m::sync::item::item(std::string name,
handle polylog,
handle linear,
handle longpoll)
handle linear)
:instance_multimap
{
std::move(name)
@ -594,10 +593,6 @@ ircd::m::sync::item::item(std::string name,
{
std::move(linear)
}
,_longpoll
{
std::move(longpoll)
}
{
log::debug
{
@ -618,23 +613,6 @@ noexcept
};
}
bool
ircd::m::sync::item::longpoll(data &data,
const m::event &event)
try
{
const auto ret
{
_longpoll(data)
};
return ret;
}
catch(const std::bad_function_call &)
{
return false;
}
bool
ircd::m::sync::item::linear(data &data,
const m::event &event)
@ -652,8 +630,18 @@ try
return ret;
}
catch(const std::bad_function_call &)
catch(const std::bad_function_call &e)
{
thread_local char rembuf[128];
log::dwarning
{
log, "linear %s %s '%s' missing handler :%s",
string(rembuf, ircd::remote(data.client)),
string_view{data.user.user_id},
name(),
e.what()
};
return false;
}
@ -687,8 +675,18 @@ try
return ret;
}
catch(const std::bad_function_call &)
catch(const std::bad_function_call &e)
{
thread_local char rembuf[128];
log::dwarning
{
log, "polylog %s %s '%s' missing handler :%s",
string(rembuf, ircd::remote(data.client)),
string_view{data.user.user_id},
name(),
e.what()
};
return false;
}
catch(const std::exception &e)

2
modules/Makefile.am
View file

@ -264,6 +264,7 @@ client_module_LTLIBRARIES += \
client_client_sync_account_data_la_SOURCES = client/sync/account_data.cc
client_client_sync_presence_la_SOURCES = client/sync/presence.cc
client_client_sync_rooms_la_SOURCES = client/sync/rooms.cc
client_client_sync_rooms_account_data_la_SOURCES = client/sync/rooms/account_data.cc
client_client_sync_rooms_receipt_la_SOURCES = client/sync/rooms/receipt.cc
client_client_sync_rooms_state_la_SOURCES = client/sync/rooms/state.cc
@ -273,6 +274,7 @@ client_client_sync_rooms_unread_notifications_la_SOURCES = client/sync/rooms/unr
client_module_LTLIBRARIES += \
client/client_sync_account_data.la \
client/client_sync_presence.la \
client/client_sync_rooms.la \
client/client_sync_rooms_account_data.la \
client/client_sync_rooms_receipt.la \
client/client_sync_rooms_state.la \

256
modules/client/sync.cc
View file

@ -98,11 +98,6 @@ try
"Since parameter is too far in the future..."
};
json::stack::object top
{
data.out
};
const size_t linear_delta_max
{
linear::delta_max
@ -113,8 +108,8 @@ try
data.delta == 0?
false:
data.delta > linear_delta_max?
polylog::handle(client, data, top):
linear::handle(client, data, top)
polylog::handle(data):
linear::handle(data)
};
// When shortpoll was successful, do nothing else.
@ -129,7 +124,7 @@ try
// 200 with empty fields rather than a 408.
const json::value next_batch
{
lex_cast(m::vm::current_sequence), json::STRING
lex_cast(m::vm::current_sequence + 1), json::STRING
};
return resource::response
@ -276,7 +271,7 @@ ircd::m::sync::longpoll::handle(client &client,
const auto &next_batch
{
int64_t(m::vm::current_sequence)
int64_t(m::vm::current_sequence + 1)
};
resource::response
@ -449,13 +444,17 @@ ircd::m::sync::linear::delta_max
};
bool
ircd::m::sync::linear::handle(client &client,
data &sp,
json::stack::object &object)
ircd::m::sync::linear::handle(data &data)
{
auto &client{data.client};
json::stack::object object
{
data.out
};
uint64_t since
{
sp.since
data.since
};
std::map<std::string, std::vector<std::string>, std::less<>> r;
@ -464,7 +463,7 @@ ircd::m::sync::linear::handle(client &client,
m::events::for_each(since, [&]
(const uint64_t &sequence, const m::event &event)
{
if(!r.empty() && (since - sp.since > 128))
if(!r.empty() && (since - data.since > 128))
{
limited = true;
return false;
@ -479,7 +478,7 @@ ircd::m::sync::linear::handle(client &client,
json::get<"room_id"_>(event)
};
if(!room.membership(sp.user.user_id))
if(!room.membership(data.user.user_id))
return true;
auto it
@ -523,7 +522,7 @@ ircd::m::sync::linear::handle(client &client,
m::event::id::buf last_read_buf;
const auto &last_read
{
m::receipt::read(last_read_buf, room_id, sp.user)
m::receipt::read(last_read_buf, room_id, data.user)
};
const auto last_read_idx
@ -536,14 +535,14 @@ ircd::m::sync::linear::handle(client &client,
const auto notes
{
last_read_idx?
notification_count(room_id, last_read_idx, sp.current):
notification_count(room_id, last_read_idx, data.current):
json::undefined_number
};
const auto highlights
{
last_read_idx?
highlight_count(room_id, sp.user, last_read_idx, sp.current):
highlight_count(room_id, data.user, last_read_idx, data.current):
json::undefined_number
};
@ -578,7 +577,7 @@ ircd::m::sync::linear::handle(client &client,
};
thread_local char membership_buf[64];
const auto membership{m::room{room_id}.membership(membership_buf, sp.user)};
const auto membership{m::room{room_id}.membership(membership_buf, data.user)};
const int ep
{
membership == "join"? 0:
@ -616,7 +615,7 @@ ircd::m::sync::linear::handle(client &client,
{
client, json::members
{
{ "next_batch", json::value { lex_cast(int64_t(since)), json::STRING } },
{ "next_batch", json::value { lex_cast(int64_t(since + 1)), json::STRING } },
{ "rooms", rooms },
{ "presence", json::object{} },
}
@ -654,25 +653,15 @@ ircd::m::sync::highlight_count(const m::room &r,
// polylog
//
namespace ircd
{
ctx::pool::opts mepool_opts
{
256_KiB
};
ctx::pool mepool
{
"me pool", mepool_opts
};
};
bool
ircd::m::sync::polylog::handle(client &client,
data &data,
json::stack::object &object)
ircd::m::sync::polylog::handle(data &data)
try
{
json::stack::object object
{
data.out
};
// Generate individual stats for sections
thread_local char iecbuf[64], rembuf[128], tmbuf[32];
sync::stats stats{data.stats};
@ -694,11 +683,16 @@ try
{
json::stack::member member{object, "rooms"};
json::stack::object object{member};
sync_rooms(data, object, "invite");
sync_rooms(data, object, "join");
sync_rooms(data, object, "leave");
sync_rooms(data, object, "ban");
const scope_restore<decltype(data.member)> theirs
{
data.member, &member
};
auto it(m::sync::item::map.find("rooms"));
assert(it != m::sync::item::map.end());
const auto &item(it->second);
assert(item);
item->polylog(data);
}
{
@ -718,7 +712,7 @@ try
{
json::stack::member member
{
object, "next_batch", json::value(lex_cast(int64_t(data.current)), json::STRING)
object, "next_batch", json::value(lex_cast(int64_t(data.current + 1)), json::STRING)
};
}
@ -747,181 +741,3 @@ catch(const std::exception &e)
throw;
}
void
ircd::m::sync::polylog::sync_rooms(data &data,
json::stack::object &out,
const string_view &membership)
{
const scope_restore<decltype(data.membership)> theirs
{
data.membership, membership
};
json::stack::member rooms_member
{
out, membership
};
json::stack::object rooms_object
{
rooms_member
};
data.user_rooms.for_each(membership, [&data, &rooms_object]
(const m::room &room, const string_view &membership)
{
if(head_idx(std::nothrow, room) <= data.since)
return;
// Generate individual stats for this room's sync
#ifdef RB_DEBUG
sync::stats stats{data.stats};
stats.timer = timer{};
#endif
// This scope ensures the object destructs and flushes before
// the log message tallying the stats for this room below.
{
json::stack::member member{rooms_object, room.room_id};
json::stack::object object{member};
sync_room(data, object, room);
}
#ifdef RB_DEBUG
thread_local char iecbuf[64], rembuf[128], tmbuf[32];
log::debug
{
log, "polylog %s %s %s %s wc:%zu in %s",
string(rembuf, ircd::remote(data.client)),
string_view{data.user.user_id},
string_view{room.room_id},
pretty(iecbuf, iec(data.stats.flush_bytes - stats.flush_bytes)),
data.stats.flush_count - stats.flush_count,
ircd::pretty(tmbuf, stats.timer.at<milliseconds>(), true)
};
#endif
});
}
void
ircd::m::sync::polylog::sync_room(data &data,
json::stack::object &out,
const m::room &room)
try
{
const scope_restore<decltype(data.room)> theirs
{
data.room, &room
};
// timeline
{
auto it(m::sync::item::map.find("rooms...timeline"));
assert(it != m::sync::item::map.end());
const auto &item(it->second);
assert(item);
json::stack::member member{out, "timeline"};
const scope_restore<decltype(data.member)> theirs
{
data.member, &member
};
item->polylog(data);
}
// state
{
auto it(m::sync::item::map.find("rooms...state"));
assert(it != m::sync::item::map.end());
const auto &item(it->second);
assert(item);
json::stack::member member
{
out, data.membership != "invite"?
"state":
"invite_state"
};
const scope_restore<decltype(data.member)> theirs
{
data.member, &member
};
item->polylog(data);
}
// ephemeral
{
auto pit
{
m::sync::item::map.equal_range("rooms...ephemeral")
};
assert(pit.first != pit.second);
json::stack::member member{out, "ephemeral"};
json::stack::object object{member};
const scope_restore<decltype(data.object)> theirs
{
data.object, &object
};
{
json::stack::member member{object, "events"};
json::stack::array array{member};
const scope_restore<decltype(data.array)> theirs
{
data.array, &array
};
for(; pit.first != pit.second; ++pit.first)
{
const auto &item(pit.first->second);
assert(item);
item->polylog(data);
}
}
}
// account_data
{
auto it(m::sync::item::map.find("rooms...account_data"));
assert(it != m::sync::item::map.end());
const auto &item(it->second);
assert(item);
json::stack::member member{out, "account_data"};
const scope_restore<decltype(data.member)> theirs
{
data.member, &member
};
item->polylog(data);
}
// unread_notifications
{
auto it(m::sync::item::map.find("rooms...unread_notifications"));
assert(it != m::sync::item::map.end());
const auto &item(it->second);
assert(item);
json::stack::member member{out, "unread_notifications"};
const scope_restore<decltype(data.member)> theirs
{
data.member, &member
};
item->polylog(data);
}
}
catch(const json::not_found &e)
{
log::critical
{
log, "polylog sync room %s error %lu to %lu (vm @ %zu) :%s"
,string_view{room.room_id}
,data.since
,data.current
,m::vm::current_sequence
,e.what()
};
}

6
modules/client/sync.h
View file

@ -70,14 +70,12 @@ namespace ircd::m::sync::linear
{
extern conf::item<size_t> delta_max;
static bool handle(client &, data &, json::stack::object &);
static bool handle(data &);
}
namespace ircd::m::sync::polylog
{
static void sync_room(data &, json::stack::object &, const m::room &);
static void sync_rooms(data &, json::stack::object &, const string_view &membership);
static bool handle(client &, data &, json::stack::object &);
static bool handle(data &);
}
/// Argument parser for the client's /sync request

53
modules/client/sync/presence.cc
View file

@ -17,6 +17,7 @@ IRCD_MODULE
namespace ircd::m::sync
{
static bool presence_polylog(data &);
static bool presence_linear(data &);
extern item presence;
}
@ -24,21 +25,44 @@ decltype(ircd::m::sync::presence)
ircd::m::sync::presence
{
"presence",
presence_polylog
presence_polylog,
presence_linear,
};
namespace ircd
bool
ircd::m::sync::presence_linear(data &data)
{
ctx::pool::opts meepool_opts
{
256_KiB
};
if(!data.event)
return false;
ctx::pool meepool
{
"meepool", meepool_opts
};
};
if(json::get<"type"_>(*data.event) != "ircd.presence")
return false;
json::stack::object object
{
*data.array
};
// sender
json::stack::member
{
object, "sender", unquote(at<"content"_>(*data.event).get("user_id"))
};
// type
json::stack::member
{
object, "type", json::value{"m.presence"}
};
// content
json::stack::member
{
object, "content", at<"content"_>(*data.event)
};
return true;
}
bool
ircd::m::sync::presence_polylog(data &data)
@ -90,6 +114,8 @@ ircd::m::sync::presence_polylog(data &data)
m::presence::get(std::nothrow, user, closure);
}};
mitsein.for_each("join", each_user);
/*
//TODO: conf
static const size_t fibers(24);
string_view q[fibers];
@ -107,8 +133,7 @@ ircd::m::sync::presence_polylog(data &data)
q[parallel.snd] = buf[parallel.snd];
parallel();
}};
mitsein.for_each("join", paraclosure);
// mitsein.for_each("join", each_user);
*/
// mitsein.for_each("join", paraclosure);
return true;
}

227
modules/client/sync/rooms.cc Normal file
View file

@ -0,0 +1,227 @@
// Matrix Construct
//
// Copyright (C) Matrix Construct Developers, Authors & Contributors
// Copyright (C) 2016-2018 Jason Volk <jason@zemos.net>
//
// 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.
ircd::mapi::header
IRCD_MODULE
{
"Client Sync :Rooms"
};
namespace ircd::m::sync
{
static void _rooms_polylog_room(data &, json::stack::object &out, const m::room &);
static void _rooms_polylog(data &, json::stack::object &out, const string_view &membership);
static bool rooms_polylog(data &);
static bool rooms_linear(data &);
extern item rooms;
}
decltype(ircd::m::sync::rooms)
ircd::m::sync::rooms
{
"rooms",
rooms_polylog,
rooms_linear
};
bool
ircd::m::sync::rooms_linear(data &data)
{
return true;
}
bool
ircd::m::sync::rooms_polylog(data &data)
{
json::stack::object object{*data.member};
_rooms_polylog(data, object, "invite");
_rooms_polylog(data, object, "join");
_rooms_polylog(data, object, "leave");
_rooms_polylog(data, object, "ban");
return true;
}
void
ircd::m::sync::_rooms_polylog(data &data,
json::stack::object &out,
const string_view &membership)
{
const scope_restore<decltype(data.membership)> theirs
{
data.membership, membership
};
json::stack::member rooms_member
{
out, membership
};
json::stack::object rooms_object
{
rooms_member
};
data.user_rooms.for_each(membership, [&data, &rooms_object]
(const m::room &room, const string_view &membership)
{
if(head_idx(std::nothrow, room) <= data.since)
return;
// Generate individual stats for this room's sync
#ifdef RB_DEBUG
sync::stats stats{data.stats};
stats.timer = timer{};
#endif
// This scope ensures the object destructs and flushes before
// the log message tallying the stats for this room below.
{
json::stack::member member{rooms_object, room.room_id};
json::stack::object object{member};
_rooms_polylog_room(data, object, room);
}
#ifdef RB_DEBUG
thread_local char iecbuf[64], rembuf[128], tmbuf[32];
log::debug
{
log, "polylog %s %s %s %s wc:%zu in %s",
string(rembuf, ircd::remote(data.client)),
string_view{data.user.user_id},
string_view{room.room_id},
pretty(iecbuf, iec(data.stats.flush_bytes - stats.flush_bytes)),
data.stats.flush_count - stats.flush_count,
ircd::pretty(tmbuf, stats.timer.at<milliseconds>(), true)
};
#endif
});
}
void
ircd::m::sync::_rooms_polylog_room(data &data,
json::stack::object &out,
const m::room &room)
try
{
const scope_restore<decltype(data.room)> theirs
{
data.room, &room
};
// state
{
auto it(m::sync::item::map.find("rooms.$membership.$room_id.state"));
assert(it != m::sync::item::map.end());
const auto &item(it->second);
assert(item);
json::stack::member member
{
out, data.membership != "invite"?
"state":
"invite_state"
};
const scope_restore<decltype(data.member)> theirs
{
data.member, &member
};
item->polylog(data);
}
// timeline
{
auto it(m::sync::item::map.find("rooms.$membership.$room_id.timeline"));
assert(it != m::sync::item::map.end());
const auto &item(it->second);
assert(item);
json::stack::member member{out, "timeline"};
const scope_restore<decltype(data.member)> theirs
{
data.member, &member
};
item->polylog(data);
}
// ephemeral
{
auto pit
{
m::sync::item::map.equal_range("rooms...ephemeral")
};
assert(pit.first != pit.second);
json::stack::member member{out, "ephemeral"};
json::stack::object object{member};
const scope_restore<decltype(data.object)> theirs
{
data.object, &object
};
{
json::stack::member member{object, "events"};
json::stack::array array{member};
const scope_restore<decltype(data.array)> theirs
{
data.array, &array
};
for(; pit.first != pit.second; ++pit.first)
{
const auto &item(pit.first->second);
assert(item);
item->polylog(data);
}
}
}
// account_data
{
auto it(m::sync::item::map.find("rooms...account_data"));
assert(it != m::sync::item::map.end());
const auto &item(it->second);
assert(item);
json::stack::member member{out, "account_data"};
const scope_restore<decltype(data.member)> theirs
{
data.member, &member
};
item->polylog(data);
}
// unread_notifications
{
auto it(m::sync::item::map.find("rooms...unread_notifications"));
assert(it != m::sync::item::map.end());
const auto &item(it->second);
assert(item);
json::stack::member member{out, "unread_notifications"};
const scope_restore<decltype(data.member)> theirs
{
data.member, &member
};
item->polylog(data);
}
}
catch(const json::not_found &e)
{
log::critical
{
log, "polylog sync room %s error %lu to %lu (vm @ %zu) :%s"
,string_view{room.room_id}
,data.since
,data.current
,m::vm::current_sequence
,e.what()
};
}

23
modules/client/sync/rooms/state.cc
View file

@ -17,16 +17,35 @@ IRCD_MODULE
namespace ircd::m::sync
{
static bool room_state_polylog(data &);
static bool room_state_linear(data &);
extern item room_state;
}
decltype(ircd::m::sync::room_state)
ircd::m::sync::room_state
{
"rooms...state",
room_state_polylog
"rooms.$membership.$room_id.state",
room_state_polylog,
room_state_linear
};
bool
ircd::m::sync::room_state_linear(data &data)
{
assert(data.event);
assert(data.room);
assert(json::get<"room_id"_>(*data.event));
if(!json::get<"state_key"_>(*data.event))
return false;
if(!data.room->membership(data.user, data.membership))
return false;
data.array->append(*data.event);
return true;
}
bool
ircd::m::sync::room_state_polylog(data &data)
{

2
modules/client/sync/rooms/timeline.cc
View file

@ -24,7 +24,7 @@ namespace ircd::m::sync
decltype(ircd::m::sync::room_timeline)
ircd::m::sync::room_timeline
{
"rooms...timeline",
"rooms.$membership.$room_id.timeline",
room_timeline_polylog
};

10
modules/client/sync/rooms/unread_notifications.cc
View file

@ -19,6 +19,7 @@ namespace ircd::m::sync
static long _notification_count(const room &, const event::idx &a, const event::idx &b);
static long _highlight_count(const room &, const user &u, const event::idx &a, const event::idx &b);
static bool room_unread_notifications_polylog(data &);
static bool room_unread_notifications_linear(data &);
extern item room_unread_notifications;
}
@ -26,9 +27,16 @@ decltype(ircd::m::sync::room_unread_notifications)
ircd::m::sync::room_unread_notifications
{
"rooms...unread_notifications",
room_unread_notifications_polylog
room_unread_notifications_polylog,
room_unread_notifications_linear
};
bool
ircd::m::sync::room_unread_notifications_linear(data &data)
{
return true;
}
bool
ircd::m::sync::room_unread_notifications_polylog(data &data)
{