mirrors/mumble
2
0
Fork 0
mirror of https://github.com/mumble-voip/mumble.git synced 2025-10-26 11:19:16 +00:00
Code Issues Packages Projects Releases Wiki Activity

GRPC: use AsyncNotifyWhenDone instead of cleanup timer

Browse Source
This commit is contained in:
Jonas Herzig 2018-01-07 22:45:25 +01:00
parent 76eb586f8d
commit 92cd615281
3 changed files with 72 additions and 61 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

112
src/murmur/MurmurGRPCImpl.cpp
View File

@ -132,76 +132,19 @@ void GRPCStop() {
}
}
MurmurRPCImpl::MurmurRPCImpl(const QString &address, std::shared_ptr<::grpc::ServerCredentials> credentials) : m_cleanupTimer(this) {
MurmurRPCImpl::MurmurRPCImpl(const QString &address, std::shared_ptr<::grpc::ServerCredentials> credentials) {
::grpc::ServerBuilder builder;
builder.AddListeningPort(u8(address), credentials);
builder.RegisterService(&m_V1Service);
m_completionQueue = builder.AddCompletionQueue();
m_server = builder.BuildAndStart();
meta->connectListener(this);
connect(&m_cleanupTimer, SIGNAL(timeout()), this, SLOT(cleanup()));
m_cleanupTimer.setSingleShot(false);
m_cleanupTimer.start(1000 * 60);
start();
}
MurmurRPCImpl::~MurmurRPCImpl() {
}
// This function periodically runs to clean up disconnected listeners. We need
// this because (as of 2015-07-21) the grpc library does not tell us when a
// client disconnects.
void MurmurRPCImpl::cleanup() {
for (auto i = m_metaServiceListeners.begin(); i != m_metaServiceListeners.end(); ) {
auto listener = *i;
if (listener->context.IsCancelled()) {
i = m_metaServiceListeners.erase(i);
listener->deref();
} else {
++i;
}
}
for (auto i = m_contextActionListeners.begin(); i != m_contextActionListeners.end(); ) {
auto &ref = i.value();
for (auto j = ref.begin(); j != ref.end(); ) {
auto listener = j.value();
if (listener->context.IsCancelled()) {
j = ref.erase(j);
listener->deref();
} else {
++j;
}
}
if (ref.isEmpty()) {
i = m_contextActionListeners.erase(i);
} else {
++i;
}
}
for (auto i = m_serverServiceListeners.begin(); i != m_serverServiceListeners.end(); ) {
auto listener = i.value();
if (listener->context.IsCancelled()) {
i = m_serverServiceListeners.erase(i);
listener->deref();
} else {
++i;
}
}
for (auto i = m_authenticators.begin(); i != m_authenticators.end(); ) {
auto listener = i.value();
if (listener->context.IsCancelled()) {
i = m_authenticators.erase(i);
listener->deref();
} else {
++i;
}
}
}
// ToRPC/FromRPC methods convert data to/from grpc protocol buffer messages.
void ToRPC(const ::Server *srv, const ::Channel *c, ::MurmurRPC::Channel *rc) {
rc->mutable_server()->set_id(srv->iServerNum);
@ -1265,6 +1208,15 @@ void V1_ServerEvents::impl(bool) {
rpc->m_serverServiceListeners.insert(server->iServerNum, this);
}
void V1_ServerEvents::done(bool) {
auto &ssls = rpc->m_serverServiceListeners;
auto i = std::find(ssls.begin(), ssls.end(), this);
if (i != ssls.end()) {
ssls.erase(i);
}
deref();
}
void V1_GetUptime::impl(bool) {
::MurmurRPC::Uptime uptime;
uptime.set_secs(meta->tUptime.elapsed()/1000000LL);
@ -1285,6 +1237,15 @@ void V1_Events::impl(bool) {
rpc->m_metaServiceListeners.insert(this);
}
void V1_Events::done(bool) {
auto &msls = rpc->m_metaServiceListeners;
auto i = std::find(msls.begin(), msls.end(), this);
if (i != msls.end()) {
msls.erase(i);
}
deref();
}
void V1_ContextActionAdd::impl(bool) {
auto server = MustServer(request);
auto user = MustUser(server, request);
@ -1350,6 +1311,24 @@ void V1_ContextActionEvents::impl(bool) {
rpc->m_contextActionListeners[server->iServerNum].insert(u8(request.action()), this);
}
void V1_ContextActionEvents::done(bool) {
auto server = MustServer(request);
auto &cals = rpc->m_contextActionListeners;
auto &scals = cals[server->iServerNum];
auto i = std::find(scals.begin(), scals.end(), this);
if (i != scals.end()) {
scals.erase(i);
}
deref();
if (scals.isEmpty()) {
auto i = std::find(cals.begin(), cals.end(), scals);
if (i != cals.end()) {
cals.erase(i);
}
}
}
void V1_TextMessageSend::impl(bool) {
auto server = MustServer(request);
@ -1393,6 +1372,15 @@ void V1_TextMessageFilter::impl(bool) {
stream.Read(&request, callback(onInitialize));
}
void V1_TextMessageFilter::done(bool) {
auto server = MustServer(request);
auto filter = rpc->m_textMessageFilters.value(server->iServerNum);
if (filter == this) {
rpc->m_textMessageFilters.remove(server->iServerNum);
}
deref();
}
void V1_LogQuery::impl(bool) {
auto serverID = MustServerID(request);
@ -1981,6 +1969,14 @@ void V1_AuthenticatorStream::impl(bool) {
stream.Read(&request, callback(onInitialize));
}
void V1_AuthenticatorStream::done(bool) {
auto i = std::find(rpc->m_authenticators.begin(), rpc->m_authenticators.end(), this);
if (i != rpc->m_authenticators.end()) {
rpc->m_authenticators.erase(i);
}
deref();
}
void V1_DatabaseUserQuery::impl(bool) {
auto server = MustServer(request);

3
src/murmur/MurmurGRPCImpl.h
View File

@ -36,7 +36,6 @@ namespace MurmurRPC {
class MurmurRPCImpl : public QThread {
Q_OBJECT;
std::unique_ptr<grpc::Server> m_server;
QTimer m_cleanupTimer;
protected:
void customEvent(QEvent *evt);
public:
@ -76,8 +75,6 @@ class MurmurRPCImpl : public QThread {
void sendServerEvent(const ::Server *s, const ::MurmurRPC::Server_Event &e);
public slots:
void cleanup();
void started(Server *server);
void stopped(Server *server);

18
src/murmur_grpcwrapper_protoc_plugin/main.cpp
View File

@ -54,6 +54,7 @@ public:
}
void impl(bool ok);
void done(bool ok);
::boost::function<void(bool)> *callback(::boost::function<void($service$_$method$ *, bool)> cb) {
auto fn = ::boost::bind(&$service$_$method$::callbackAction, this, cb, _1);
@ -66,8 +67,16 @@ public:
QCoreApplication::instance()->postEvent(rpc, ie);
}
void handleDone(bool ok) {
auto ie = new RPCExecEvent(::boost::bind(&$service$_$method$::done, this, ok), this);
QCoreApplication::instance()->postEvent(rpc, ie);
}
static void create(MurmurRPCImpl *rpc, ::$ns$::$service$::AsyncService *service) {
auto call = new $service$_$method$(rpc, service);
auto done_fn = ::boost::bind(&$service$_$method$::handleDone, call, _1);
auto done_fn_ptr = new ::boost::function<void(bool)>(done_fn);
call->context.AsyncNotifyWhenDone(done_fn_ptr);
auto fn = ::boost::bind(&$service$_$method$::handle, call, _1);
auto fn_ptr = new ::boost::function<void(bool)>(fn);
service->Request$method$(&call->context, &call->request, &call->stream, rpc->m_completionQueue.get(), rpc->m_completionQueue.get(), fn_ptr);
@ -122,6 +131,7 @@ public:
}
void impl(bool ok);
void done(bool ok);
bool write() {
bool processed = false;
@ -166,8 +176,16 @@ public:
QCoreApplication::instance()->postEvent(rpc, ie);
}
void handleDone(bool ok) {
auto ie = new RPCExecEvent(::boost::bind(&$service$_$method$::done, this, ok), this);
QCoreApplication::instance()->postEvent(rpc, ie);
}
static void create(MurmurRPCImpl *rpc, ::$ns$::$service$::AsyncService *service) {
auto call = new $service$_$method$(rpc, service);
auto done_fn = ::boost::bind(&$service$_$method$::handleDone, call, _1);
auto done_fn_ptr = new ::boost::function<void(bool)>(done_fn);
call->context.AsyncNotifyWhenDone(done_fn_ptr);
auto fn = ::boost::bind(&$service$_$method$::handle, call, _1);
auto fn_ptr = new ::boost::function<void(bool)>(fn);
service->Request$method$(&call->context, &call->stream, rpc->m_completionQueue.get(), rpc->m_completionQueue.get(), fn_ptr);