hcq/MaxScale
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

MXS-2218 Replace mq housekeeper task with delayed call

Browse Source
This commit is contained in:
Johan Wikman
2019-01-08 12:25:55 +02:00
parent e179395819
commit 29b6e53eb8
1 changed files with 11 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
server/modules/filter/mqfilter/mqfilter.cc
View File

@ -76,7 +76,7 @@
#include <maxscale/protocol/mysql.hh> #include <maxscale/protocol/mysql.hh>
#include <maxscale/query_classifier.h> #include <maxscale/query_classifier.h>
#include <maxscale/session.hh> #include <maxscale/session.hh>
#include <maxscale/housekeeper.h> #include <maxscale/mainworker.hh>
#include <maxscale/alloc.h> #include <maxscale/alloc.h>
static int uid_gen; static int uid_gen;
@ -241,7 +241,7 @@ typedef struct
bool was_query; /**True if the previous routeQuery call had valid content*/ bool was_query; /**True if the previous routeQuery call had valid content*/
} MQ_SESSION; } MQ_SESSION;
bool sendMessage(void* data); bool sendMessage(mxb::Worker::Call::action_t action, MQ_INSTANCE* instance);
static const MXS_ENUM_VALUE trigger_values[] = static const MXS_ENUM_VALUE trigger_values[] =
{ {
@ -659,9 +659,9 @@ static MXS_FILTER* createInstance(const char* name, MXS_CONFIG_PARAMETER* params
/**Connect to the server*/ /**Connect to the server*/
init_conn(my_instance); init_conn(my_instance);
char taskname[512]; mxb::Worker& worker = mxs::MainWorker::get();
snprintf(taskname, 511, "mqtask%d", atomic_add(&hktask_id, 1));
hktask_add(taskname, sendMessage, (void*)my_instance, 5); worker.delayed_call(5000, sendMessage, my_instance);
} }
return (MXS_FILTER*)my_instance; return (MXS_FILTER*)my_instance;
@ -718,9 +718,13 @@ int declareQueue(MQ_INSTANCE* my_instance, MQ_SESSION* my_session, char* qname)
* the housekeeper thread. * the housekeeper thread.
* @param data MQfilter instance * @param data MQfilter instance
*/ */
bool sendMessage(void* data) bool sendMessage(mxb::Worker::Call::action_t action, MQ_INSTANCE* instance)
{ {
MQ_INSTANCE* instance = (MQ_INSTANCE*) data; if (action == mxb::Worker::Call::CANCEL)
{
return true;
}
mqmessage* tmp; mqmessage* tmp;
int err_num = AMQP_STATUS_OK; int err_num = AMQP_STATUS_OK;