mirror of
https://github.com/matrix-construct/construct
synced 2024-11-26 00:32:35 +01:00
modules/media: Distill out download stack w/ concurrent request barrier.
This commit is contained in:
Jason Volk
parent
6add3e9952
commit
bf68626712
4 changed files with 203 additions and 169 deletions
|
|
@ -59,23 +59,10 @@ get__download(client &client,
|
|||
|
||||
const m::room::id::buf room_id
|
||||
{
|
||||
file_room_id(server, file)
|
||||
download(server, file)
|
||||
};
|
||||
|
||||
m::vm::opts::commit vmopts;
|
||||
vmopts.history = false;
|
||||
const m::room room
|
||||
{
|
||||
room_id, &vmopts
|
||||
};
|
||||
|
||||
if(m::exists(room))
|
||||
return get__download_local(client, request, server, file, room);
|
||||
|
||||
throw m::NOT_FOUND
|
||||
{
|
||||
"Media not found"
|
||||
};
|
||||
return get__download_local(client, request, server, file, room_id);
|
||||
}
|
||||
|
||||
static resource::response
|
||||
|
|
|
|||
|
|
@ -22,6 +22,179 @@ media_log
|
|||
"media"
|
||||
};
|
||||
|
||||
std::set<m::room::id>
|
||||
downloading;
|
||||
|
||||
ctx::dock
|
||||
downloading_dock;
|
||||
|
||||
m::room::id::buf
|
||||
download(const string_view &server,
|
||||
const string_view &mediaid,
|
||||
const net::hostport &remote)
|
||||
{
|
||||
const m::room::id::buf room_id
|
||||
{
|
||||
file_room_id(server, mediaid)
|
||||
};
|
||||
|
||||
download(server, mediaid, remote, room_id);
|
||||
return room_id;
|
||||
}
|
||||
|
||||
m::room
|
||||
download(const string_view &server,
|
||||
const string_view &mediaid,
|
||||
const net::hostport &remote,
|
||||
const m::room::id &room_id)
|
||||
try
|
||||
{
|
||||
auto iit
|
||||
{
|
||||
downloading.emplace(room_id)
|
||||
};
|
||||
|
||||
if(!iit.second)
|
||||
{
|
||||
do
|
||||
{
|
||||
downloading_dock.wait();
|
||||
}
|
||||
while(downloading.count(room_id));
|
||||
return room_id;
|
||||
}
|
||||
|
||||
const unwind uw{[&iit]
|
||||
{
|
||||
downloading.erase(iit.first);
|
||||
downloading_dock.notify_all();
|
||||
}};
|
||||
|
||||
if(exists(m::room{room_id}))
|
||||
return room_id;
|
||||
|
||||
const unique_buffer<mutable_buffer> buf
|
||||
{
|
||||
16_KiB
|
||||
};
|
||||
|
||||
const auto pair
|
||||
{
|
||||
download(buf, server, mediaid, remote)
|
||||
};
|
||||
|
||||
const auto &head
|
||||
{
|
||||
pair.first
|
||||
};
|
||||
|
||||
const const_buffer &content
|
||||
{
|
||||
pair.second
|
||||
};
|
||||
|
||||
char mime_type_buf[64];
|
||||
const auto &content_type
|
||||
{
|
||||
magic::mime(mime_type_buf, content)
|
||||
};
|
||||
|
||||
if(content_type != head.content_type) log::warning
|
||||
{
|
||||
media_log, "Server %s claims thumbnail %s is '%s' but we think it is '%s'",
|
||||
string(remote),
|
||||
mediaid,
|
||||
head.content_type,
|
||||
content_type
|
||||
};
|
||||
|
||||
m::vm::opts::commit vmopts;
|
||||
vmopts.history = false;
|
||||
const m::room room
|
||||
{
|
||||
room_id, &vmopts
|
||||
};
|
||||
|
||||
create(room, m::me.user_id, "file");
|
||||
|
||||
const size_t written
|
||||
{
|
||||
write_file(room, content, content_type)
|
||||
};
|
||||
|
||||
return room;
|
||||
}
|
||||
catch(const ircd::server::unavailable &e)
|
||||
{
|
||||
throw http::error
|
||||
{
|
||||
http::BAD_GATEWAY, e.what()
|
||||
};
|
||||
}
|
||||
|
||||
std::pair<http::response::head, unique_buffer<mutable_buffer>>
|
||||
download(const mutable_buffer &head_buf,
|
||||
const string_view &server,
|
||||
const string_view &mediaid,
|
||||
net::hostport remote,
|
||||
server::request::opts *const opts)
|
||||
{
|
||||
if(!remote)
|
||||
remote = server;
|
||||
|
||||
window_buffer wb{head_buf};
|
||||
thread_local char uri[4_KiB];
|
||||
http::request
|
||||
{
|
||||
wb, host(remote), "GET", fmt::sprintf
|
||||
{
|
||||
uri, "/_matrix/media/r0/download/%s/%s", server, mediaid
|
||||
}
|
||||
};
|
||||
|
||||
const const_buffer out_head
|
||||
{
|
||||
wb.completed()
|
||||
};
|
||||
|
||||
// Remaining space in buffer is used for received head
|
||||
const mutable_buffer in_head
|
||||
{
|
||||
data(head_buf) + size(out_head), size(head_buf) - size(out_head)
|
||||
};
|
||||
|
||||
//TODO: --- This should use the progress callback to build blocks
|
||||
// Null content buffer will cause dynamic allocation internally.
|
||||
const mutable_buffer in_content{};
|
||||
server::request remote_request
|
||||
{
|
||||
remote, { out_head }, { in_head, in_content }, opts
|
||||
};
|
||||
|
||||
//TODO: conf
|
||||
if(!remote_request.wait(seconds(10), std::nothrow))
|
||||
throw http::error
|
||||
{
|
||||
http::REQUEST_TIMEOUT
|
||||
};
|
||||
|
||||
const auto &code
|
||||
{
|
||||
remote_request.get()
|
||||
};
|
||||
|
||||
if(code != http::OK)
|
||||
return {};
|
||||
|
||||
parse::buffer pb{remote_request.in.head};
|
||||
parse::capstan pc{pb};
|
||||
pc.read += size(remote_request.in.head);
|
||||
return
|
||||
{
|
||||
http::response::head{pc}, std::move(remote_request.in.dynamic)
|
||||
};
|
||||
}
|
||||
|
||||
size_t
|
||||
write_file(const m::room &room,
|
||||
const const_buffer &content,
|
||||
|
|
|
|||
|
|
@ -22,5 +22,29 @@ m::room::id::buf
|
|||
file_room_id(const string_view &server,
|
||||
const string_view &file);
|
||||
|
||||
size_t read_each_block(const m::room &, const std::function<void (const const_buffer &)> &);
|
||||
size_t write_file(const m::room &room, const const_buffer &content, const string_view &content_type);
|
||||
extern "C" size_t
|
||||
read_each_block(const m::room &,
|
||||
const std::function<void (const const_buffer &)> &);
|
||||
|
||||
extern "C" size_t
|
||||
write_file(const m::room &room,
|
||||
const const_buffer &content,
|
||||
const string_view &content_type);
|
||||
|
||||
std::pair<http::response::head, unique_buffer<mutable_buffer>>
|
||||
download(const mutable_buffer &head_buf,
|
||||
const string_view &server,
|
||||
const string_view &mediaid,
|
||||
net::hostport remote = {},
|
||||
server::request::opts *const opts = nullptr);
|
||||
|
||||
m::room
|
||||
download(const string_view &server,
|
||||
const string_view &mediaid,
|
||||
const net::hostport &remote,
|
||||
const m::room::id &room_id);
|
||||
|
||||
extern "C" m::room::id::buf
|
||||
download(const string_view &server,
|
||||
const string_view &mediaid,
|
||||
const net::hostport &remote = {});
|
||||
|
|
|
|||
|
|
@ -36,14 +36,6 @@ thumbnail_resource
|
|||
}
|
||||
};
|
||||
|
||||
static resource::response
|
||||
get__thumbnail_remote(client &client,
|
||||
const resource::request &request,
|
||||
const string_view &remote,
|
||||
const string_view &server,
|
||||
const string_view &file,
|
||||
const m::room &room);
|
||||
|
||||
static resource::response
|
||||
get__thumbnail_local(client &client,
|
||||
const resource::request &request,
|
||||
|
|
@ -79,45 +71,10 @@ get__thumbnail(client &client,
|
|||
|
||||
const m::room::id::buf room_id
|
||||
{
|
||||
file_room_id(server, file)
|
||||
download(server, file)
|
||||
};
|
||||
|
||||
m::vm::opts::commit vmopts;
|
||||
vmopts.history = false;
|
||||
const m::room room
|
||||
{
|
||||
room_id, &vmopts
|
||||
};
|
||||
|
||||
//TODO: ABA
|
||||
if(m::exists(room))
|
||||
return get__thumbnail_local(client, request, server, file, room);
|
||||
|
||||
//TODO: XXX conf
|
||||
//TODO: XXX vector
|
||||
static const string_view secondary
|
||||
{
|
||||
"matrix.org"
|
||||
};
|
||||
|
||||
const string_view &remote
|
||||
{
|
||||
my_host(server)?
|
||||
secondary:
|
||||
server
|
||||
};
|
||||
|
||||
if(!my_host(remote))
|
||||
{
|
||||
//TODO: ABA TXN
|
||||
create(room, m::me.user_id, "file");
|
||||
return get__thumbnail_remote(client, request, remote, server, file, room);
|
||||
}
|
||||
|
||||
throw m::NOT_FOUND
|
||||
{
|
||||
"Media not found"
|
||||
};
|
||||
return get__thumbnail_local(client, request, server, file, room_id);
|
||||
}
|
||||
|
||||
static resource::method
|
||||
|
|
@ -132,113 +89,6 @@ method_get
|
|||
thumbnail_resource, "GET", get__thumbnail
|
||||
};
|
||||
|
||||
static resource::response
|
||||
get__thumbnail_remote(client &client,
|
||||
const resource::request &request,
|
||||
const string_view &hostname,
|
||||
const string_view &server,
|
||||
const string_view &mediaid,
|
||||
const m::room &room)
|
||||
try
|
||||
{
|
||||
const net::hostport remote
|
||||
{
|
||||
hostname
|
||||
};
|
||||
|
||||
const unique_buffer<mutable_buffer> buf
|
||||
{
|
||||
16_KiB
|
||||
};
|
||||
|
||||
window_buffer wb{buf};
|
||||
thread_local char uri[4_KiB];
|
||||
http::request
|
||||
{
|
||||
wb, hostname, "GET", fmt::sprintf
|
||||
{
|
||||
uri, "/_matrix/media/r0/download/%s/%s", server, mediaid
|
||||
}
|
||||
};
|
||||
|
||||
const const_buffer out_head
|
||||
{
|
||||
wb.completed()
|
||||
};
|
||||
|
||||
// Remaining space in buffer is used for received head
|
||||
const mutable_buffer in_head
|
||||
{
|
||||
data(buf) + size(out_head), size(buf) - size(out_head)
|
||||
};
|
||||
|
||||
//TODO: --- This should use the progress callback to build blocks
|
||||
|
||||
// Null content buffer will cause dynamic allocation internally.
|
||||
const mutable_buffer in_content{};
|
||||
|
||||
struct server::request::opts opts;
|
||||
server::request remote_request
|
||||
{
|
||||
remote, { out_head }, { in_head, in_content }, &opts
|
||||
};
|
||||
|
||||
if(!remote_request.wait(seconds(10), std::nothrow))
|
||||
throw http::error
|
||||
{
|
||||
http::REQUEST_TIMEOUT
|
||||
};
|
||||
|
||||
//TODO: ---
|
||||
|
||||
const auto &code
|
||||
{
|
||||
remote_request.get()
|
||||
};
|
||||
|
||||
char mime_type_buf[64];
|
||||
const string_view content_type
|
||||
{
|
||||
magic::mime(mime_type_buf, remote_request.in.content)
|
||||
};
|
||||
|
||||
const size_t file_size
|
||||
{
|
||||
size(remote_request.in.content)
|
||||
};
|
||||
|
||||
parse::buffer pb{remote_request.in.head};
|
||||
parse::capstan pc{pb};
|
||||
pc.read += size(remote_request.in.head);
|
||||
const http::response::head head{pc};
|
||||
if(content_type != head.content_type)
|
||||
log::warning
|
||||
{
|
||||
"Server %s claims thumbnail %s is '%s' but we think it is '%s'",
|
||||
hostname,
|
||||
mediaid,
|
||||
head.content_type,
|
||||
content_type
|
||||
};
|
||||
|
||||
const size_t written
|
||||
{
|
||||
write_file(room, remote_request.in.content, content_type)
|
||||
};
|
||||
|
||||
return resource::response
|
||||
{
|
||||
client, remote_request.in.content, content_type
|
||||
};
|
||||
}
|
||||
catch(const ircd::server::unavailable &e)
|
||||
{
|
||||
throw http::error
|
||||
{
|
||||
http::BAD_GATEWAY, e.what()
|
||||
};
|
||||
}
|
||||
|
||||
static resource::response
|
||||
get__thumbnail_local(client &client,
|
||||
const resource::request &request,
|
||||
|
|
|
|||
Loading…
Reference in a new issue