1 Star 0 Fork 345

wukong / swoole-src

forked from swoole / swoole-src 
加入 Gitee
与超过 1200万 开发者一起发现、参与优秀开源项目,私有仓库也完全免费 :)
免费加入
克隆/下载
swoole_mysql_coro.cc 61.07 KB
一键复制 编辑 原始数据 按行查看 历史
韩天峰 提交于 2019-01-18 18:27 . remove SW_MAKE_STD_ZVAL
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968
/*
+----------------------------------------------------------------------+
| Swoole |
+----------------------------------------------------------------------+
| Copyright (c) 2012-2018 The Swoole Group |
+----------------------------------------------------------------------+
| This source file is subject to version 2.0 of the Apache license, |
| that is bundled with this package in the file LICENSE, and is |
| available through the world-wide-web at the following url: |
| http://www.apache.org/licenses/LICENSE-2.0.html |
| If you did not receive a copy of the Apache2.0 license and are unable|
| to obtain it through the world-wide-web, please send a note to |
| license@swoole.com so we can mail you a copy immediately. |
+----------------------------------------------------------------------+
| Author: Tianfeng Han <mikan.tenny@gmail.com> |
+----------------------------------------------------------------------+
*/
#include "php_swoole_cxx.h"
#include "swoole_coroutine.h"
#include "swoole_mysql.h"
using namespace swoole;
static PHP_METHOD(swoole_mysql_coro, __construct);
static PHP_METHOD(swoole_mysql_coro, __destruct);
static PHP_METHOD(swoole_mysql_coro, connect);
static PHP_METHOD(swoole_mysql_coro, query);
static PHP_METHOD(swoole_mysql_coro, recv);
static PHP_METHOD(swoole_mysql_coro, nextResult);
#ifdef SW_USE_MYSQLND
static PHP_METHOD(swoole_mysql_coro, escape);
#endif
static PHP_METHOD(swoole_mysql_coro, begin);
static PHP_METHOD(swoole_mysql_coro, commit);
static PHP_METHOD(swoole_mysql_coro, rollback);
static PHP_METHOD(swoole_mysql_coro, prepare);
static PHP_METHOD(swoole_mysql_coro, setDefer);
static PHP_METHOD(swoole_mysql_coro, getDefer);
static PHP_METHOD(swoole_mysql_coro, close);
static PHP_METHOD(swoole_mysql_coro_statement, __destruct);
static PHP_METHOD(swoole_mysql_coro_statement, execute);
static PHP_METHOD(swoole_mysql_coro_statement, fetch);
static PHP_METHOD(swoole_mysql_coro_statement, fetchAll);
static PHP_METHOD(swoole_mysql_coro_statement, nextResult);
ZEND_BEGIN_ARG_INFO_EX(arginfo_swoole_void, 0, 0, 0)
ZEND_END_ARG_INFO()
ZEND_BEGIN_ARG_INFO_EX(arginfo_swoole_mysql_coro_connect, 0, 0, 1)
ZEND_ARG_ARRAY_INFO(0, server_config, 0)
ZEND_END_ARG_INFO()
ZEND_BEGIN_ARG_INFO_EX(arginfo_swoole_mysql_coro_query, 0, 0, 1)
ZEND_ARG_INFO(0, sql)
ZEND_ARG_INFO(0, timeout)
ZEND_END_ARG_INFO()
ZEND_BEGIN_ARG_INFO_EX(arginfo_swoole_mysql_coro_begin, 0, 0, 0)
ZEND_ARG_INFO(0, timeout)
ZEND_END_ARG_INFO()
ZEND_BEGIN_ARG_INFO_EX(arginfo_swoole_mysql_coro_commit, 0, 0, 0)
ZEND_ARG_INFO(0, timeout)
ZEND_END_ARG_INFO()
ZEND_BEGIN_ARG_INFO_EX(arginfo_swoole_mysql_coro_rollback, 0, 0, 0)
ZEND_ARG_INFO(0, timeout)
ZEND_END_ARG_INFO()
ZEND_BEGIN_ARG_INFO_EX(arginfo_swoole_mysql_coro_prepare, 0, 0, 1)
ZEND_ARG_INFO(0, query)
ZEND_ARG_INFO(0, timeout)
ZEND_END_ARG_INFO()
ZEND_BEGIN_ARG_INFO_EX(arginfo_swoole_mysql_coro_setDefer, 0, 0, 0)
ZEND_ARG_INFO(0, defer)
ZEND_END_ARG_INFO()
#ifdef SW_USE_MYSQLND
ZEND_BEGIN_ARG_INFO_EX(arginfo_swoole_mysql_coro_escape, 0, 0, 1)
ZEND_ARG_INFO(0, string)
ZEND_ARG_INFO(0, flags)
ZEND_END_ARG_INFO()
#endif
ZEND_BEGIN_ARG_INFO_EX(arginfo_swoole_mysql_coro_statement_execute, 0, 0, 0)
ZEND_ARG_INFO(0, params)
ZEND_ARG_INFO(0, timeout)
ZEND_END_ARG_INFO()
ZEND_BEGIN_ARG_INFO_EX(arginfo_swoole_mysql_coro_statement_fetch, 0, 0, 0)
ZEND_END_ARG_INFO()
ZEND_BEGIN_ARG_INFO_EX(arginfo_swoole_mysql_coro_statement_fetchAll, 0, 0, 0)
ZEND_END_ARG_INFO()
ZEND_BEGIN_ARG_INFO_EX(arginfo_swoole_mysql_coro_statement_nextResult, 0, 0, 0)
ZEND_END_ARG_INFO()
static zend_class_entry swoole_mysql_coro_ce;
static zend_class_entry *swoole_mysql_coro_ce_ptr;
static zend_object_handlers swoole_mysql_coro_handlers;
static zend_class_entry swoole_mysql_coro_exception_ce;
static zend_class_entry *swoole_mysql_coro_exception_ce_ptr;
static zend_object_handlers swoole_mysql_coro_exception_handlers;
static zend_class_entry swoole_mysql_coro_statement_ce;
static zend_class_entry *swoole_mysql_coro_statement_ce_ptr;
static zend_object_handlers swoole_mysql_coro_statement_handlers;
static const zend_function_entry swoole_mysql_coro_methods[] =
{
PHP_ME(swoole_mysql_coro, __construct, arginfo_swoole_void, ZEND_ACC_PUBLIC)
PHP_ME(swoole_mysql_coro, __destruct, arginfo_swoole_void, ZEND_ACC_PUBLIC)
PHP_ME(swoole_mysql_coro, connect, arginfo_swoole_mysql_coro_connect, ZEND_ACC_PUBLIC)
PHP_ME(swoole_mysql_coro, query, arginfo_swoole_mysql_coro_query, ZEND_ACC_PUBLIC)
PHP_ME(swoole_mysql_coro, recv, arginfo_swoole_void, ZEND_ACC_PUBLIC)
PHP_ME(swoole_mysql_coro, nextResult, arginfo_swoole_void, ZEND_ACC_PUBLIC)
#ifdef SW_USE_MYSQLND
PHP_ME(swoole_mysql_coro, escape, arginfo_swoole_mysql_coro_escape, ZEND_ACC_PUBLIC)
#endif
PHP_ME(swoole_mysql_coro, begin, arginfo_swoole_mysql_coro_begin, ZEND_ACC_PUBLIC)
PHP_ME(swoole_mysql_coro, commit, arginfo_swoole_mysql_coro_commit, ZEND_ACC_PUBLIC)
PHP_ME(swoole_mysql_coro, rollback, arginfo_swoole_mysql_coro_rollback, ZEND_ACC_PUBLIC)
PHP_ME(swoole_mysql_coro, prepare, arginfo_swoole_mysql_coro_prepare, ZEND_ACC_PUBLIC)
PHP_ME(swoole_mysql_coro, setDefer, arginfo_swoole_mysql_coro_setDefer, ZEND_ACC_PUBLIC)
PHP_ME(swoole_mysql_coro, getDefer, arginfo_swoole_void, ZEND_ACC_PUBLIC)
PHP_ME(swoole_mysql_coro, close, arginfo_swoole_void, ZEND_ACC_PUBLIC)
PHP_FE_END
};
static const zend_function_entry swoole_mysql_coro_statement_methods[] =
{
PHP_ME(swoole_mysql_coro_statement, execute, arginfo_swoole_mysql_coro_statement_execute, ZEND_ACC_PUBLIC)
PHP_ME(swoole_mysql_coro_statement, fetch, arginfo_swoole_mysql_coro_statement_fetch, ZEND_ACC_PUBLIC)
PHP_ME(swoole_mysql_coro_statement, fetchAll, arginfo_swoole_mysql_coro_statement_fetchAll, ZEND_ACC_PUBLIC)
PHP_ME(swoole_mysql_coro_statement, nextResult, arginfo_swoole_mysql_coro_statement_nextResult, ZEND_ACC_PUBLIC)
PHP_ME(swoole_mysql_coro_statement, __destruct, arginfo_swoole_void, ZEND_ACC_PUBLIC)
PHP_FE_END
};
static int swoole_mysql_coro_onRead(swReactor *reactor, swEvent *event);
static int swoole_mysql_coro_onWrite(swReactor *reactor, swEvent *event);
static int swoole_mysql_coro_onError(swReactor *reactor, swEvent *event);
static void swoole_mysql_coro_onConnect(mysql_client *client);
static void swoole_mysql_coro_onTimeout(swTimer *timer, swTimer_node *tnode);
static void swoole_mysql_coro_free_object(zend_object *object);
static zend_object *swoole_mysql_coro_create_object(zend_class_entry *ce)
{
zend_object *object;
object = zend_objects_new(ce);
object->handlers = &swoole_mysql_coro_handlers;
object_properties_init(object, ce);
PHPCoroutine::check();
mysql_client *client = (mysql_client *) emalloc(sizeof(mysql_client));
bzero(client, sizeof(mysql_client));
swoole_set_object_by_handle(object->handle, client);
return object;
}
void swoole_mysql_coro_init(int module_number)
{
SWOOLE_INIT_CLASS_ENTRY(swoole_mysql_coro, "Swoole\\Coroutine\\MySQL", NULL, "Co\\MySQL", swoole_mysql_coro_methods);
SWOOLE_SET_CLASS_SERIALIZABLE(swoole_mysql_coro, zend_class_serialize_deny, zend_class_unserialize_deny);
SWOOLE_SET_CLASS_CLONEABLE(swoole_mysql_coro, zend_class_clone_deny);
SWOOLE_SET_CLASS_UNSET_PROPERTY_HANDLER(swoole_mysql_coro, zend_class_unset_property_deny);
SWOOLE_SET_CLASS_CREATE_AND_FREE(swoole_mysql_coro, swoole_mysql_coro_create_object, swoole_mysql_coro_free_object);
SWOOLE_INIT_CLASS_ENTRY(swoole_mysql_coro_statement, "Swoole\\Coroutine\\MySQL\\Statement", NULL, "Co\\MySQL\\Statement", swoole_mysql_coro_statement_methods);
SWOOLE_SET_CLASS_SERIALIZABLE(swoole_mysql_coro_statement, zend_class_serialize_deny, zend_class_unserialize_deny);
SWOOLE_SET_CLASS_CLONEABLE(swoole_mysql_coro_statement, zend_class_clone_deny);
SWOOLE_SET_CLASS_UNSET_PROPERTY_HANDLER(swoole_mysql_coro_statement, zend_class_unset_property_deny);
SWOOLE_INIT_CLASS_ENTRY_EX(swoole_mysql_coro_exception, "Swoole\\Coroutine\\MySQL\\Exception", NULL, "Co\\MySQL\\Exception", NULL, swoole_exception);
SWOOLE_SET_CLASS_SERIALIZABLE(swoole_mysql_coro_exception, zend_class_serialize_deny, zend_class_unserialize_deny);
SWOOLE_SET_CLASS_CLONEABLE(swoole_mysql_coro_exception, zend_class_clone_deny);
SWOOLE_SET_CLASS_UNSET_PROPERTY_HANDLER(swoole_mysql_coro_exception, zend_class_unset_property_deny);
zend_declare_property_string(swoole_mysql_coro_ce_ptr, ZEND_STRL("serverInfo"), "", ZEND_ACC_PUBLIC);
zend_declare_property_long(swoole_mysql_coro_ce_ptr, ZEND_STRL("sock"), -1, ZEND_ACC_PUBLIC);
zend_declare_property_bool(swoole_mysql_coro_ce_ptr, ZEND_STRL("connected"), 0, ZEND_ACC_PUBLIC);
zend_declare_property_string(swoole_mysql_coro_ce_ptr, ZEND_STRL("connect_error"), "", ZEND_ACC_PUBLIC);
zend_declare_property_long(swoole_mysql_coro_ce_ptr, ZEND_STRL("connect_errno"), 0, ZEND_ACC_PUBLIC);
zend_declare_property_long(swoole_mysql_coro_ce_ptr, ZEND_STRL("affected_rows"), 0, ZEND_ACC_PUBLIC);
zend_declare_property_long(swoole_mysql_coro_ce_ptr, ZEND_STRL("insert_id"), 0, ZEND_ACC_PUBLIC);
zend_declare_property_string(swoole_mysql_coro_ce_ptr, ZEND_STRL("error"), "", ZEND_ACC_PUBLIC);
zend_declare_property_long(swoole_mysql_coro_ce_ptr, ZEND_STRL("errno"), 0, ZEND_ACC_PUBLIC);
zend_declare_property_long(swoole_mysql_coro_statement_ce_ptr, ZEND_STRL("affected_rows"), 0, ZEND_ACC_PUBLIC);
zend_declare_property_long(swoole_mysql_coro_statement_ce_ptr, ZEND_STRL("insert_id"), 0, ZEND_ACC_PUBLIC);
zend_declare_property_string(swoole_mysql_coro_statement_ce_ptr, ZEND_STRL("error"), "", ZEND_ACC_PUBLIC);
zend_declare_property_long(swoole_mysql_coro_statement_ce_ptr, ZEND_STRL("errno"), 0, ZEND_ACC_PUBLIC);
}
int mysql_query(zval *zobject, mysql_client *client, swString *sql, zval *callback);
static int swoole_mysql_coro_execute(zval *zobject, mysql_client *client, zval *params)
{
PHPCoroutine::check_bind("mysql client", client->cid);
if (!client->cli)
{
swoole_php_fatal_error(E_WARNING, "mysql connection#%d is closed.", client->fd);
return SW_ERR;
}
if (client->state != SW_MYSQL_STATE_QUERY)
{
swoole_php_fatal_error(E_WARNING, "mysql client is waiting response, cannot send new sql query.");
return SW_ERR;
}
mysql_statement *statement = (mysql_statement *) swoole_get_object(zobject);
if (!statement)
{
swoole_php_fatal_error(E_WARNING, "mysql preparation is not ready.");
return SW_ERR;
}
long lval;
char buf[10];
zval *value;
uint16_t param_count = 0;
if (params)
{
param_count = php_swoole_array_length(params);
}
if (param_count != statement->param_count)
{
swoole_php_fatal_error(E_WARNING, "mysql statement#%u expects %u parameter, %u given.", statement->id, statement->param_count, param_count);
return SW_ERR;
}
swString_clear(mysql_request_buffer);
client->cmd = SW_MYSQL_COM_STMT_EXECUTE;
client->statement = statement;
bzero(mysql_request_buffer->str, 5);
//command
mysql_request_buffer->str[4] = SW_MYSQL_COM_STMT_EXECUTE;
mysql_request_buffer->length = 5;
char *p = mysql_request_buffer->str;
p += 5;
// stmt.id
mysql_int4store(p, statement->id);
p += 4;
// flags = CURSOR_TYPE_NO_CURSOR
mysql_int1store(p, 0);
p += 1;
// iteration_count
mysql_int4store(p, 1);
p += 4;
mysql_request_buffer->length += 9;
if (param_count != 0)
{
//null bitmap
size_t null_start_offset = p - mysql_request_buffer->str;
unsigned int map_size = (param_count + 7) / 8;
memset(p, 0, map_size);
p += map_size;
mysql_request_buffer->length += map_size;
//rebind
mysql_int1store(p, 1);
p += 1;
mysql_request_buffer->length += 1;
size_t type_start_offset = p - mysql_request_buffer->str;
p += param_count * 2;
mysql_request_buffer->length += param_count * 2;
zend_ulong index = 0;
ZEND_HASH_FOREACH_VAL(Z_ARRVAL_P(params), value)
{
if (ZVAL_IS_NULL(value))
{
*((mysql_request_buffer->str + null_start_offset) + (index / 8)) |= (1UL << (index % 8));
mysql_int2store((mysql_request_buffer->str + type_start_offset) + (index * 2), SW_MYSQL_TYPE_NULL);
}
else
{
mysql_int2store((mysql_request_buffer->str + type_start_offset) + (index * 2), SW_MYSQL_TYPE_VAR_STRING);
zend::string str_value(value);
if (str_value.len() > 0xffff)
{
buf[0] = (char) SW_MYSQL_TYPE_VAR_STRING;
if (swString_append_ptr(mysql_request_buffer, buf, 1) < 0)
{
return SW_ERR;
}
}
else if (str_value.len() > 250)
{
buf[0] = (char) SW_MYSQL_TYPE_BLOB;
if (swString_append_ptr(mysql_request_buffer, buf, 1) < 0)
{
return SW_ERR;
}
}
lval = mysql_write_lcb(buf, str_value.len());
if (swString_append_ptr(mysql_request_buffer, buf, lval) < 0)
{
return SW_ERR;
}
if (swString_append_ptr(mysql_request_buffer, str_value.val(), str_value.len()) < 0)
{
return SW_ERR;
}
}
index++;
}
ZEND_HASH_FOREACH_END();
}
//length
mysql_pack_length(mysql_request_buffer->length - 4, mysql_request_buffer->str);
//send data
if (SwooleG.main_reactor->write(SwooleG.main_reactor, client->fd, mysql_request_buffer->str, mysql_request_buffer->length) < 0)
{
//connection is closed
if (swConnection_error(errno) == SW_CLOSE)
{
zend_update_property_bool(swoole_mysql_coro_ce_ptr, zobject, ZEND_STRL("connected"), 0);
zend_update_property_long(swoole_mysql_coro_ce_ptr, zobject, ZEND_STRL("errno"), 2013);
zend_update_property_string(swoole_mysql_coro_ce_ptr, zobject, ZEND_STRL("error"), "Lost connection to MySQL server during query");
}
return SW_ERR;
}
else
{
client->state = SW_MYSQL_STATE_READ_START;
return SW_OK;
}
}
static int swoole_mysql_coro_parse_response(mysql_client *client, zval **result, int from_next_result)
{
zval *zobject = client->object;
int ret = mysql_response(client);
if (ret < 0)
{
if (ret == SW_AGAIN)
{
return SW_AGAIN;
}
else // handler error
{
static const char* errmsg = "mysql response packet parse error.";
client->response.response_type = SW_MYSQL_PACKET_ERR;
client->response.error_code = ret;
client->response.server_msg = (char *) errmsg;
client->response.l_server_msg = strlen(errmsg);
if (client->response.result_array)
{
sw_zval_free(client->response.result_array);
client->response.result_array = nullptr;
}
if (client->cmd == SW_MYSQL_COM_STMT_EXECUTE)
{
if (client->statement && client->statement->result)
{
sw_zval_free(client->statement->result);
client->statement->result = NULL;
}
}
}
}
//remove from eventloop
//reactor->del(reactor, event->fd);
zend_update_property_long(
swoole_mysql_coro_ce_ptr, zobject,
ZEND_STRL("affected_rows"), client->response.affected_rows
);
zend_update_property_long(
swoole_mysql_coro_ce_ptr, zobject,
ZEND_STRL("insert_id"), client->response.insert_id
);
if (client->cmd == SW_MYSQL_COM_STMT_EXECUTE)
{
zend_update_property_long(
swoole_mysql_coro_statement_ce_ptr, client->statement->object,
ZEND_STRL("affected_rows"), client->response.affected_rows
);
zend_update_property_long(
swoole_mysql_coro_statement_ce_ptr, client->statement->object,
ZEND_STRL("insert_id"), client->response.insert_id
);
}
client->state = SW_MYSQL_STATE_QUERY;
// OK
if (client->response.response_type == SW_MYSQL_PACKET_OK)
{
*result = sw_malloc_zval();
// prepare finished and create statement
if (client->cmd == SW_MYSQL_COM_STMT_PREPARE)
{
if (client->statement_list == NULL)
{
client->statement_list = swLinkedList_new(0, NULL);
}
swLinkedList_append(client->statement_list, client->statement);
object_init_ex(*result, swoole_mysql_coro_statement_ce_ptr);
swoole_set_object(*result, client->statement);
client->statement->object = sw_zval_dup(*result);
}
else
{
if (from_next_result)
{
// pass the ok response ret val
ZVAL_NULL(*result);
}
else
{
ZVAL_TRUE(*result);
}
}
}
// ERROR
else if (client->response.response_type == SW_MYSQL_PACKET_ERR)
{
*result = sw_malloc_zval();
ZVAL_FALSE(*result);
zend_update_property_stringl(
swoole_mysql_coro_ce_ptr, zobject, ZEND_STRL("error"),
client->response.server_msg, client->response.l_server_msg
);
zend_update_property_long(
swoole_mysql_coro_ce_ptr, zobject, ZEND_STRL("errno"),
client->response.error_code
);
if (client->cmd == SW_MYSQL_COM_STMT_EXECUTE)
{
zend_update_property_stringl(
swoole_mysql_coro_statement_ce_ptr, client->statement->object,
ZEND_STRL("error"), client->response.server_msg, client->response.l_server_msg
);
zend_update_property_long(
swoole_mysql_coro_statement_ce_ptr, client->statement->object,
ZEND_STRL("errno"), client->response.error_code
);
}
}
// ResultSet
else
{
if (client->connector.fetch_mode && client->cmd == SW_MYSQL_COM_STMT_EXECUTE)
{
if (client->statement->result)
{
// free the last one
sw_zval_free(client->statement->result);
client->statement->result = NULL;
}
// save result on statement and wait for fetch
client->statement->result = client->response.result_array;
client->response.result_array = NULL;
// return true (success)]
*result = sw_malloc_zval();
ZVAL_TRUE(*result);
}
else
{
*result = client->response.result_array;
}
}
return ret;
}
static void swoole_mysql_coro_parse_end(mysql_client *client, swString *buffer)
{
if (client->response.status_code & SW_MYSQL_SERVER_MORE_RESULTS_EXISTS)
{
swTraceLog(SW_TRACE_MYSQL_CLIENT, "remaining %ju, more results exists", (uintmax_t) (buffer->length - buffer->offset));
}
else
{
// no more, clean up
swString_clear(buffer);
}
bzero(&client->response, sizeof(client->response));
client->statement = NULL;
client->cmd = SW_MYSQL_COM_NULL;
}
static int swoole_mysql_coro_statement_free(mysql_statement *stmt)
{
if (stmt->object)
{
swoole_set_object(stmt->object, NULL);
efree(stmt->object);
}
if (stmt->buffer)
{
swString_free(stmt->buffer);
}
if (stmt->result)
{
sw_zval_free(stmt->result);
}
return SW_OK;
}
static int swoole_mysql_coro_statement_close(mysql_statement *stmt)
{
// WARNING: it's wrong operation, we send the close statement packet silently, don't change any property in the client!
// stmt->client->cmd = SW_MYSQL_COM_STMT_CLOSE;
// call mysql-server to destruct this statement
swString_clear(mysql_request_buffer);
bzero(mysql_request_buffer->str, 5);
//command
mysql_request_buffer->str[4] = SW_MYSQL_COM_STMT_CLOSE;
mysql_request_buffer->length = 5;
char *p = mysql_request_buffer->str;
p += 5;
// stmt.id
mysql_int4store(p, stmt->id);
p += 4;
mysql_request_buffer->length += 4;
//length
mysql_pack_length(mysql_request_buffer->length - 4, mysql_request_buffer->str);
//tell sever to close the statement, mysql-server would not reply
SwooleG.main_reactor->write(SwooleG.main_reactor, stmt->client->fd, mysql_request_buffer->str, mysql_request_buffer->length);
return SW_OK;
}
static int swoole_mysql_coro_close(zval *zobject)
{
mysql_client *client = (mysql_client *) swoole_get_object(zobject);
if (!client)
{
swoole_php_fatal_error(E_WARNING, "object is not instanceof swoole_mysql_coro.");
return FAILURE;
}
if (!client->cli)
{
return FAILURE;
}
if (client->connected)
{
//send quit command
swString_clear(mysql_request_buffer);
client->cmd = SW_MYSQL_COM_QUIT;
bzero(mysql_request_buffer->str, 5);
mysql_request_buffer->str[4] = SW_MYSQL_COM_QUIT;//command
mysql_request_buffer->length = 5;
mysql_pack_length(mysql_request_buffer->length - 4, mysql_request_buffer->str);
SwooleG.main_reactor->write(SwooleG.main_reactor, client->fd, mysql_request_buffer->str, mysql_request_buffer->length);
client->connected = 0;
}
zend_update_property_bool(swoole_mysql_coro_ce_ptr, zobject, ZEND_STRL("connected"), 0);
SwooleG.main_reactor->del(SwooleG.main_reactor, client->fd);
swConnection *_socket = swReactor_get(SwooleG.main_reactor, client->fd);
_socket->object = NULL;
_socket->active = 0;
if (client->timer)
{
swTimer_del(&SwooleG.timer, client->timer);
client->timer = NULL;
}
if (client->statement_list)
{
swLinkedList_node *node = client->statement_list->head;
while (node)
{
mysql_statement *stmt = (mysql_statement *) node->data;
// after connection closed, mysql stmt cache closed too
// so we needn't send stmt close command here like pdo.
swoole_mysql_coro_statement_free(stmt);
efree(stmt);
node = node->next;
}
swLinkedList_free(client->statement_list);
client->statement_list = NULL;
}
//clear connector
if (client->connector.host)
{
efree(client->connector.host);
client->connector.host = NULL;
}
if (client->connector.user)
{
efree(client->connector.user);
client->connector.user = NULL;
}
if (client->connector.password)
{
efree(client->connector.password);
client->connector.password = NULL;
}
if (client->connector.database)
{
efree(client->connector.database);
client->connector.database = NULL;
}
client->cli->close(client->cli);
swClient_free(client->cli);
efree(client->cli);
client->cli = NULL;
client->state = SW_MYSQL_STATE_CLOSED;
client->iowait = SW_MYSQL_CORO_STATUS_CLOSED;
return SUCCESS;
}
static PHP_METHOD(swoole_mysql_coro, __construct)
{
}
static PHP_METHOD(swoole_mysql_coro, __destruct)
{
}
static PHP_METHOD(swoole_mysql_coro, connect)
{
zval *server_info;
ZEND_PARSE_PARAMETERS_START(1, 1)
Z_PARAM_ARRAY(server_info)
ZEND_PARSE_PARAMETERS_END_EX(RETURN_FALSE);
php_swoole_array_separate(server_info);
HashTable *_ht = Z_ARRVAL_P(server_info);
zval *value;
mysql_client *client = (mysql_client *) swoole_get_object(getThis());
if (client->cli)
{
swoole_php_fatal_error(E_WARNING, "connection to the server has already been established.");
RETURN_FALSE;
}
mysql_connector *connector = &client->connector;
zend::string str_host;
zend::string str_user;
zend::string str_database;
zend::string str_password;
if (php_swoole_array_get_value(_ht, "host", value))
{
str_host = value;
connector->host = str_host.val();
connector->host_len = str_host.len();
}
else
{
zend_throw_exception(swoole_mysql_coro_exception_ce_ptr, "HOST parameter is required.", 11);
zval_ptr_dtor(server_info);
RETURN_FALSE;
}
if (php_swoole_array_get_value(_ht, "port", value))
{
connector->port = zval_get_long(value);
}
else
{
connector->port = SW_MYSQL_DEFAULT_PORT;
}
if (php_swoole_array_get_value(_ht, "user", value))
{
str_user = value;
connector->user = str_user.val();
connector->user_len = str_user.len();
}
else
{
zend_throw_exception(swoole_mysql_coro_exception_ce_ptr, "USER parameter is required.", 11);
zval_ptr_dtor(server_info);
RETURN_FALSE;
}
if (php_swoole_array_get_value(_ht, "password", value))
{
str_password = value;
connector->password = str_password.val();
connector->password_len = str_password.len();
}
else
{
zend_throw_exception(swoole_mysql_coro_exception_ce_ptr, "PASSWORD parameter is required.", 11);
zval_ptr_dtor(server_info);
RETURN_FALSE;
}
if (php_swoole_array_get_value(_ht, "database", value))
{
str_database = value;
connector->database = str_database.val();
connector->database_len = str_database.len();
}
else
{
zend_throw_exception(swoole_mysql_coro_exception_ce_ptr, "DATABASE parameter is required.", 11);
zval_ptr_dtor(server_info);
RETURN_FALSE;
}
if (php_swoole_array_get_value(_ht, "timeout", value))
{
connector->timeout = zval_get_double(value);
}
else
{
connector->timeout = PHPCoroutine::socket_connect_timeout;
}
if (php_swoole_array_get_value(_ht, "charset", value))
{
zend::string str_charset(value);
connector->character_set = mysql_get_charset(str_charset.val());
if (connector->character_set < 0)
{
char buf[64];
snprintf(buf, sizeof(buf), "unknown charset [%s].", str_charset.val());
zend_throw_exception(swoole_mysql_coro_exception_ce_ptr, buf, 11);
zval_ptr_dtor(server_info);
RETURN_FALSE;
}
}
else
{
connector->character_set = SW_MYSQL_DEFAULT_CHARSET;
}
if (php_swoole_array_get_value(_ht, "strict_type", value))
{
connector->strict_type = zval_is_true(value);
}
if (php_swoole_array_get_value(_ht, "fetch_mode", value))
{
connector->fetch_mode = zval_is_true(value);
}
swClient *cli = (swClient *) emalloc(sizeof(swClient));
int type = SW_SOCK_TCP;
if (strncasecmp(connector->host, ZEND_STRL("unix:/")) == 0)
{
connector->host = connector->host + 5;
connector->host_len = connector->host_len - 5;
type = SW_SOCK_UNIX_STREAM;
}
else if (strchr(connector->host, ':'))
{
type = SW_SOCK_TCP6;
}
php_swoole_check_reactor();
if (!swReactor_handle_isset(SwooleG.main_reactor, PHP_SWOOLE_FD_MYSQL_CORO))
{
SwooleG.main_reactor->setHandle(SwooleG.main_reactor, PHP_SWOOLE_FD_MYSQL_CORO | SW_EVENT_READ, swoole_mysql_coro_onRead);
SwooleG.main_reactor->setHandle(SwooleG.main_reactor, PHP_SWOOLE_FD_MYSQL_CORO | SW_EVENT_WRITE, swoole_mysql_coro_onWrite);
SwooleG.main_reactor->setHandle(SwooleG.main_reactor, PHP_SWOOLE_FD_MYSQL_CORO | SW_EVENT_ERROR, swoole_mysql_coro_onError);
}
if (swClient_create(cli, type, 0) < 0)
{
swoole_php_fatal_error(E_WARNING, "swClient_create() failed. Error: %s [%d]", strerror(errno), errno);
_failed:
zend_update_property_string(swoole_mysql_coro_ce_ptr, getThis(), ZEND_STRL("connect_error"), strerror(errno));
zend_update_property_long(swoole_mysql_coro_ce_ptr, getThis(), ZEND_STRL("connect_errno"), errno);
efree(cli);
zval_ptr_dtor(server_info);
RETURN_FALSE;
}
//tcp nodelay
if (type != SW_SOCK_UNIX_STREAM)
{
int tcp_nodelay = 1;
if (setsockopt(cli->socket->fd, IPPROTO_TCP, TCP_NODELAY, (const void *) &tcp_nodelay, sizeof(int)) == -1)
{
swoole_php_sys_error(E_WARNING, "setsockopt(%d, IPPROTO_TCP, TCP_NODELAY) failed.", cli->socket->fd);
}
}
int ret = cli->connect(cli, connector->host, connector->port, connector->timeout, 1);
if ((ret < 0 && errno == EINPROGRESS) || ret == 0)
{
if (SwooleG.main_reactor->add(SwooleG.main_reactor, cli->socket->fd, PHP_SWOOLE_FD_MYSQL_CORO | SW_EVENT_WRITE) < 0)
{
goto _failed;
}
}
else
{
goto _failed;
}
zend_update_property(swoole_mysql_coro_ce_ptr, getThis(), ZEND_STRL("serverInfo"), server_info);
zval_ptr_dtor(server_info);
zend_update_property_long(swoole_mysql_coro_ce_ptr, getThis(), ZEND_STRL("sock"), cli->socket->fd);
if (!client->buffer)
{
client->buffer = swString_new(SW_BUFFER_SIZE_BIG);
}
else
{
swString_clear(client->buffer);
bzero(&client->response, sizeof(client->response));
}
client->fd = cli->socket->fd;
client->object = getThis();
client->cli = cli;
sw_copy_to_stack(client->object, client->_object);
swConnection *_socket = swReactor_get(SwooleG.main_reactor, cli->socket->fd);
_socket->object = client;
_socket->active = 0;
php_coro_context *context = (php_coro_context *) swoole_get_property(getThis(), 0);
if (!context)
{
context = (php_coro_context *) emalloc(sizeof(php_coro_context));
swoole_set_property(getThis(), 0, context);
}
context->state = SW_CORO_CONTEXT_RUNNING;
context->coro_params = *getThis();
connector->host = estrndup(connector->host, connector->host_len);
connector->user = estrndup(connector->user, connector->user_len);
connector->password = estrndup(connector->password, connector->password_len);
connector->database = estrndup(connector->database, connector->database_len);
if (connector->timeout > 0)
{
connector->timer = swTimer_add(&SwooleG.timer, (long) (connector->timeout * 1000), 0, context, swoole_mysql_coro_onTimeout);
}
client->cid = PHPCoroutine::get_cid();
PHPCoroutine::yield_m(return_value, context);
}
static PHP_METHOD(swoole_mysql_coro, query)
{
swString sql;
bzero(&sql, sizeof(sql));
mysql_client *client = (mysql_client *) swoole_get_object(getThis());
if (!client || client->state == SW_MYSQL_STATE_CLOSED)
{
SwooleG.error = SW_ERROR_CLIENT_NO_CONNECTION;
zend_update_property_long(swoole_mysql_coro_ce_ptr, getThis(), ZEND_STRL("errCode"), SwooleG.error);
swoole_php_fatal_error(E_WARNING, "The MySQL connection is not established.");
RETURN_FALSE;
}
if (client->iowait == SW_MYSQL_CORO_STATUS_DONE)
{
swoole_php_fatal_error(E_WARNING, "mysql client is waiting for calling recv, cannot send new sql query.");
RETURN_FALSE;
}
PHPCoroutine::check_bind("mysql client", client->cid);
double timeout = PHPCoroutine::socket_timeout;
if (zend_parse_parameters(ZEND_NUM_ARGS(), "s|d", &sql.str, &sql.length, &timeout) == FAILURE)
{
RETURN_FALSE;
}
if (sql.length <= 0)
{
swoole_php_fatal_error(E_WARNING, "Query is empty.");
RETURN_FALSE;
}
if (mysql_query(getThis(), client, &sql, NULL) < 0)
{
RETURN_FALSE;
}
client->state = SW_MYSQL_STATE_READ_START;
php_coro_context *context = (php_coro_context *) swoole_get_property(getThis(), 0);
if (timeout > 0)
{
client->timer = swTimer_add(&SwooleG.timer, (long) (timeout * 1000), 0, context, swoole_mysql_coro_onTimeout);
if (client->timer && client->defer)
{
context->state = SW_CORO_CONTEXT_IN_DELAYED_TIMEOUT_LIST;
}
}
if (client->defer)
{
client->iowait = SW_MYSQL_CORO_STATUS_WAIT;
RETURN_TRUE;
}
client->suspending = 1;
client->cid = PHPCoroutine::get_cid();
PHPCoroutine::yield_m(return_value, context);
}
static PHP_METHOD(swoole_mysql_coro, nextResult)
{
mysql_client *client = (mysql_client *) swoole_get_object(getThis());
if (!client)
{
RETURN_FALSE;
}
if (client->buffer && (size_t) client->buffer->offset < client->buffer->length)
{
client->cmd = SW_MYSQL_COM_QUERY;
client->state = SW_MYSQL_STATE_READ_START;
client->statement = nullptr;
zval *result = NULL;
if (swoole_mysql_coro_parse_response(client, &result, 1) == SW_OK)
{
swoole_mysql_coro_parse_end(client, client->buffer); // ending tidy up
zval _result = *result;
efree(result);
result = &_result;
RETURN_ZVAL(result, 0, 1);
}
else
{
RETURN_FALSE;
}
}
else
{
RETURN_NULL();
}
}
static void swoole_mysql_coro_query_transcation(const char* command, uint8_t in_transaction, zend_execute_data *execute_data, zval *return_value)
{
mysql_client *client = (mysql_client *) swoole_get_object(getThis());
if (!client)
{
swoole_php_fatal_error(E_WARNING, "object is not instanceof swoole_mysql.");
RETURN_FALSE;
}
PHPCoroutine::check_bind("mysql client", client->cid);
// we deny the dangerous operation of transaction
// if developers need use defer to begin transaction, they can use query("begin/commit/rollback") with defer
// to make sure they know what they are doing
if (unlikely(client->defer))
{
swoole_php_fatal_error(
E_DEPRECATED,
"you should not use defer to handle transaction, "
"if you want, please use `query` instead."
);
RETURN_FALSE;
}
if (in_transaction && client->transaction)
{
zend_throw_exception(swoole_mysql_coro_exception_ce_ptr, "There is already an active transaction.", 21);
RETURN_FALSE;
}
if (!in_transaction && !client->transaction)
{
zend_throw_exception(swoole_mysql_coro_exception_ce_ptr, "There is no active transaction.", 22);
RETURN_FALSE;
}
swString sql;
bzero(&sql, sizeof(sql));
swString_append_ptr(&sql, command, strlen(command));
if (mysql_query(getThis(), client, &sql, NULL) < 0)
{
RETURN_FALSE;
}
else
{
double timeout = PHPCoroutine::socket_timeout;
if (zend_parse_parameters(ZEND_NUM_ARGS(), "|d", &timeout) == FAILURE)
{
RETURN_FALSE;
}
php_coro_context *context = (php_coro_context *) swoole_get_property(getThis(), 0);
if (timeout > 0)
{
client->timer = swTimer_add(&SwooleG.timer, (long) (timeout * 1000), 0, context, swoole_mysql_coro_onTimeout);
}
client->cid = PHPCoroutine::get_cid();
// coro_use_return_value
*(zend_uchar *) &execute_data->prev_execute_data->opline->result_type = IS_VAR;
PHPCoroutine::yield_m(return_value, context);
// resume true
if (Z_BVAL_P(return_value))
{
client->transaction = in_transaction;
}
}
}
static PHP_METHOD(swoole_mysql_coro, begin)
{
swoole_mysql_coro_query_transcation("BEGIN", 1, execute_data, return_value);
}
static PHP_METHOD(swoole_mysql_coro, commit)
{
swoole_mysql_coro_query_transcation("COMMIT", 0, execute_data, return_value);
}
static PHP_METHOD(swoole_mysql_coro, rollback)
{
swoole_mysql_coro_query_transcation("ROLLBACK", 0, execute_data, return_value);
}
static PHP_METHOD(swoole_mysql_coro, getDefer)
{
mysql_client *client = (mysql_client *) swoole_get_object(getThis());
RETURN_BOOL(client->defer);
}
static PHP_METHOD(swoole_mysql_coro, setDefer)
{
zend_bool defer = 1;
if (zend_parse_parameters(ZEND_NUM_ARGS(), "|b", &defer) == FAILURE)
{
RETURN_FALSE;
}
mysql_client *client = (mysql_client *) swoole_get_object(getThis());
if (client->iowait > SW_MYSQL_CORO_STATUS_READY)
{
RETURN_BOOL(defer);
}
client->defer = defer;
RETURN_TRUE
}
static PHP_METHOD(swoole_mysql_coro, recv)
{
mysql_client *client = (mysql_client *) swoole_get_object(getThis());
if (!client->defer)
{
swoole_php_fatal_error(E_WARNING, "you should not use recv without defer ");
RETURN_FALSE;
}
PHPCoroutine::check_bind("mysql client", client->cid);
if (client->iowait == SW_MYSQL_CORO_STATUS_DONE)
{
client->iowait = SW_MYSQL_CORO_STATUS_READY;
zval _result = *client->result;
efree(client->result);
zval *result = &_result;
client->result = NULL;
RETURN_ZVAL(result, 0, 1);
}
if (client->iowait != SW_MYSQL_CORO_STATUS_WAIT)
{
swoole_php_fatal_error(E_WARNING, "no request.");
RETURN_FALSE;
}
client->suspending = 1;
client->cid = PHPCoroutine::get_cid();
php_coro_context *context = (php_coro_context *) swoole_get_property(getThis(), 0);
PHPCoroutine::yield_m(return_value, context);
}
static PHP_METHOD(swoole_mysql_coro, prepare)
{
swString sql;
bzero(&sql, sizeof(sql));
mysql_client *client = (mysql_client *) swoole_get_object(getThis());
if (!client || client->state == SW_MYSQL_STATE_CLOSED)
{
SwooleG.error = SW_ERROR_CLIENT_NO_CONNECTION;
zend_update_property_long(swoole_mysql_coro_ce_ptr, getThis(), ZEND_STRL("errCode"), SwooleG.error);
swoole_php_fatal_error(E_WARNING, "The MySQL connection is not established.");
RETURN_FALSE;
}
if (client->state != SW_MYSQL_STATE_QUERY)
{
swoole_php_fatal_error(E_WARNING, "mysql client is waiting response, cannot send new sql query.");
RETURN_FALSE;
}
PHPCoroutine::check_bind("mysql client", client->cid);
double timeout = PHPCoroutine::socket_timeout;
if (zend_parse_parameters(ZEND_NUM_ARGS(), "s|d", &sql.str, &sql.length, &timeout) == FAILURE)
{
RETURN_FALSE;
}
if (sql.length <= 0)
{
swoole_php_fatal_error(E_WARNING, "Query is empty.");
RETURN_FALSE;
}
if (client->buffer)
{
swString_clear(client->buffer);
}
client->cmd = SW_MYSQL_COM_STMT_PREPARE;
client->state = SW_MYSQL_STATE_READ_START;
if (mysql_prepare_pack(&sql, mysql_request_buffer) < 0)
{
RETURN_FALSE;
}
//send prepare command
if (SwooleG.main_reactor->write(SwooleG.main_reactor, client->fd, mysql_request_buffer->str, mysql_request_buffer->length) < 0)
{
//connection is closed
if (swConnection_error(errno) == SW_CLOSE)
{
zval *zobject = getThis();
zend_update_property_bool(swoole_mysql_coro_ce_ptr, zobject, ZEND_STRL("connected"), 0);
zend_update_property_long(swoole_mysql_coro_ce_ptr, zobject, ZEND_STRL("errno"), 2013);
zend_update_property_string(swoole_mysql_coro_ce_ptr, zobject, ZEND_STRL("error"), "Lost connection to MySQL server during query");
}
RETURN_FALSE;
}
if (client->defer)
{
client->iowait = SW_MYSQL_CORO_STATUS_WAIT;
RETURN_TRUE;
}
php_coro_context *context = (php_coro_context *) swoole_get_property(getThis(), 0);
if (timeout > 0)
{
client->timer = swTimer_add(&SwooleG.timer, (long) (timeout * 1000), 0, context, swoole_mysql_coro_onTimeout);
}
client->suspending = 1;
client->cid = PHPCoroutine::get_cid();
PHPCoroutine::yield_m(return_value, context);
}
static PHP_METHOD(swoole_mysql_coro_statement, execute)
{
zval *params = NULL;
mysql_statement *stmt = (mysql_statement *) swoole_get_object(getThis());
if (!stmt)
{
RETURN_FALSE;
}
mysql_client *client = stmt->client;
if (!client->cli)
{
swoole_php_fatal_error(E_WARNING, "mysql connection#%d is closed.", client->fd);
RETURN_FALSE;
}
double timeout = PHPCoroutine::socket_timeout;
if (zend_parse_parameters(ZEND_NUM_ARGS(), "|ad", &params, &timeout) == FAILURE)
{
RETURN_FALSE;
}
if (stmt->buffer)
{
swString_clear(stmt->buffer);
}
if (swoole_mysql_coro_execute(getThis(), client, params) < 0)
{
RETURN_FALSE;
}
php_coro_context *context = (php_coro_context *) swoole_get_property(client->object, 0);
if (timeout > 0)
{
client->timer = swTimer_add(&SwooleG.timer, (long) (timeout * 1000), 0, context, swoole_mysql_coro_onTimeout);
if (client->timer && client->defer)
{
context->state = SW_CORO_CONTEXT_IN_DELAYED_TIMEOUT_LIST;
}
}
if (client->defer)
{
client->iowait = SW_MYSQL_CORO_STATUS_WAIT;
RETURN_TRUE;
}
client->suspending = 1;
client->cid = PHPCoroutine::get_cid();
PHPCoroutine::yield_m(return_value, context);
}
static PHP_METHOD(swoole_mysql_coro_statement, fetch)
{
mysql_statement *stmt = (mysql_statement *) swoole_get_object(getThis());
if (!stmt)
{
RETURN_FALSE;
}
if (!stmt->client->connector.fetch_mode)
{
RETURN_FALSE;
}
if (stmt->result)
{
zval args[1];
// the function argument is a reference
ZVAL_NEW_REF(stmt->result, stmt->result);
args[0] = *stmt->result;
zval fcn;
ZVAL_STRING(&fcn, "array_shift");
int ret;
zval retval;
ret = call_user_function_ex(EG(function_table), NULL, &fcn, &retval, 1, args, 0, NULL);
zval_ptr_dtor(&fcn);
ZVAL_UNREF(stmt->result);
if (ret == FAILURE)
{
if (stmt->result)
{
sw_zval_free(stmt->result);
stmt->result = NULL;
}
RETURN_NULL();
}
else
{
if (php_swoole_array_length(stmt->result) == 0)
{
sw_zval_free(stmt->result);
stmt->result = NULL;
}
RETURN_ZVAL(&retval, 0, 1);
}
}
else
{
RETURN_NULL();
}
}
static PHP_METHOD(swoole_mysql_coro_statement, fetchAll)
{
mysql_statement *stmt = (mysql_statement *) swoole_get_object(getThis());
if (!stmt)
{
RETURN_FALSE;
}
if (!stmt->client->connector.fetch_mode)
{
RETURN_FALSE;
}
if (stmt->result)
{
zval _result = *stmt->result;
efree(stmt->result);
zval *result = &_result;
stmt->result = NULL;
RETURN_ZVAL(result, 0, 1);
}
else
{
RETURN_NULL();
}
}
static PHP_METHOD(swoole_mysql_coro_statement, nextResult)
{
mysql_statement *stmt = (mysql_statement *) swoole_get_object(getThis());
if (!stmt)
{
RETURN_FALSE;
}
mysql_client *client = stmt->client;
if (stmt->buffer && (size_t) stmt->buffer->offset < stmt->buffer->length)
{
client->cmd = SW_MYSQL_COM_STMT_EXECUTE;
client->state = SW_MYSQL_STATE_READ_START;
client->statement = stmt;
zval *result = NULL;
if (swoole_mysql_coro_parse_response(client, &result, 1) == SW_OK)
{
swoole_mysql_coro_parse_end(client, stmt->buffer); // ending tidy up
zval _result = *result;
efree(result);
result = &_result;
RETURN_ZVAL(result, 0, 1);
}
else
{
RETURN_FALSE;
}
}
else
{
RETURN_NULL();
}
}
static PHP_METHOD(swoole_mysql_coro_statement, __destruct)
{
SW_PREVENT_USER_DESTRUCT;
mysql_statement *stmt = (mysql_statement *) swoole_get_object(getThis());
if (!stmt)
{
return;
}
swoole_mysql_coro_statement_close(stmt);
swoole_mysql_coro_statement_free(stmt);
swLinkedList_remove(stmt->client->statement_list, stmt);
efree(stmt);
}
#ifdef SW_USE_MYSQLND
static PHP_METHOD(swoole_mysql_coro, escape)
{
swString str;
bzero(&str, sizeof(str));
zend_long flags = 0;
if (zend_parse_parameters(ZEND_NUM_ARGS(), "s|l", &str.str, &str.length, &flags) == FAILURE)
{
RETURN_FALSE;
}
if (str.length <= 0)
{
swoole_php_fatal_error(E_WARNING, "String is empty.");
RETURN_FALSE;
}
mysql_client *client = (mysql_client *) swoole_get_object(getThis());
if (!client)
{
swoole_php_fatal_error(E_WARNING, "object is not instanceof swoole_mysql.");
RETURN_FALSE;
}
if (!client->cli)
{
swoole_php_fatal_error(E_WARNING, "mysql connection#%d is closed.", client->fd);
RETURN_FALSE;
}
char *newstr = (char *) safe_emalloc(2, str.length + 1, 1);
if (newstr == NULL)
{
swoole_php_fatal_error(E_ERROR, "emalloc(%ld) failed.", str.length + 1);
RETURN_FALSE;
}
const MYSQLND_CHARSET* cset = mysqlnd_find_charset_nr(client->connector.character_set);
if (cset == NULL)
{
swoole_php_fatal_error(E_ERROR, "unknown mysql charset[%d].", client->connector.character_set);
RETURN_FALSE;
}
int newstr_len = mysqlnd_cset_escape_slashes(cset, newstr, str.str, str.length);
if (newstr_len < 0)
{
swoole_php_fatal_error(E_ERROR, "mysqlnd_cset_escape_slashes() failed.");
RETURN_FALSE;
}
RETVAL_STRINGL(newstr, newstr_len);
efree(newstr);
return;
}
#endif
static PHP_METHOD(swoole_mysql_coro, close)
{
if (swoole_mysql_coro_close(getThis()) == FAILURE)
{
RETURN_FALSE;
}
RETURN_TRUE;
}
static void swoole_mysql_coro_free_object(zend_object *object)
{
// as __destruct
uint32_t handle = object->handle;
zval _zobject, *zobject = &_zobject;
ZVAL_OBJ(zobject, object);
mysql_client *client = (mysql_client *) swoole_get_object_by_handle(handle);
if (client)
{
if (client->state != SW_MYSQL_STATE_CLOSED && client->cli)
{
swoole_mysql_coro_close(zobject);
}
if (client->buffer)
{
swString_free(client->buffer);
}
efree(client);
swoole_set_object_by_handle(handle, NULL);
}
php_coro_context *context = (php_coro_context *) swoole_get_property_by_handle(handle, 0);
if (context)
{
efree(context);
swoole_set_property_by_handle(handle, 0, NULL);
}
zend_object_std_dtor(object);
}
static int swoole_mysql_coro_onError(swReactor *reactor, swEvent *event)
{
zval *retval = NULL, *result = sw_malloc_zval();;
mysql_client *client = (mysql_client *) event->socket->object;
zval *zobject = client->object;
swoole_mysql_coro_close(zobject);
zend_update_property_string(swoole_mysql_coro_ce_ptr, zobject, ZEND_STRL("connect_error"), "EPOLLERR/EPOLLHUP/EPOLLRDHUP happen!");
zend_update_property_long(swoole_mysql_coro_ce_ptr, zobject, ZEND_STRL("connect_errno"), ECONNRESET);
ZVAL_BOOL(result, 0);
if (client->defer && !client->suspending)
{
client->result = result;
return SW_OK;
}
client->suspending = 0;
client->cid = 0;
php_coro_context *sw_current_context = (php_coro_context *) swoole_get_property(zobject, 0);
int ret = PHPCoroutine::resume_m(sw_current_context, result, retval);
sw_zval_free(result);
if (ret == SW_CORO_ERR_END && retval)
{
zval_ptr_dtor(retval);
}
return SW_OK;
}
static void swoole_mysql_coro_onConnect(mysql_client *client)
{
zval *zobject = client->object;
zval *retval = NULL;
zval result;
if (client->connector.timer)
{
swTimer_del(&SwooleG.timer, client->connector.timer);
client->connector.timer = NULL;
}
//SwooleG.main_reactor->del(SwooleG.main_reactor, client->fd);
if (client->connector.error_code > 0)
{
zend_update_property_stringl(swoole_mysql_coro_ce_ptr, zobject, ZEND_STRL("connect_error"), client->connector.error_msg, client->connector.error_length);
zend_update_property_long(swoole_mysql_coro_ce_ptr, zobject, ZEND_STRL("connect_errno"), client->connector.error_code);
ZVAL_BOOL(&result, 0);
swoole_mysql_coro_close(zobject);
}
else
{
client->state = SW_MYSQL_STATE_QUERY;
client->iowait = SW_MYSQL_CORO_STATUS_READY;
zend_update_property_bool(swoole_mysql_coro_ce_ptr, zobject, ZEND_STRL("connected"), 1);
client->connected = 1;
ZVAL_BOOL(&result, 1);
}
client->cid = 0;
php_coro_context *sw_current_context = (php_coro_context *) swoole_get_property(zobject, 0);
int ret = PHPCoroutine::resume_m(sw_current_context, &result, retval);
if (ret == SW_CORO_ERR_END && retval)
{
zval_ptr_dtor(retval);
}
}
static void swoole_mysql_coro_onTimeout(swTimer *timer, swTimer_node *tnode)
{
zval *result = sw_malloc_zval();;
zval *retval = NULL;
php_coro_context *ctx = (php_coro_context *) tnode->data;
zval _zobject = ctx->coro_params;
zval *zobject = & _zobject;
ZVAL_BOOL(result, 0);
mysql_client *client = (mysql_client *) swoole_get_object(zobject);
if (client->handshake != SW_MYSQL_HANDSHAKE_COMPLETED)
{
zend_update_property_string(swoole_mysql_coro_ce_ptr, zobject, ZEND_STRL("connect_error"), "connect timeout");
zend_update_property_long(swoole_mysql_coro_ce_ptr, zobject, ZEND_STRL("connect_errno"), ETIMEDOUT);
}
else
{
zend_update_property_string(swoole_mysql_coro_ce_ptr, zobject, ZEND_STRL("error"), "query timeout");
zend_update_property_long(swoole_mysql_coro_ce_ptr, zobject, ZEND_STRL("errno"), ETIMEDOUT);
}
//timeout close conncttion
client->timer = NULL;
client->state = SW_MYSQL_STATE_QUERY;
swoole_mysql_coro_close(zobject);
if (client->defer && !client->suspending)
{
client->result = result;
return;
}
client->suspending = 0;
client->cid = 0;
int ret = PHPCoroutine::resume_m(ctx, result, retval);
if (ret == SW_CORO_ERR_END && retval)
{
zval_ptr_dtor(retval);
}
sw_zval_free(result);
}
static int swoole_mysql_coro_onWrite(swReactor *reactor, swEvent *event)
{
if (event->socket->active)
{
return swReactor_onWrite(SwooleG.main_reactor, event);
}
socklen_t len = sizeof(SwooleG.error);
if (getsockopt(event->fd, SOL_SOCKET, SO_ERROR, &SwooleG.error, &len) < 0)
{
swWarn("getsockopt(%d) failed. Error: %s[%d]", event->fd, strerror(errno), errno);
return SW_ERR;
}
mysql_client *client = (mysql_client *) event->socket->object;
//success
if (SwooleG.error == 0)
{
//listen read event
SwooleG.main_reactor->set(SwooleG.main_reactor, event->fd, PHP_SWOOLE_FD_MYSQL_CORO | SW_EVENT_READ);
//connected
event->socket->active = 1;
client->handshake = SW_MYSQL_HANDSHAKE_WAIT_REQUEST;
}
else
{
client->connector.error_code = SwooleG.error;
client->connector.error_msg = strerror(SwooleG.error);
client->connector.error_length = strlen(client->connector.error_msg);
swoole_mysql_coro_onConnect(client);
}
return SW_OK;
}
static int swoole_mysql_coro_onHandShake(mysql_client *client)
{
swString *buffer = client->buffer;
swClient *cli = client->cli;
mysql_connector *connector = &client->connector;
int n = cli->recv(cli, buffer->str + buffer->length, buffer->size - buffer->length, 0);
if (n < 0)
{
switch (swConnection_error(errno))
{
case SW_ERROR:
swSysError("Read from socket[%d] failed.", cli->socket->fd);
return SW_ERR;
case SW_CLOSE:
_system_call_error: connector->error_code = errno;
connector->error_msg = strerror(errno);
connector->error_length = strlen(connector->error_msg);
swoole_mysql_coro_onConnect(client);
return SW_OK;
case SW_WAIT:
return SW_OK;
default:
return SW_ERR;
}
}
else if (n == 0)
{
errno = ECONNRESET;
goto _system_call_error;
}
buffer->length += n;
int ret = 0;
_again:
swTraceLog(SW_TRACE_MYSQL_CLIENT, "handshake on %d", client->handshake);
if (client->switch_check)
{
// after handshake we need check if server request us to switch auth type first
goto _check_switch;
}
switch(client->handshake)
{
case SW_MYSQL_HANDSHAKE_WAIT_REQUEST:
{
client->switch_check = 1;
ret = mysql_handshake(connector, buffer->str, buffer->length);
if (ret < 0)
{
goto _error;
}
else if (ret > 0)
{
_send:
if (cli->send(cli, connector->buf, connector->packet_length + 4, 0) < 0)
{
goto _system_call_error;
}
else
{
// clear for the new packet
swString_clear(buffer);
// mysql_handshake will return the next state flag
client->handshake = ret;
}
}
break;
}
case SW_MYSQL_HANDSHAKE_WAIT_SWITCH:
{
_check_switch:
client->switch_check = 0;
int next_state;
// handle auth switch request
switch (next_state = mysql_auth_switch(connector, buffer->str, buffer->length))
{
case SW_AGAIN:
return SW_OK;
case SW_ERR:
// not the switch packet, go to the next
goto _again;
default:
ret = next_state;
goto _send;
}
break;
}
case SW_MYSQL_HANDSHAKE_WAIT_SIGNATURE:
{
switch (mysql_parse_auth_signature(buffer, connector))
{
case SW_MYSQL_AUTH_SIGNATURE_SUCCESS:
{
client->handshake = SW_MYSQL_HANDSHAKE_WAIT_RESULT;
break;
}
case SW_MYSQL_AUTH_SIGNATURE_FULL_AUTH_REQUIRED:
{
// send response and wait RSA public key
ret = SW_MYSQL_HANDSHAKE_WAIT_RSA; // handshake = ret
goto _send;
}
default:
{
goto _error;
}
}
// may be more packets
if ((size_t) buffer->offset < buffer->length)
{
goto _again;
}
else
{
swString_clear(buffer);
}
break;
}
case SW_MYSQL_HANDSHAKE_WAIT_RSA:
{
// encode by RSA
#ifdef SW_MYSQL_RSA_SUPPORT
switch (mysql_parse_rsa(connector, SW_STRINGCVL(buffer)))
{
case SW_AGAIN:
return SW_OK;
case SW_OK:
ret = SW_MYSQL_HANDSHAKE_WAIT_RESULT; // handshake = ret
goto _send;
default:
goto _error;
}
#else
connector->error_code = -1;
connector->error_msg = (char *) "MySQL8 RSA-Auth need enable OpenSSL!";
connector->error_length = strlen(connector->error_msg);
swoole_mysql_coro_onConnect(client);
return SW_OK;
#endif
break;
}
default:
{
ret = mysql_get_result(connector, SW_STRINGCVL(buffer));
if (ret < 0)
{
_error:
swoole_mysql_coro_onConnect(client);
}
else if (ret > 0)
{
swString_clear(buffer);
client->handshake = SW_MYSQL_HANDSHAKE_COMPLETED;
swoole_mysql_coro_onConnect(client);
}
// else recv again
}
}
return SW_OK;
}
static int swoole_mysql_coro_onRead(swReactor *reactor, swEvent *event)
{
mysql_client *client = (mysql_client *) event->socket->object;
if (client->handshake != SW_MYSQL_HANDSHAKE_COMPLETED)
{
return swoole_mysql_coro_onHandShake(client);
}
if (client->timer)
{
swTimer_del(&SwooleG.timer, client->timer);
client->timer = NULL;
}
int sock = event->fd;
int ret;
zval *zobject = client->object;
swString *buffer;
if (client->cmd == SW_MYSQL_COM_STMT_EXECUTE)
{
if (client->statement->buffer == NULL)
{
// statement save the response data itself
client->statement->buffer = swString_new(SW_BUFFER_SIZE_BIG);
}
buffer = client->statement->buffer;
}
else
{
buffer = client->buffer;
}
zval *retval = NULL;
zval *result = NULL;
while(1)
{
ret = recv(sock, buffer->str + buffer->length, buffer->size - buffer->length, 0);
swTraceLog(SW_TRACE_MYSQL_CLIENT, "recv-ret=%d, buffer-length=%zu.", ret, buffer->length);
if (ret < 0)
{
if (errno == EINTR)
{
continue;
}
else
{
switch (swConnection_error(errno))
{
case SW_ERROR:
swSysError("Read from socket[%d] failed.", event->fd);
return SW_ERR;
case SW_CLOSE:
goto _close_fd;
case SW_WAIT:
return SW_OK;
default:
return SW_ERR;
}
}
}
else if (ret == 0)
{
_close_fd:
if (client->state == SW_MYSQL_STATE_READ_END)
{
goto _parse_response;
}
zend_update_property_long(swoole_mysql_coro_ce_ptr, zobject, ZEND_STRL("connect_errno"), 111);
zend_update_property_string(swoole_mysql_coro_ce_ptr, zobject, ZEND_STRL("connect_error"), "connection close by peer");
if (client->connected)
{
client->connected = 0;
zend_update_property_long(swoole_mysql_coro_ce_ptr, zobject, ZEND_STRL("errno"), 2006);
zend_update_property_string(swoole_mysql_coro_ce_ptr, zobject, ZEND_STRL("error"), "MySQL server has gone away");
}
_active_close:
swoole_mysql_coro_close(zobject);
if (!client->cid)
{
return SW_OK;
}
result = sw_malloc_zval();
ZVAL_BOOL(result, 0);
if (client->defer && !client->suspending)
{
client->iowait = SW_MYSQL_CORO_STATUS_DONE;
client->result = result;
return SW_OK;
}
client->suspending = 0;
client->cid = 0;
php_coro_context *sw_current_context = (php_coro_context *) swoole_get_property(zobject, 0);
ret = PHPCoroutine::resume_m(sw_current_context, result, retval);
sw_zval_free(result);
if (ret == SW_CORO_ERR_END && retval)
{
zval_ptr_dtor(retval);
}
client->state = SW_MYSQL_STATE_QUERY;
return SW_OK;
}
else
{
buffer->length += ret;
//recv again
if (buffer->length == buffer->size)
{
if (swString_extend(buffer, buffer->size * 2) < 0)
{
swoole_php_fatal_error(E_ERROR, "malloc failed.");
reactor->del(SwooleG.main_reactor, event->fd);
}
continue;
}
_parse_response:
if (client->tmp_result)
{
_check_over:
// maybe more responses has already received in buffer, we check it now.
if (mysql_is_over(client) != SW_OK)
{
// the **last** sever status flag shows that more results exist but we hasn't received.
return SW_OK;
}
else
{
result = client->tmp_result;
client->tmp_result = NULL;
}
}
else
{
ret = swoole_mysql_coro_parse_response(client, &result, 0);
if (ret == SW_AGAIN)
{
return SW_OK; // parse error or need again
}
if (client->response.status_code & SW_MYSQL_SERVER_MORE_RESULTS_EXISTS)
{
client->tmp_result = result;
goto _check_over;
}
}
swoole_mysql_coro_parse_end(client, buffer); // ending tidy up
if (client->defer && !client->suspending)
{
client->iowait = SW_MYSQL_CORO_STATUS_DONE;
client->result = result;
return SW_OK;
}
if (!client->cid)
{
goto _active_close; // error
}
client->suspending = 0;
client->iowait = SW_MYSQL_CORO_STATUS_READY;
client->cid = 0;
php_coro_context *sw_current_context = (php_coro_context *) swoole_get_property(zobject, 0);
ret = PHPCoroutine::resume_m(sw_current_context, result, retval);
if (result)
{
sw_zval_free(result);
}
if (ret == SW_CORO_ERR_END && retval)
{
zval_ptr_dtor(retval);
}
return SW_OK;
}
}
return SW_OK;
}
C
1
https://gitee.com/scaly/swoole.git
git@gitee.com:scaly/swoole.git
scaly
swoole
swoole-src
master

搜索帮助