1 Star 0 Fork 345

衣小白 / swoole-src

forked from swoole / swoole-src 
加入 Gitee
与超过 1200万 开发者一起发现、参与优秀开源项目,私有仓库也完全免费 :)
免费加入
克隆/下载
swoole_mysql_coro.cc 81.73 KB
一键复制 编辑 原始数据 按行查看 历史
twosee 提交于 2019-10-08 15:52 . Fix #2869
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131213221332134213521362137213821392140214121422143214421452146214721482149215021512152215321542155215621572158215921602161216221632164216521662167216821692170217121722173217421752176217721782179218021812182218321842185218621872188218921902191219221932194219521962197219821992200220122022203220422052206220722082209221022112212221322142215221622172218221922202221222222232224222522262227222822292230223122322233223422352236223722382239224022412242224322442245224622472248224922502251225222532254225522562257225822592260226122622263226422652266226722682269227022712272227322742275227622772278227922802281228222832284228522862287228822892290229122922293229422952296229722982299230023012302230323042305230623072308230923102311231223132314231523162317231823192320232123222323232423252326232723282329233023312332233323342335233623372338233923402341234223432344234523462347234823492350235123522353235423552356235723582359236023612362236323642365236623672368236923702371237223732374237523762377237823792380238123822383238423852386238723882389239023912392239323942395239623972398239924002401240224032404240524062407240824092410241124122413241424152416241724182419242024212422242324242425242624272428242924302431243224332434243524362437243824392440244124422443244424452446244724482449245024512452245324542455245624572458245924602461246224632464246524662467246824692470247124722473247424752476247724782479248024812482248324842485248624872488248924902491249224932494249524962497249824992500250125022503250425052506250725082509251025112512251325142515251625172518251925202521252225232524
/*
+----------------------------------------------------------------------+
| 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: Twosee <twose@qq.com> |
| Author: Tianfeng Han <mikan.tenny@gmail.com> |
+----------------------------------------------------------------------+
*/
#include "php_swoole_cxx.h"
#include "swoole_mysql_proto.h"
// see mysqlnd 'L64' macro redefined
#undef L64
extern "C" {
#include "ext/hash/php_hash.h"
#include "ext/hash/php_hash_sha.h"
#include "ext/standard/php_math.h"
#ifdef SW_USE_MYSQLND
#include "ext/mysqlnd/mysqlnd.h"
#include "ext/mysqlnd/mysqlnd_charset.h"
#endif
}
#include <unordered_map>
using namespace swoole;
using swoole::coroutine::Socket;
/* keep same with pdo and mysqli */
#define MYSQLND_UNKNOWN_SQLSTATE "HY000"
#define MYSQLND_SERVER_GONE "MySQL server has gone away"
#define MYSQLND_CR_UNKNOWN_ERROR 2000
#define MYSQLND_CR_CONNECTION_ERROR 2002
#define MYSQLND_CR_SERVER_GONE_ERROR 2006
#define MYSQLND_CR_OUT_OF_MEMORY 2008
#define MYSQLND_CR_SERVER_LOST 2013
#define MYSQLND_CR_COMMANDS_OUT_OF_SYNC 2014
#define MYSQLND_CR_CANT_FIND_CHARSET 2019
#define MYSQLND_CR_MALFORMED_PACKET 2027
#define MYSQLND_CR_NOT_IMPLEMENTED 2054
#define MYSQLND_CR_NO_PREPARE_STMT 2030
#define MYSQLND_CR_PARAMS_NOT_BOUND 2031
#define MYSQLND_CR_INVALID_PARAMETER_NO 2034
#define MYSQLND_CR_INVALID_BUFFER_USE 2035
namespace swoole
{
class mysql_statement;
class mysql_client
{
public:
/* session related {{{ */
Socket *socket = nullptr;
Socket::timeout_controller *tc = nullptr;
enum sw_mysql_state state = SW_MYSQL_STATE_CLOSED;
bool quit = false;
mysql::result_info result;
std::unordered_map<uint32_t, mysql_statement*> statements;
mysql_statement* statement = nullptr;
/* }}} */
std::string host = SW_MYSQL_DEFAULT_HOST;
uint16_t port = SW_MYSQL_DEFAULT_PORT;
bool ssl = false;
std::string user = "root";
std::string password = "root";
std::string database = "test";
char charset = SW_MYSQL_DEFAULT_CHARSET;
double connect_timeout = Socket::default_connect_timeout;
bool strict_type = false;
inline int get_error_code()
{
return error_code;
}
inline const char* get_error_msg()
{
return error_msg.c_str();
}
inline void non_sql_error(int code, const char *msg)
{
error_code = code;
error_msg = cpp_string::format("SQLSTATE[" MYSQLND_UNKNOWN_SQLSTATE "] [%d] %s", code, msg);
}
template<typename ...Args>
inline void non_sql_error(int code, const char *format, Args ...args)
{
error_code = code;
error_msg = cpp_string::format(
"SQLSTATE[" MYSQLND_UNKNOWN_SQLSTATE "] [%d] %s",
code, cpp_string::format(format, args...).c_str()
);
}
void io_error()
{
if (state == SW_MYSQL_STATE_CLOSED)
{
non_sql_error(MYSQLND_CR_CONNECTION_ERROR, socket->errMsg);
}
else
{
non_sql_error(
MYSQLND_CR_SERVER_GONE_ERROR, MYSQLND_SERVER_GONE "%s%s",
socket->errCode ? " due to " : "", socket->errCode ? socket->errMsg : ""
);
}
/* don't send QUIT after IO error */
quit = true;
close();
}
void proto_error(const char *data, const enum sw_mysql_packet_types expected_type)
{
mysql::server_packet packet(data);
non_sql_error(
MYSQLND_CR_MALFORMED_PACKET, "Unexpected mysql packet length=%u, number=%u, type=%u, expected_type=%u",
packet.header.length, packet.header.number, (uint8_t) data[SW_MYSQL_PACKET_HEADER_SIZE], expected_type
);
close();
}
void server_error(const char *data)
{
mysql::err_packet err_packet(data);
error_code = err_packet.code;
error_msg = cpp_string::format("SQLSTATE[%s] [%d] %s", err_packet.sql_state, err_packet.code, err_packet.msg.c_str());
state = SW_MYSQL_STATE_IDLE;
}
inline bool get_fetch_mode()
{
return fetch_mode;
}
inline bool set_fetch_mode(bool v)
{
if (sw_unlikely(socket && v))
{
non_sql_error(ENOTSUP, "Can not use fetch mode after the connection is established");
return false;
}
fetch_mode = v;
return true;
}
inline bool get_defer()
{
return defer;
}
inline bool set_defer(bool v)
{
// if (sw_unlikely(fetch_mode && v))
// {
// non_sql_error(ENOTSUP, "Can not use defer mode when fetch mode is on");
// return false;
// }
defer = v;
return true;
}
void add_timeout_controller(double timeout, const enum swTimeout_type type)
{
if (sw_unlikely(!socket))
{
return;
}
// Notice: `timeout > 0` is wrong, maybe -1
if (timeout != 0)
{
SW_ASSERT(!tc);
tc = new Socket::timeout_controller(socket, timeout, type);
}
}
inline bool has_timedout(enum swTimeout_type type)
{
return tc && tc->has_timedout(type);
}
void del_timeout_controller()
{
if (tc)
{
delete tc;
tc = nullptr;
}
}
bool connect(std::string host, uint16_t port, bool ssl);
inline bool connect()
{
return connect(host, port, ssl);
}
inline bool is_connect()
{
return socket && socket->is_connect();
}
inline int get_fd()
{
return socket ? socket->get_fd() : -1;
}
inline bool check_connection()
{
if (sw_unlikely(!is_connect()))
{
non_sql_error(MYSQLND_CR_CONNECTION_ERROR, "%s or %s", strerror(ECONNRESET), strerror(ENOTCONN));
return false;
}
return true;
}
inline bool check_liveness()
{
if (sw_unlikely(!check_connection()))
{
return false;
}
if (sw_unlikely(!socket->check_liveness()))
{
non_sql_error(MYSQLND_CR_SERVER_GONE_ERROR, MYSQLND_SERVER_GONE);
close();
return false;
}
return true;
}
inline bool is_writable()
{
return is_connect() && !socket->has_bound(SW_EVENT_WRITE);
}
bool is_available_for_new_reuqest()
{
if (sw_unlikely(state != SW_MYSQL_STATE_IDLE && state != SW_MYSQL_STATE_CLOSED))
{
if (socket)
{
socket->check_bound_co(SW_EVENT_RDWR);
}
non_sql_error(
EINPROGRESS,
"MySQL client is busy now on state#%d, "
"please use recv/fetchAll/nextResult to get all unread data "
"and wait for response then try again",
state
);
return false;
}
if (sw_unlikely(!check_liveness()))
{
return false;
}
else
{
/* without unread data */
swString *buffer = socket->get_read_buffer();
SW_ASSERT(buffer->length == (size_t) buffer->offset);
swString_clear(buffer);
return true;
}
}
const char* recv_packet();
inline const char* recv_none_error_packet()
{
const char *data = recv_packet();
if (sw_unlikely(data && mysql::server_packet::is_err(data)))
{
server_error(data);
return nullptr;
}
return data;
}
inline const char* recv_eof_packet()
{
const char *data = recv_packet();
if (sw_unlikely(data && !mysql::server_packet::is_eof(data)))
{
proto_error(data, SW_MYSQL_PACKET_EOF);
return nullptr;
}
#ifdef SW_LOG_TRACE_OPEN
mysql::eof_packet eof_packet(data);
#endif
return data;
}
inline bool send_raw(const char *data, size_t length)
{
if (sw_unlikely(!check_connection()))
{
return false;
}
else
{
if (sw_unlikely(has_timedout(SW_TIMEOUT_WRITE)))
{
io_error();
return false;
}
if (sw_unlikely(socket->send_all(data, length) != (ssize_t ) length))
{
io_error();
return false;
}
return true;
}
}
bool send_packet(mysql::client_packet *packet);
bool send_command(enum sw_mysql_command command, const char* sql = nullptr, size_t length = 0);
// just for internal
void send_command_without_check(enum sw_mysql_command command, const char* sql = nullptr, size_t length = 0);
void query(zval *return_value, const char *statement, size_t statement_length);
void send_query_request(zval *return_value, const char *statement, size_t statement_length);
void recv_query_response(zval *return_value);
const char* handle_row_data_size(mysql::row_data *row_data, uint8_t size);
bool handle_row_data_lcb(mysql::row_data *row_data);
void handle_row_data_text(zval *return_value, mysql::row_data *row_data, mysql::field_packet *field);
void handle_strict_type(zval *ztext, mysql::field_packet *field);
void fetch(zval *return_value);
void fetch_all(zval *return_value);
void next_result(zval *return_value);
bool recv();
bool send_prepare_request(const char *statement, size_t statement_length);
mysql_statement* recv_prepare_response();
void close();
~mysql_client()
{
SW_ASSERT(statements.empty());
close();
}
private:
int error_code = 0;
std::string error_msg = "";
/* unable to support both features at the same time, so we have to set them by method {{{ */
bool fetch_mode = false;
bool defer = false;
/* }}} */
// recv data of specified length
const char* recv_length(size_t need_length, const bool try_to_recycle = false);
// usually mysql->connect = connect(TCP) + handshake
bool handshake();
};
class mysql_statement
{
public:
std::string statement;
mysql::statement info;
mysql::result_info result;
mysql_statement(mysql_client *client, const char *statement, size_t statement_length) :
client(client)
{
this->statement = std::string(statement, statement_length);
}
inline mysql_client* get_client()
{
return client;
}
inline int get_error_code()
{
return sw_likely(client) ? client->get_error_code() : error_code;
}
inline const char* get_error_msg()
{
return sw_likely(client) ? client->get_error_msg() : error_msg.c_str();
}
inline bool is_available()
{
if (sw_unlikely(!client))
{
error_code = ECONNRESET;
error_msg = "statement must to be recompiled after the connection is broken";
return false;
}
return true;
}
inline bool is_available_for_new_reuqest()
{
if (sw_unlikely(!is_available()))
{
return false;
}
if (sw_unlikely(!client->is_available_for_new_reuqest()))
{
return false;
}
return true;
}
inline void add_timeout_controller(double timeout, const enum swTimeout_type type)
{
if (sw_likely(client))
{
client->add_timeout_controller(timeout, type);
}
}
inline void del_timeout_controller()
{
if (sw_likely(client))
{
client->del_timeout_controller();
}
}
// [notify = false]: mysql_client actively close
inline void close(const bool notify = true)
{
if (client)
{
// if client point exists, socket is always available
if (notify)
{
if (sw_likely(client->is_writable()))
{
char id[4];
sw_mysql_int4store(id, info.id);
client->send_command_without_check(SW_MYSQL_COM_STMT_CLOSE, id, sizeof(id));
}
client->statements.erase(info.id);
}
else
{
error_code = client->get_error_code();
error_msg = client->get_error_msg();
}
client = nullptr;
}
}
~mysql_statement()
{
close();
}
bool send_prepare_request();
bool recv_prepare_response();
void execute(zval *return_value, zval *params);
void send_execute_request(zval *return_value, zval *params);
void recv_execute_response(zval *return_value);
void fetch(zval *return_value);
void fetch_all(zval *return_value);
void next_result(zval *return_value);
private:
mysql_client *client = nullptr;
int error_code = 0;
std::string error_msg;
};
}
static zend_class_entry *swoole_mysql_coro_ce;
static zend_object_handlers swoole_mysql_coro_handlers;
static zend_class_entry *swoole_mysql_coro_exception_ce;
static zend_object_handlers swoole_mysql_coro_exception_handlers;
static zend_class_entry *swoole_mysql_coro_statement_ce;
static zend_object_handlers swoole_mysql_coro_statement_handlers;
typedef struct {
mysql_client *client;
zend_object std;
} mysql_coro_t;
typedef struct {
mysql_statement *statement;
zend_object *zclient;
zend_object std;
} mysql_coro_statement_t;
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, getDefer);
static PHP_METHOD(swoole_mysql_coro, setDefer);
static PHP_METHOD(swoole_mysql_coro, query);
static PHP_METHOD(swoole_mysql_coro, fetch);
static PHP_METHOD(swoole_mysql_coro, fetchAll);
static PHP_METHOD(swoole_mysql_coro, nextResult);
static PHP_METHOD(swoole_mysql_coro, prepare);
static PHP_METHOD(swoole_mysql_coro, recv);
static PHP_METHOD(swoole_mysql_coro, begin);
static PHP_METHOD(swoole_mysql_coro, commit);
static PHP_METHOD(swoole_mysql_coro, rollback);
#ifdef SW_USE_MYSQLND
static PHP_METHOD(swoole_mysql_coro, escape);
#endif
static PHP_METHOD(swoole_mysql_coro, close);
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);
static PHP_METHOD(swoole_mysql_coro_statement, recv);
static PHP_METHOD(swoole_mysql_coro_statement, close);
ZEND_BEGIN_ARG_INFO_EX(arginfo_swoole_void, 0, 0, 0)
ZEND_END_ARG_INFO()
ZEND_BEGIN_ARG_INFO_EX(arginfo_swoole_optional_timeout, 0, 0, 0)
ZEND_ARG_INFO(0, timeout)
ZEND_END_ARG_INFO()
ZEND_BEGIN_ARG_INFO_EX(arginfo_swoole_mysql_coro_connect, 0, 0, 0)
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_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()
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, getDefer, arginfo_swoole_void, ZEND_ACC_PUBLIC)
PHP_ME(swoole_mysql_coro, setDefer, arginfo_swoole_mysql_coro_setDefer, 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, fetch, arginfo_swoole_void, ZEND_ACC_PUBLIC)
PHP_ME(swoole_mysql_coro, fetchAll, arginfo_swoole_void, ZEND_ACC_PUBLIC)
PHP_ME(swoole_mysql_coro, nextResult, arginfo_swoole_void, ZEND_ACC_PUBLIC)
PHP_ME(swoole_mysql_coro, prepare, arginfo_swoole_mysql_coro_prepare, ZEND_ACC_PUBLIC)
PHP_ME(swoole_mysql_coro, recv, arginfo_swoole_void, ZEND_ACC_PUBLIC)
PHP_ME(swoole_mysql_coro, begin, arginfo_swoole_optional_timeout, ZEND_ACC_PUBLIC)
PHP_ME(swoole_mysql_coro, commit, arginfo_swoole_optional_timeout, ZEND_ACC_PUBLIC)
PHP_ME(swoole_mysql_coro, rollback, arginfo_swoole_optional_timeout, 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, 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_optional_timeout, ZEND_ACC_PUBLIC)
PHP_ME(swoole_mysql_coro_statement, fetchAll, arginfo_swoole_optional_timeout, ZEND_ACC_PUBLIC)
PHP_ME(swoole_mysql_coro_statement, nextResult, arginfo_swoole_optional_timeout, ZEND_ACC_PUBLIC)
PHP_ME(swoole_mysql_coro_statement, recv, arginfo_swoole_optional_timeout, ZEND_ACC_PUBLIC)
PHP_ME(swoole_mysql_coro_statement, close, arginfo_swoole_void, ZEND_ACC_PUBLIC)
PHP_FE_END
};
void swoole_sha256(const char *str, int len, unsigned char *digest)
{
PHP_SHA256_CTX context;
PHP_SHA256Init(&context);
PHP_SHA256Update(&context, (unsigned char *) str, len);
PHP_SHA256Final(digest, &context);
}
bool mysql_client::connect(std::string host, uint16_t port, bool ssl)
{
if (socket && (host != this->host || port != this->port || ssl != this->ssl))
{
close();
}
if (!socket)
{
if (host.compare(0, 6, "unix:/", 0, 6) == 0)
{
host = host.substr(sizeof("unix:") - 1);
host.erase(0, host.find_first_not_of('/') - 1);
socket = new Socket(SW_SOCK_UNIX_STREAM);
}
else if (host.find(':') != std::string::npos)
{
socket = new Socket(SW_SOCK_TCP6);
}
else
{
socket = new Socket(SW_SOCK_TCP);
}
if (sw_unlikely(socket->get_fd() < 0))
{
php_swoole_fatal_error(E_WARNING, "new Socket() failed. Error: %s [%d]", strerror(errno), errno);
non_sql_error(MYSQLND_CR_CONNECTION_ERROR, strerror(errno));
delete socket;
socket = nullptr;
return false;
}
#ifdef SW_USE_OPENSSL
socket->open_ssl = ssl;
#endif
socket->set_timeout(connect_timeout, SW_TIMEOUT_CONNECT);
add_timeout_controller(connect_timeout, SW_TIMEOUT_ALL);
if (!socket->connect(host, port))
{
io_error();
return false;
}
this->host = host;
this->port = port;
#ifdef SW_USE_OPENSSL
this->ssl = ssl;
#endif
if (!handshake())
{
close();
return false;
}
state = SW_MYSQL_STATE_IDLE;
quit = false;
del_timeout_controller();
}
return true;
}
const char* mysql_client::recv_length(size_t need_length, const bool try_to_recycle)
{
if (sw_likely(check_connection()))
{
ssize_t retval;
swString *buffer = socket->get_read_buffer();
off_t offset = buffer->offset; // save offset instead of buffer point (due to realloc)
size_t read_n = buffer->length - buffer->offset; // readable bytes
if (try_to_recycle && read_n == 0)
{
swTraceLog(SW_TRACE_MYSQL_CLIENT, "mysql buffer will be recycled, length=%zu, offset=%jd", buffer->length, (intmax_t) offset);
swString_clear(buffer);
offset = 0;
}
while (read_n < need_length)
{
if (sw_unlikely(has_timedout(SW_TIMEOUT_READ)))
{
io_error();
return nullptr;
}
if (sw_unlikely(buffer->length == buffer->size))
{
/* offset + need_length = new size (min) */
if (sw_unlikely(swString_extend(buffer, SW_MEM_ALIGNED_SIZE_EX(offset + need_length, SwooleG.pagesize)) != SW_OK))
{
non_sql_error(MYSQLND_CR_OUT_OF_MEMORY, strerror(ENOMEM));
return nullptr;
}
else
{
swTraceLog(SW_TRACE_MYSQL_CLIENT, "mysql buffer extend to %zu", buffer->size);
}
}
retval = socket->recv(buffer->str + buffer->length, buffer->size - buffer->length);
if (sw_unlikely(retval <= 0))
{
io_error();
return nullptr;
}
read_n += retval;
buffer->length += retval;
}
buffer->offset += need_length;
return buffer->str + offset;
}
return nullptr;
}
const char* mysql_client::recv_packet()
{
const char *p;
uint32_t length;
p = recv_length(SW_MYSQL_PACKET_HEADER_SIZE, true);
if (sw_unlikely(!p))
{
return nullptr;
}
length = mysql::packet::get_length(p);
swTraceLog(SW_TRACE_MYSQL_CLIENT, "recv packet length=%u, number=%u", length, mysql::packet::get_number(p));
p = recv_length(length);
if (sw_unlikely(!p))
{
return nullptr;
}
/* Notice: why we do this? because buffer maybe reallocated when recv data */
return p - SW_MYSQL_PACKET_HEADER_SIZE;
}
bool mysql_client::send_packet(mysql::client_packet *packet)
{
const char *data = packet->get_data();
uint32_t length = SW_MYSQL_PACKET_HEADER_SIZE + packet->get_length();
if (sw_likely(send_raw(data, length)))
{
return true;
}
return false;
}
bool mysql_client::send_command(enum sw_mysql_command command, const char* sql, size_t length)
{
if (sw_likely(SW_MYSQL_PACKET_HEADER_SIZE + 1 + length <= SwooleG.pagesize))
{
mysql::command_packet command_packet(command, sql, length);
return send_raw(command_packet.get_data(), command_packet.get_data_length());
}
else
{
/* if the data is larger than page_size, copy memory to the kernel buffer multiple times is much faster */
size_t send_s = SW_MIN(length, SW_MYSQL_MAX_PACKET_BODY_SIZE - 1), send_n = send_s, number = 0;
mysql::command_packet command_packet(command);
command_packet.set_header(1 + send_s, number++);
if (sw_unlikely(
!send_raw(command_packet.get_data(), SW_MYSQL_PACKET_HEADER_SIZE + 1)) ||
!send_raw(sql, send_s)
)
{
return false;
}
/* MySQL single packet size is 16M, we must subpackage */
while (send_n < length)
{
send_s = length - send_n;
send_s = SW_MIN(send_s, SW_MYSQL_MAX_PACKET_BODY_SIZE);
command_packet.set_header(send_s, number++);
if (sw_unlikely(
!send_raw(command_packet.get_data(), SW_MYSQL_PACKET_HEADER_SIZE)) ||
!send_raw(sql + send_n, send_s)
)
{
return false;
}
send_n += send_s;
}
return true;
}
}
void mysql_client::send_command_without_check(enum sw_mysql_command command, const char* sql, size_t length)
{
mysql::command_packet command_packet(command, sql, length);
(void) (socket && socket->send(command_packet.get_data(), command_packet.get_data_length()));
}
bool mysql_client::handshake()
{
const char *data;
// recv greeting pakcet
if (sw_unlikely(!(data = recv_none_error_packet())))
{
return false;
}
mysql::greeting_packet greeting_packet(data);
// generate login packet
do {
mysql::login_packet login_packet(&greeting_packet, user, password, database, charset);
if (sw_unlikely(!send_packet(&login_packet)))
{
return false;
}
} while (0);
// recv auth switch request packet, 4 possible packet types
switch (mysql::server_packet::parse_type(data = recv_packet()))
{
case SW_MYSQL_PACKET_AUTH_SWITCH_REQUEST:
{
mysql::auth_switch_request_packet request(data);
mysql::auth_switch_response_packet response(&request, password);
if (sw_unlikely(!send_packet(&response)))
{
return false;
}
break;
}
case SW_MYSQL_PACKET_AUTH_SIGNATURE_REQUEST:
{
mysql::auth_signature_request_packet request(data);
if (sw_unlikely(!request.is_vaild()))
{
goto _proto_error;
}
if (sw_likely(!request.is_full_auth_required()))
{
break;
}
// no cache, need full auth with rsa key (openssl required)
#ifdef SW_MYSQL_RSA_SUPPORT
// tell the server we are prepared
do {
mysql::auth_signature_prepared_packet prepared(request.header.number + 1);
if (sw_unlikely(!send_packet(&prepared)))
{
return false;
}
} while (0);
// recv rsa key and encode the password
do {
if (sw_unlikely(!(data = recv_none_error_packet())))
{
return false;
}
mysql::raw_data_packet raw_data_packet(data);
mysql::auth_signature_response_packet response(&raw_data_packet, password, greeting_packet.auth_plugin_data);
if (sw_unlikely(!send_packet(&response)))
{
return false;
}
} while (0);
break;
#else
error_code = EPROTONOSUPPORT;
error_msg = SW_MYSQL_NO_RSA_ERROR;
return false;
#endif
}
case SW_MYSQL_PACKET_OK:
{
#ifdef SW_LOG_TRACE_OPEN
mysql::ok_packet ok_packet(data);
#endif
return true;
}
case SW_MYSQL_PACKET_ERR:
server_error(data);
return false;
case SW_MYSQL_PACKET_NULL:
// io_error
return false;
default:
_proto_error:
proto_error(data, SW_MYSQL_PACKET_AUTH_SWITCH_REQUEST);
return false;
}
// maybe ok packet or err packet
if (sw_unlikely(!(data = recv_none_error_packet())))
{
return false;
}
#ifdef SW_LOG_TRACE_OPEN
mysql::ok_packet ok_packet(data);
#endif
return true;
}
void mysql_client::query(zval *return_value, const char *statement, size_t statement_length)
{
send_query_request(return_value, statement, statement_length);
if (EXPECTED(!defer && Z_TYPE_P(return_value) == IS_TRUE))
{
recv_query_response(return_value);
}
}
void mysql_client::send_query_request(zval *return_value, const char *statement, size_t statement_length)
{
if (sw_unlikely(!is_available_for_new_reuqest()))
{
RETURN_FALSE;
}
if (sw_unlikely(!send_command(SW_MYSQL_COM_QUERY, statement, statement_length)))
{
RETURN_FALSE;
}
state = SW_MYSQL_STATE_QUERY;
RETURN_TRUE;
};
void mysql_client::recv_query_response(zval *return_value)
{
const char *data;
if (sw_unlikely(!(data = recv_none_error_packet())))
{
RETURN_FALSE;
}
if (mysql::server_packet::is_ok(data))
{
mysql::ok_packet ok_packet(data);
result.ok = ok_packet;
state = ok_packet.server_status.more_results_exists() ? SW_MYSQL_STATE_QUERY_MORE_RESULTS : SW_MYSQL_STATE_IDLE;
RETURN_TRUE;
}
do {
mysql::lcb_packet lcb_packet(data);
if (sw_unlikely(lcb_packet.length == 0))
{
// is it possible?
proto_error(data, SW_MYSQL_PACKET_FIELD);
RETURN_FALSE;
}
result.alloc_fields(lcb_packet.length);
for (uint32_t i = 0; i < lcb_packet.length; i++)
{
if (sw_unlikely(!(data = recv_packet())))
{
RETURN_FALSE;
}
result.set_field(i, data);
}
} while (0);
// expect eof
if (sw_unlikely(!(data = recv_eof_packet())))
{
RETURN_FALSE;
}
state = SW_MYSQL_STATE_QUERY_FETCH;
if (get_fetch_mode())
{
RETURN_TRUE;
}
fetch_all(return_value);
}
const char* mysql_client::handle_row_data_size(mysql::row_data *row_data, uint8_t size)
{
const char *p, *data;
SW_ASSERT(size < sizeof(row_data->stack_buffer));
if (sw_unlikely(!(p = row_data->read(size))))
{
uint8_t received = row_data->recv(row_data->stack_buffer, size);
if (sw_unlikely(!(data = recv_packet())))
{
return nullptr;
}
row_data->next_packet(data);
received += row_data->recv(row_data->stack_buffer + received, size - received);
if (sw_unlikely(received != size))
{
proto_error(data, SW_MYSQL_PACKET_ROW_DATA);
return nullptr;
}
p = row_data->stack_buffer;
}
return p;
}
bool mysql_client::handle_row_data_lcb(mysql::row_data *row_data)
{
const char *p, *data;
// recv 1 byte to get binary code size
if (sw_unlikely(row_data->eof()))
{
if (sw_unlikely(!(data = recv_packet())))
{
return false;
}
row_data->next_packet(data);
if (sw_unlikely(row_data->eof()))
{
proto_error(data, SW_MYSQL_PACKET_ROW_DATA);
return false;
}
}
// decode lcb (use 0 to prevent read_ptr from moving)
// recv "size" bytes to get binary code length
p = handle_row_data_size(row_data, mysql::read_lcb_size(row_data->read(0)));
if (sw_unlikely(!p))
{
return false;
}
mysql::read_lcb(p, &row_data->text.length, &row_data->text.nul);
return true;
}
void mysql_client::handle_row_data_text(zval *return_value, mysql::row_data *row_data, mysql::field_packet *field)
{
const char *p, *data;
if (sw_unlikely(!handle_row_data_lcb(row_data)))
{
RETURN_FALSE;
}
if (sw_unlikely(!(p = row_data->read(row_data->text.length))))
{
size_t received = 0, required = row_data->text.length;
if (required < sizeof(row_data->stack_buffer))
{
p = handle_row_data_size(row_data, required);
if (sw_unlikely(!p))
{
RETURN_FALSE;
}
}
else
{
zend_string *zstring = zend_string_alloc(required, 0);
do {
received += row_data->recv(ZSTR_VAL(zstring) + received, required - received);
if (received == required)
{
break;
}
if (row_data->eof())
{
if (sw_unlikely(!(data = recv_packet())))
{
RETURN_FALSE;
}
row_data->next_packet(data);
}
} while (true);
ZSTR_VAL(zstring)[ZSTR_LEN(zstring)] = '\0';
RETVAL_STR(zstring);
goto _return;
}
}
if (row_data->text.nul || field->type == SW_MYSQL_TYPE_NULL)
{
swTraceLog(SW_TRACE_MYSQL_CLIENT, "%.*s is null", field->name_length, field->name);
RETURN_NULL();
}
else
{
RETVAL_STRINGL(p, row_data->text.length);
_return:
swTraceLog(
SW_TRACE_MYSQL_CLIENT, "%.*s=[%zu]%.*s%s",
field->name_length, field->name, Z_STRLEN_P(return_value),
SW_MIN(32, Z_STRLEN_P(return_value)), Z_STRVAL_P(return_value),
(Z_STRLEN_P(return_value) > 32 ? "..." : "")
);
}
}
void mysql_client::handle_strict_type(zval *ztext, mysql::field_packet *field)
{
if (sw_likely(Z_TYPE_P(ztext) == IS_STRING))
{
char *error;
switch (field->type)
{
/* String */
case SW_MYSQL_TYPE_TINY_BLOB:
case SW_MYSQL_TYPE_MEDIUM_BLOB:
case SW_MYSQL_TYPE_LONG_BLOB:
case SW_MYSQL_TYPE_BLOB:
case SW_MYSQL_TYPE_DECIMAL:
case SW_MYSQL_TYPE_NEWDECIMAL:
case SW_MYSQL_TYPE_BIT:
case SW_MYSQL_TYPE_STRING:
case SW_MYSQL_TYPE_VAR_STRING:
case SW_MYSQL_TYPE_VARCHAR:
case SW_MYSQL_TYPE_NEWDATE:
/* Date Time */
case SW_MYSQL_TYPE_TIME:
case SW_MYSQL_TYPE_YEAR:
case SW_MYSQL_TYPE_TIMESTAMP:
case SW_MYSQL_TYPE_DATETIME:
case SW_MYSQL_TYPE_DATE:
case SW_MYSQL_TYPE_JSON:
return;
/* Integer */
case SW_MYSQL_TYPE_TINY:
case SW_MYSQL_TYPE_SHORT:
case SW_MYSQL_TYPE_INT24:
case SW_MYSQL_TYPE_LONG:
if (field->flags & SW_MYSQL_UNSIGNED_FLAG)
{
ulong_t uint = strtoul(Z_STRVAL_P(ztext), &error, 10);
if (sw_likely(*error == '\0'))
{
zend_string_release(Z_STR_P(ztext));
ZVAL_LONG(ztext, uint);
}
}
else
{
long sint = strtol(Z_STRVAL_P(ztext), &error, 10);
if (sw_likely(*error == '\0'))
{
zend_string_release(Z_STR_P(ztext));
ZVAL_LONG(ztext, sint);
}
}
break;
case SW_MYSQL_TYPE_LONGLONG:
if (field->flags & SW_MYSQL_UNSIGNED_FLAG)
{
unsigned long long ubigint = strtoull(Z_STRVAL_P(ztext), &error, 10);
if (sw_likely(*error == '\0' && ubigint <= ZEND_LONG_MAX))
{
zend_string_release(Z_STR_P(ztext));
ZVAL_LONG(ztext, ubigint);
}
}
else
{
long long sbigint = strtoll(Z_STRVAL_P(ztext), &error, 10);
if (sw_likely(*error == '\0'))
{
zend_string_release(Z_STR_P(ztext));
ZVAL_LONG(ztext, sbigint);
}
}
break;
case SW_MYSQL_TYPE_FLOAT:
case SW_MYSQL_TYPE_DOUBLE:
{
double mdouble = strtod(Z_STRVAL_P(ztext), &error);
if (sw_likely(*error == '\0'))
{
zend_string_release(Z_STR_P(ztext));
ZVAL_DOUBLE(ztext, mdouble);
}
break;
}
default:
{
swWarn("unknown type[%d] for field [%.*s].", field->type, field->name_length, field->name);
break;
}
}
}
}
void mysql_client::fetch(zval *return_value)
{
if (sw_unlikely(!is_connect()))
{
RETURN_FALSE;
}
if (sw_unlikely(state != SW_MYSQL_STATE_QUERY_FETCH))
{
RETURN_NULL();
}
const char *data;
if (sw_unlikely(!(data = recv_packet())))
{
RETURN_FALSE;
}
if (mysql::server_packet::is_eof(data))
{
mysql::eof_packet eof_packet(data);
state = eof_packet.server_status.more_results_exists() ? SW_MYSQL_STATE_QUERY_MORE_RESULTS : SW_MYSQL_STATE_IDLE;
RETURN_NULL();
}
do {
mysql::row_data row_data(data);
array_init_size(return_value, result.get_fields_length());
for (uint32_t i = 0; i < result.get_fields_length(); i++)
{
mysql::field_packet *field = result.get_field(i);
zval ztext;
handle_row_data_text(&ztext, &row_data, field);
if (sw_unlikely(Z_TYPE_P(&ztext) == IS_FALSE))
{
zval_ptr_dtor(return_value);
RETURN_FALSE;
}
if (strict_type)
{
handle_strict_type(&ztext, field);
}
add_assoc_zval_ex(return_value, field->name, field->name_length, &ztext);
}
} while (0);
}
void mysql_client::fetch_all(zval *return_value)
{
array_init(return_value);
while (true)
{
zval zrow;
fetch(&zrow);
if (sw_unlikely(ZVAL_IS_NULL(&zrow)))
{
// eof
return;
}
if (sw_unlikely(Z_TYPE_P(&zrow) == IS_FALSE))
{
// error
zval_ptr_dtor(return_value);
RETURN_FALSE;
}
(void) add_next_index_zval(return_value, &zrow);
}
}
void mysql_client::next_result(zval *return_value)
{
if (sw_unlikely(state == SW_MYSQL_STATE_QUERY_FETCH))
{
// skip unread data
fetch_all(return_value);
zval_ptr_dtor(return_value);
next_result(return_value);
}
else if (sw_likely(state == SW_MYSQL_STATE_QUERY_MORE_RESULTS))
{
recv_query_response(return_value);
}
else if (state == SW_MYSQL_STATE_IDLE)
{
RETURN_NULL();
}
else
{
RETURN_FALSE;
}
}
bool mysql_client::send_prepare_request(const char *statement, size_t statement_length)
{
this->statement = new mysql_statement(this, statement, statement_length);
if (sw_unlikely(!this->statement->send_prepare_request()))
{
delete this->statement;
this->statement = nullptr;
return false;
}
return true;
}
mysql_statement* mysql_client::recv_prepare_response()
{
if (sw_likely(state == SW_MYSQL_STATE_PREPARE))
{
mysql_statement *statement = this->statement;
SW_ASSERT(statement != nullptr);
this->statement = nullptr;
if (sw_unlikely(!statement->recv_prepare_response()))
{
delete statement;
return nullptr;
}
statements[statement->info.id] = statement;
return statement;
}
return nullptr;
}
void mysql_client::close()
{
state = SW_MYSQL_STATE_CLOSED;
Socket *socket = this->socket;
if (socket)
{
del_timeout_controller();
if (!quit && is_writable())
{
send_command_without_check(SW_MYSQL_COM_QUIT);
quit = true;
}
// make statements non-available
while (!statements.empty())
{
auto i = statements.begin();
i->second->close(false);
statements.erase(i);
}
if (sw_likely(!socket->has_bound()))
{
this->socket = nullptr;
}
if (sw_likely(socket->close()))
{
delete socket;
}
}
}
bool mysql_statement::send_prepare_request()
{
if (sw_unlikely(!is_available_for_new_reuqest()))
{
return false;
}
if (sw_unlikely(!client->send_command(SW_MYSQL_COM_STMT_PREPARE, statement.c_str(), statement.length())))
{
return false;
}
client->state = SW_MYSQL_STATE_PREPARE;
return true;
}
bool mysql_statement::recv_prepare_response()
{
if (sw_unlikely(!is_available()))
{
return false;
}
else
{
client->state = SW_MYSQL_STATE_IDLE;
}
const char *data;
if (sw_unlikely(!(data = client->recv_none_error_packet())))
{
return false;
}
info = mysql::statement(data);
if (sw_likely(info.param_count != 0))
{
for (uint16_t i = info.param_count; i--;)
{
if (sw_unlikely(!(data = client->recv_packet())))
{
return false;
}
#ifdef SW_LOG_TRACE_OPEN
mysql::param_packet param_packet(data);
#endif
}
if (sw_unlikely(!(data = client->recv_eof_packet())))
{
return false;
}
}
if (info.field_count != 0)
{
result.alloc_fields(info.field_count);
for (uint16_t i = 0; i < info.field_count; i++)
{
if (sw_unlikely(!(data = client->recv_packet())))
{
return false;
}
result.set_field(i, data);
}
if (sw_unlikely(!(data = client->recv_eof_packet())))
{
return false;
}
}
return true;
}
void mysql_statement::execute(zval *return_value, zval *params)
{
send_execute_request(return_value, params);
/* Notice: must check return_value first */
if (EXPECTED(Z_TYPE_P(return_value) == IS_TRUE && !client->get_defer()))
{
recv_execute_response(return_value);
}
}
void mysql_statement::send_execute_request(zval *return_value, zval *params)
{
if (sw_unlikely(!is_available_for_new_reuqest()))
{
RETURN_FALSE;
}
uint32_t param_count = params ? php_swoole_array_length(params) : 0;
if (sw_unlikely(param_count != info.param_count))
{
client->non_sql_error(
MYSQLND_CR_INVALID_PARAMETER_NO,
"Statement#%u expects %u parameter, %u given.",
info.id, info.param_count, param_count
);
RETURN_FALSE;
}
swString *buffer = SwooleTG.buffer_stack;
char *p = buffer->str;
memset(p, 0, 5);
// command
buffer->str[4] = SW_MYSQL_COM_STMT_EXECUTE;
buffer->length = 5;
p += 5;
// stmt.id
sw_mysql_int4store(p, info.id);
p += 4;
// flags = CURSOR_TYPE_NO_CURSOR
sw_mysql_int1store(p, 0);
p += 1;
// iteration_count
sw_mysql_int4store(p, 1);
p += 4;
buffer->length += 9;
// TODO: support more types
if (param_count != 0)
{
// null bitmap
size_t null_start_offset = p - buffer->str;
unsigned int map_size = (param_count + 7) / 8;
memset(p, 0, map_size);
p += map_size;
buffer->length += map_size;
// rebind
sw_mysql_int1store(p, 1);
p += 1;
buffer->length += 1;
size_t type_start_offset = p - buffer->str;
p += param_count * 2;
buffer->length += param_count * 2;
char stack_buffer[10];
zend_ulong index = 0;
zval *value;
ZEND_HASH_FOREACH_VAL(Z_ARRVAL_P(params), value)
{
if (ZVAL_IS_NULL(value))
{
*((buffer->str + null_start_offset) + (index / 8)) |= (1UL << (index % 8));
sw_mysql_int2store((buffer->str + type_start_offset) + (index * 2), SW_MYSQL_TYPE_NULL);
}
else
{
zend::string str_value(value);
uint8_t lcb_size = mysql::write_lcb(stack_buffer, str_value.len());
sw_mysql_int2store((buffer->str + type_start_offset) + (index * 2), SW_MYSQL_TYPE_VAR_STRING);
if (swString_append_ptr(buffer, stack_buffer, lcb_size) < 0)
{
RETURN_FALSE;
}
if (swString_append_ptr(buffer, str_value.val(), str_value.len()) < 0)
{
RETURN_FALSE;
}
}
index++;
}
ZEND_HASH_FOREACH_END();
}
do {
size_t length = buffer->length - SW_MYSQL_PACKET_HEADER_SIZE;
size_t send_s = SW_MIN(length, SW_MYSQL_MAX_PACKET_BODY_SIZE);
mysql::packet::set_header(buffer->str, send_s, 0);
if (sw_unlikely(!client->send_raw(buffer->str, SW_MYSQL_PACKET_HEADER_SIZE + send_s)))
{
RETURN_FALSE;
}
if (sw_unlikely(length > SW_MYSQL_MAX_PACKET_BODY_SIZE))
{
size_t send_n = SW_MYSQL_MAX_PACKET_BODY_SIZE, number = 1;
/* MySQL single packet size is 16M, we must subpackage */
while (send_n < length)
{
send_s = length - send_n;
send_s = SW_MIN(send_s, SW_MYSQL_MAX_PACKET_BODY_SIZE);
mysql::packet::set_header(buffer->str, send_s, number++);
if (sw_unlikely(
!client->send_raw(buffer->str, SW_MYSQL_PACKET_HEADER_SIZE)) ||
!client->send_raw(buffer->str + SW_MYSQL_PACKET_HEADER_SIZE + send_n, send_s)
)
{
RETURN_FALSE;
}
send_n += send_s;
}
}
} while (0);
client->state = SW_MYSQL_STATE_EXECUTE;
RETURN_TRUE;
}
void mysql_statement::recv_execute_response(zval *return_value)
{
if (sw_unlikely(!is_available()))
{
RETURN_FALSE;
}
const char *data;
if (sw_unlikely(!(data = client->recv_none_error_packet())))
{
RETURN_FALSE;
}
if (mysql::server_packet::is_ok(data))
{
mysql::ok_packet ok_packet(data);
result.ok = ok_packet;
client->state = ok_packet.server_status.more_results_exists() ? SW_MYSQL_STATE_EXECUTE_MORE_RESULTS : SW_MYSQL_STATE_IDLE;
RETURN_TRUE;
}
do {
mysql::lcb_packet lcb_packet(data);
if (sw_unlikely(lcb_packet.length == 0))
{
// is it possible?
client->proto_error(data, SW_MYSQL_PACKET_FIELD);
RETURN_FALSE;
}
// although we have already known the field data when we prepared the statement,
// we don't know if the data is always reliable, such as when we using stored procedure...
// so we should not optimize here for the time being for stability
result.alloc_fields(lcb_packet.length);
for (size_t i = 0; i < result.get_fields_length(); i++)
{
if (sw_unlikely(!(data = client->recv_packet())))
{
RETURN_FALSE;
}
result.set_field(i, data);
}
} while (0);
// expect eof
if (sw_unlikely(!(data = client->recv_eof_packet())))
{
RETURN_FALSE;
}
client->state = SW_MYSQL_STATE_EXECUTE_FETCH;
if (client->get_fetch_mode())
{
RETURN_TRUE;
}
fetch_all(return_value);
}
void mysql_statement::fetch(zval *return_value)
{
if (sw_unlikely(!is_available()))
{
RETURN_FALSE;
}
if (sw_unlikely(client->state != SW_MYSQL_STATE_EXECUTE_FETCH))
{
RETURN_NULL();
}
const char *data;
if (sw_unlikely(!(data = client->recv_packet())))
{
RETURN_FALSE;
}
if (mysql::server_packet::is_eof(data))
{
mysql::eof_packet eof_packet(data);
client->state = eof_packet.server_status.more_results_exists() ? SW_MYSQL_STATE_EXECUTE_MORE_RESULTS : SW_MYSQL_STATE_IDLE;
RETURN_NULL();
}
do {
mysql::row_data row_data(data);
uint32_t null_bitmap_size = mysql::null_bitmap::get_size(result.get_fields_length());
mysql::null_bitmap null_bitmap(row_data.read(null_bitmap_size), null_bitmap_size);
array_init_size(return_value, result.get_fields_length());
for (uint32_t i = 0; i < result.get_fields_length(); i++)
{
mysql::field_packet *field = result.get_field(i);
/* to check Null-Bitmap @see https://dev.mysql.com/doc/internals/en/null-bitmap.html */
if (null_bitmap.is_null(i) || field->type == SW_MYSQL_TYPE_NULL)
{
swTraceLog(SW_TRACE_MYSQL_CLIENT, "%.*s is null", field->name_length, field->name);
add_assoc_null_ex(return_value, field->name, field->name_length);
continue;
}
switch (field->type)
{
/* String */
case SW_MYSQL_TYPE_TINY_BLOB:
case SW_MYSQL_TYPE_MEDIUM_BLOB:
case SW_MYSQL_TYPE_LONG_BLOB:
case SW_MYSQL_TYPE_BLOB:
case SW_MYSQL_TYPE_DECIMAL:
case SW_MYSQL_TYPE_NEWDECIMAL:
case SW_MYSQL_TYPE_BIT:
case SW_MYSQL_TYPE_JSON:
case SW_MYSQL_TYPE_STRING:
case SW_MYSQL_TYPE_VAR_STRING:
case SW_MYSQL_TYPE_VARCHAR:
case SW_MYSQL_TYPE_NEWDATE:
{
_add_string:
zval ztext;
client->handle_row_data_text(&ztext, &row_data, field);
if (sw_unlikely(Z_TYPE_P(&ztext) == IS_FALSE))
{
zval_ptr_dtor(return_value);
RETURN_FALSE;
}
add_assoc_zval_ex(return_value, field->name, field->name_length, &ztext);
break;
}
default:
{
const char *p = nullptr;
uint8_t lcb = mysql::get_static_type_size(field->type);
if (lcb == 0)
{
client->handle_row_data_lcb(&row_data);
lcb = row_data.text.length;
}
p = client->handle_row_data_size(&row_data, lcb);
if (sw_unlikely(!p))
{
zval_ptr_dtor(return_value);
RETURN_FALSE;
}
/* Date Time */
switch(field->type)
{
case SW_MYSQL_TYPE_TIMESTAMP:
case SW_MYSQL_TYPE_DATETIME:
{
std::string datetime = mysql::datetime(p, row_data.text.length, field->decimals);
add_assoc_stringl_ex(return_value, field->name, field->name_length, (char *) datetime.c_str(), datetime.length());
swTraceLog(SW_TRACE_MYSQL_CLIENT, "%.*s=%s", field->name_length, field->name, datetime.c_str());
break;
}
case SW_MYSQL_TYPE_TIME:
{
std::string time = mysql::time(p, row_data.text.length, field->decimals);
add_assoc_stringl_ex(return_value, field->name, field->name_length, (char *) time.c_str(), time.length());
swTraceLog(SW_TRACE_MYSQL_CLIENT, "%.*s=%s", field->name_length, field->name, time.c_str());
break;
}
case SW_MYSQL_TYPE_DATE:
{
std::string date = mysql::date(p, row_data.text.length);
add_assoc_stringl_ex(return_value, field->name, field->name_length, (char *) date.c_str(), date.length());
swTraceLog(SW_TRACE_MYSQL_CLIENT, "%.*s=%s", field->name_length, field->name, date.c_str());
break;
}
case SW_MYSQL_TYPE_YEAR:
{
add_assoc_long_ex(return_value, field->name, field->name_length, sw_mysql_uint2korr2korr(p));
swTraceLog(SW_TRACE_MYSQL_CLIENT, "%.*s=%d", field->name_length, field->name, sw_mysql_uint2korr2korr(p));
break;
}
/* Number */
case SW_MYSQL_TYPE_TINY:
if (field->flags & SW_MYSQL_UNSIGNED_FLAG)
{
add_assoc_long_ex(return_value, field->name, field->name_length, *(uint8_t *) p);
swTraceLog(SW_TRACE_MYSQL_CLIENT, "%.*s=%u", field->name_length, field->name, *(uint8_t *) p);
}
else
{
add_assoc_long_ex(return_value, field->name, field->name_length, *(int8_t *) p);
swTraceLog(SW_TRACE_MYSQL_CLIENT, "%.*s=%d", field->name_length, field->name, *(int8_t *) p);
}
break;
case SW_MYSQL_TYPE_SHORT:
if (field->flags & SW_MYSQL_UNSIGNED_FLAG)
{
add_assoc_long_ex(return_value, field->name, field->name_length, *(uint16_t *) p);
swTraceLog(SW_TRACE_MYSQL_CLIENT, "%.*s=%u", field->name_length, field->name, *(uint16_t *) p);
}
else
{
add_assoc_long_ex(return_value, field->name, field->name_length, *(int16_t *) p);
swTraceLog(SW_TRACE_MYSQL_CLIENT, "%.*s=%d", field->name_length, field->name, *(int16_t *) p);
}
break;
case SW_MYSQL_TYPE_INT24:
case SW_MYSQL_TYPE_LONG:
if (field->flags & SW_MYSQL_UNSIGNED_FLAG)
{
add_assoc_long_ex(return_value, field->name, field->name_length, *(uint32_t *) p);
swTraceLog(SW_TRACE_MYSQL_CLIENT, "%.*s=%u", field->name_length, field->name, *(uint32_t *) p);
}
else
{
add_assoc_long_ex(return_value, field->name, field->name_length, *(int32_t *) p);
swTraceLog(SW_TRACE_MYSQL_CLIENT, "%.*s=%d", field->name_length, field->name, *(int32_t *) p);
}
break;
case SW_MYSQL_TYPE_LONGLONG:
if (field->flags & SW_MYSQL_UNSIGNED_FLAG)
{
add_assoc_ulong_safe_ex(return_value, field->name, field->name_length, *(uint64_t *) p);
swTraceLog(SW_TRACE_MYSQL_CLIENT, "%.*s=%llu", field->name_length, field->name, *(uint64_t *) p);
}
else
{
add_assoc_long_ex(return_value, field->name, field->name_length, *(int64_t *) p);
swTraceLog(SW_TRACE_MYSQL_CLIENT, "%.*s=%lld", field->name_length, field->name, *(int64_t *) p);
}
break;
case SW_MYSQL_TYPE_FLOAT:
{
double dv = sw_php_math_round(*(float *) p, 5, PHP_ROUND_HALF_DOWN);
add_assoc_double_ex(return_value, field->name, field->name_length, dv);
swTraceLog(SW_TRACE_MYSQL_CLIENT, "%.*s=%.7f", field->name_length, field->name, dv);
}
break;
case SW_MYSQL_TYPE_DOUBLE:
{
add_assoc_double_ex(return_value, field->name, field->name_length, *(double *) p);
swTraceLog(SW_TRACE_MYSQL_CLIENT, "%.*s=%.16f", field->name_length, field->name, *(double *) p);
}
break;
default:
swWarn("unknown type[%d] for field [%.*s].", field->type, field->name_length, field->name);
goto _add_string;
}
}
}
}
} while (0);
}
void mysql_statement::fetch_all(zval *return_value)
{
if (sw_unlikely(!is_available()))
{
RETURN_FALSE;
}
zval zrow;
array_init(return_value);
while (true)
{
fetch(&zrow);
if (sw_unlikely(ZVAL_IS_NULL(&zrow)))
{
// eof
return;
}
if (sw_unlikely(Z_TYPE_P(&zrow) == IS_FALSE))
{
// error
zval_ptr_dtor(return_value);
RETURN_FALSE;
}
(void) add_next_index_zval(return_value, &zrow);
}
}
void mysql_statement::next_result(zval *return_value)
{
if (sw_unlikely(!is_available()))
{
RETURN_FALSE;
}
if (sw_unlikely(client->state == SW_MYSQL_STATE_EXECUTE_FETCH))
{
// skip unread data
fetch_all(return_value);
zval_ptr_dtor(return_value);
next_result(return_value);
}
else if (sw_likely(client->state == SW_MYSQL_STATE_EXECUTE_MORE_RESULTS))
{
recv_execute_response(return_value);
}
else if (client->state == SW_MYSQL_STATE_IDLE)
{
RETURN_NULL();
}
else
{
RETURN_FALSE;
}
}
static sw_inline mysql_coro_t* swoole_mysql_coro_fetch_object(zend_object *obj)
{
return (mysql_coro_t *) ((char *) obj - swoole_mysql_coro_handlers.offset);
}
static sw_inline mysql_client* swoole_get_mysql_client(zval *zobject)
{
return swoole_mysql_coro_fetch_object(Z_OBJ_P(zobject))->client;
}
static void swoole_mysql_coro_free_object(zend_object *object)
{
mysql_coro_t *zmc = swoole_mysql_coro_fetch_object(object);
delete zmc->client;
zend_object_std_dtor(&zmc->std);
}
static zend_object *swoole_mysql_coro_create_object(zend_class_entry *ce)
{
mysql_coro_t *zmc = (mysql_coro_t *) ecalloc(1, sizeof(mysql_coro_t) + zend_object_properties_size(ce));
zend_object_std_init(&zmc->std, ce);
object_properties_init(&zmc->std, ce);
zmc->std.handlers = &swoole_mysql_coro_handlers;
zmc->client = new mysql_client;
return &zmc->std;
}
static sw_inline mysql_coro_statement_t* swoole_mysql_coro_statement_fetch_object(zend_object *obj)
{
return (mysql_coro_statement_t *) ((char *) obj - swoole_mysql_coro_statement_handlers.offset);
}
static sw_inline mysql_statement* swoole_get_mysql_statement(zval *zobject)
{
return swoole_mysql_coro_statement_fetch_object(Z_OBJ_P(zobject))->statement;
}
static void swoole_mysql_coro_statement_free_object(zend_object *object)
{
mysql_coro_statement_t *zms = swoole_mysql_coro_statement_fetch_object(object);
delete zms->statement;
OBJ_RELEASE(zms->zclient);
zend_object_std_dtor(&zms->std);
}
static sw_inline zend_object* swoole_mysql_coro_statement_create_object(zend_class_entry *ce, mysql_statement *statement, zend_object *client)
{
zval zobject;
mysql_coro_statement_t *zms = (mysql_coro_statement_t *) ecalloc(1, sizeof(mysql_coro_statement_t) + zend_object_properties_size(ce));
zend_object_std_init(&zms->std, ce);
object_properties_init(&zms->std, ce);
zms->std.handlers = &swoole_mysql_coro_statement_handlers;
ZVAL_OBJ(&zobject, &zms->std);
zend_update_property_long(ce, &zobject, ZEND_STRL("id"), statement->info.id);
zms->statement = statement;
zms->zclient = client;
GC_ADDREF(client);
return &zms->std;
}
static sw_inline zend_object* swoole_mysql_coro_statement_create_object(mysql_statement *statement, zend_object *client)
{
return swoole_mysql_coro_statement_create_object(swoole_mysql_coro_statement_ce, statement, client);
}
static zend_object* swoole_mysql_coro_statement_create_object(zend_class_entry *ce)
{
php_swoole_fatal_error(E_ERROR, "you must create mysql statement object by prepare method");
return nullptr;
}
static sw_inline void swoole_mysql_coro_sync_error_properties(zval *zobject, int error_code, const char *error_msg, const bool connected = true)
{
SW_ASSERT(instanceof_function(Z_OBJCE_P(zobject), swoole_mysql_coro_ce) || instanceof_function(Z_OBJCE_P(zobject), swoole_mysql_coro_statement_ce));
zend_update_property_long(Z_OBJCE_P(zobject), zobject, ZEND_STRL("errno"), error_code);
zend_update_property_string(Z_OBJCE_P(zobject), zobject, ZEND_STRL("error"), error_msg);
if (!connected)
{
zend_update_property_bool(Z_OBJCE_P(zobject), zobject, ZEND_STRL("connected"), connected);
}
}
static sw_inline void swoole_mysql_coro_sync_query_result_properties(zval *zobject, mysql_client *mc, zval *return_value)
{
switch (Z_TYPE_P(return_value))
{
case IS_TRUE:
{
mysql::ok_packet *ok_packet = &mc->result.ok;
zend_update_property_long(Z_OBJCE_P(zobject), zobject, ZEND_STRL("affected_rows"), ok_packet->affected_rows);
zend_update_property_long(Z_OBJCE_P(zobject), zobject, ZEND_STRL("insert_id"), ok_packet->last_insert_id);
break;
}
case IS_FALSE:
{
swoole_mysql_coro_sync_error_properties(zobject, mc->get_error_code(), mc->get_error_msg());
break;
}
default:
break;
}
}
static sw_inline void swoole_mysql_coro_sync_execute_error_properties(zval *zobject, int error_code, const char *error_msg, const bool connected = true)
{
swoole_mysql_coro_sync_error_properties(zobject, error_code, error_msg, connected);
/* backward compatibility (sync error info to client) */
zval zclient;
ZVAL_OBJ(&zclient, swoole_mysql_coro_statement_fetch_object(Z_OBJ_P(zobject))->zclient);
swoole_mysql_coro_sync_error_properties(&zclient, error_code, error_msg, connected);
}
static sw_inline void swoole_mysql_coro_sync_execute_result_properties(zval *zobject, zval *return_value)
{
mysql_coro_statement_t *zms = swoole_mysql_coro_statement_fetch_object(Z_OBJ_P(zobject));
mysql_statement *ms = zms->statement;
switch (Z_TYPE_P(return_value))
{
case IS_TRUE:
{
mysql::ok_packet *ok_packet = &ms->result.ok;
zend_update_property_long(Z_OBJCE_P(zobject), zobject, ZEND_STRL("affected_rows"), ok_packet->affected_rows);
zend_update_property_long(Z_OBJCE_P(zobject), zobject, ZEND_STRL("insert_id"), ok_packet->last_insert_id);
/* backward compatibility (sync result info to client) */
zval zclient;
ZVAL_OBJ(&zclient, zms->zclient);
zend_update_property_long(Z_OBJCE_P(&zclient), &zclient, ZEND_STRL("affected_rows"), ok_packet->affected_rows);
zend_update_property_long(Z_OBJCE_P(&zclient), &zclient, ZEND_STRL("insert_id"), ok_packet->last_insert_id);
break;
}
case IS_FALSE:
{
swoole_mysql_coro_sync_execute_error_properties(zobject, ms->get_error_code(), ms->get_error_msg());
break;
}
default:
break;
}
}
void php_swoole_mysql_coro_minit(int module_number)
{
SW_INIT_CLASS_ENTRY(swoole_mysql_coro, "Swoole\\Coroutine\\MySQL", NULL, "Co\\MySQL", swoole_mysql_coro_methods);
SW_SET_CLASS_SERIALIZABLE(swoole_mysql_coro, zend_class_serialize_deny, zend_class_unserialize_deny);
SW_SET_CLASS_CLONEABLE(swoole_mysql_coro, sw_zend_class_clone_deny);
SW_SET_CLASS_UNSET_PROPERTY_HANDLER(swoole_mysql_coro, sw_zend_class_unset_property_deny);
SW_SET_CLASS_CUSTOM_OBJECT(swoole_mysql_coro, swoole_mysql_coro_create_object, swoole_mysql_coro_free_object, mysql_coro_t, std);
SW_INIT_CLASS_ENTRY(swoole_mysql_coro_statement, "Swoole\\Coroutine\\MySQL\\Statement", NULL, "Co\\MySQL\\Statement", swoole_mysql_coro_statement_methods);
SW_SET_CLASS_SERIALIZABLE(swoole_mysql_coro_statement, zend_class_serialize_deny, zend_class_unserialize_deny);
SW_SET_CLASS_CLONEABLE(swoole_mysql_coro_statement, sw_zend_class_clone_deny);
SW_SET_CLASS_UNSET_PROPERTY_HANDLER(swoole_mysql_coro_statement, sw_zend_class_unset_property_deny);
SW_SET_CLASS_CUSTOM_OBJECT(swoole_mysql_coro_statement, swoole_mysql_coro_statement_create_object, swoole_mysql_coro_statement_free_object, mysql_coro_statement_t, std);
SW_INIT_CLASS_ENTRY_EX(swoole_mysql_coro_exception, "Swoole\\Coroutine\\MySQL\\Exception", NULL, "Co\\MySQL\\Exception", NULL, swoole_exception);
SW_SET_CLASS_SERIALIZABLE(swoole_mysql_coro_exception, zend_class_serialize_deny, zend_class_unserialize_deny);
SW_SET_CLASS_CLONEABLE(swoole_mysql_coro_exception, sw_zend_class_clone_deny);
SW_SET_CLASS_UNSET_PROPERTY_HANDLER(swoole_mysql_coro_exception, sw_zend_class_unset_property_deny);
SW_SET_CLASS_CREATE_WITH_ITS_OWN_HANDLERS(swoole_mysql_coro_exception);
zend_declare_property_null(swoole_mysql_coro_ce, ZEND_STRL("serverInfo"), ZEND_ACC_PUBLIC);
zend_declare_property_long(swoole_mysql_coro_ce, ZEND_STRL("sock"), -1, ZEND_ACC_PUBLIC);
zend_declare_property_bool(swoole_mysql_coro_ce, ZEND_STRL("connected"), 0, ZEND_ACC_PUBLIC);
zend_declare_property_long(swoole_mysql_coro_ce, ZEND_STRL("connect_errno"), 0, ZEND_ACC_PUBLIC);
zend_declare_property_string(swoole_mysql_coro_ce, ZEND_STRL("connect_error"), "", ZEND_ACC_PUBLIC);
zend_declare_property_long(swoole_mysql_coro_ce, ZEND_STRL("affected_rows"), 0, ZEND_ACC_PUBLIC);
zend_declare_property_long(swoole_mysql_coro_ce, ZEND_STRL("insert_id"), 0, ZEND_ACC_PUBLIC);
zend_declare_property_string(swoole_mysql_coro_ce, ZEND_STRL("error"), "", ZEND_ACC_PUBLIC);
zend_declare_property_long(swoole_mysql_coro_ce, ZEND_STRL("errno"), 0, ZEND_ACC_PUBLIC);
zend_declare_property_long(swoole_mysql_coro_statement_ce, ZEND_STRL("id"), 0, ZEND_ACC_PUBLIC);
zend_declare_property_long(swoole_mysql_coro_statement_ce, ZEND_STRL("affected_rows"), 0, ZEND_ACC_PUBLIC);
zend_declare_property_long(swoole_mysql_coro_statement_ce, ZEND_STRL("insert_id"), 0, ZEND_ACC_PUBLIC);
zend_declare_property_string(swoole_mysql_coro_statement_ce, ZEND_STRL("error"), "", ZEND_ACC_PUBLIC);
zend_declare_property_long(swoole_mysql_coro_statement_ce, ZEND_STRL("errno"), 0, ZEND_ACC_PUBLIC);
SW_REGISTER_LONG_CONSTANT("SWOOLE_MYSQLND_CR_UNKNOWN_ERROR", MYSQLND_CR_UNKNOWN_ERROR);
SW_REGISTER_LONG_CONSTANT("SWOOLE_MYSQLND_CR_CONNECTION_ERROR", MYSQLND_CR_CONNECTION_ERROR);
SW_REGISTER_LONG_CONSTANT("SWOOLE_MYSQLND_CR_SERVER_GONE_ERROR", MYSQLND_CR_SERVER_GONE_ERROR);
SW_REGISTER_LONG_CONSTANT("SWOOLE_MYSQLND_CR_OUT_OF_MEMORY", MYSQLND_CR_OUT_OF_MEMORY);
SW_REGISTER_LONG_CONSTANT("SWOOLE_MYSQLND_CR_SERVER_LOST", MYSQLND_CR_SERVER_LOST);
SW_REGISTER_LONG_CONSTANT("SWOOLE_MYSQLND_CR_COMMANDS_OUT_OF_SYNC", MYSQLND_CR_COMMANDS_OUT_OF_SYNC);
SW_REGISTER_LONG_CONSTANT("SWOOLE_MYSQLND_CR_CANT_FIND_CHARSET", MYSQLND_CR_CANT_FIND_CHARSET);
SW_REGISTER_LONG_CONSTANT("SWOOLE_MYSQLND_CR_MALFORMED_PACKET", MYSQLND_CR_MALFORMED_PACKET);
SW_REGISTER_LONG_CONSTANT("SWOOLE_MYSQLND_CR_NOT_IMPLEMENTED", MYSQLND_CR_NOT_IMPLEMENTED);
SW_REGISTER_LONG_CONSTANT("SWOOLE_MYSQLND_CR_NO_PREPARE_STMT", MYSQLND_CR_NO_PREPARE_STMT);
SW_REGISTER_LONG_CONSTANT("SWOOLE_MYSQLND_CR_PARAMS_NOT_BOUND", MYSQLND_CR_PARAMS_NOT_BOUND);
SW_REGISTER_LONG_CONSTANT("SWOOLE_MYSQLND_CR_INVALID_PARAMETER_NO", MYSQLND_CR_INVALID_PARAMETER_NO);
SW_REGISTER_LONG_CONSTANT("SWOOLE_MYSQLND_CR_INVALID_BUFFER_USE", MYSQLND_CR_INVALID_BUFFER_USE);
}
static PHP_METHOD(swoole_mysql_coro, __construct) { }
static PHP_METHOD(swoole_mysql_coro, __destruct) { }
static PHP_METHOD(swoole_mysql_coro, connect)
{
mysql_client *mc = swoole_get_mysql_client(ZEND_THIS);
zval *zserver_info = nullptr;
ZEND_PARSE_PARAMETERS_START(0, 1)
Z_PARAM_OPTIONAL
Z_PARAM_ARRAY_EX(zserver_info, 1, 0)
ZEND_PARSE_PARAMETERS_END_EX(RETURN_FALSE);
if (zserver_info)
{
HashTable *ht = Z_ARRVAL_P(zserver_info);
zval *ztmp;
if (php_swoole_array_get_value(ht, "host", ztmp))
{
mc->host = std::string(zend::string(ztmp).val());
}
else
{
zend_throw_exception(swoole_mysql_coro_exception_ce, "Parameter [host] is required", EINVAL);
RETURN_FALSE;
}
if (php_swoole_array_get_value(ht, "port", ztmp))
{
mc->port = zval_get_long(ztmp);
}
if (php_swoole_array_get_value(ht, "ssl", ztmp))
{
mc->ssl = zval_is_true(ztmp);
#ifndef SW_USE_OPENSSL
if (sw_unlikely(mc->ssl))
{
zend_throw_exception_ex(
swoole_mysql_coro_exception_ce,
EPROTONOSUPPORT, "you must configure with `enable-openssl` to support ssl connection"
);
RETURN_FALSE;
}
#endif
}
if (php_swoole_array_get_value(ht, "user", ztmp))
{
mc->user = std::string(zend::string(ztmp).val());
}
else
{
zend_throw_exception(swoole_mysql_coro_exception_ce, "Parameter [user] is required", EINVAL);
RETURN_FALSE;
}
if (php_swoole_array_get_value(ht, "password", ztmp))
{
mc->password = std::string(zend::string(ztmp).val());
}
else
{
zend_throw_exception(swoole_mysql_coro_exception_ce, "Parameter [password] is required", EINVAL);
RETURN_FALSE;
}
if (php_swoole_array_get_value(ht, "database", ztmp))
{
mc->database = std::string(zend::string(ztmp).val());
}
else
{
zend_throw_exception(swoole_mysql_coro_exception_ce, "Parameter [database] is required", EINVAL);
RETURN_FALSE;
}
if (php_swoole_array_get_value(ht, "timeout", ztmp))
{
mc->connect_timeout = zval_get_double(ztmp);
}
if (php_swoole_array_get_value(ht, "charset", ztmp))
{
zend::string zstr_charset(ztmp);
char charset = mysql::get_charset(zstr_charset.val());
if (UNEXPECTED(charset < 0))
{
zend_throw_exception_ex(swoole_mysql_coro_exception_ce, EINVAL, "Unknown charset [%s]", zstr_charset.val());
RETURN_FALSE;
}
mc->charset = charset;
}
if (php_swoole_array_get_value(ht, "strict_type", ztmp))
{
mc->strict_type = zval_is_true(ztmp);
}
if (php_swoole_array_get_value(ht, "fetch_mode", ztmp))
{
if (UNEXPECTED(!mc->set_fetch_mode(zval_is_true(ztmp))))
{
zend_throw_exception_ex(swoole_mysql_coro_exception_ce, mc->get_error_code(), "%s", mc->get_error_msg());
RETURN_FALSE;
}
}
}
if (!mc->connect())
{
zend_update_property_long(swoole_mysql_coro_ce, ZEND_THIS, ZEND_STRL("connect_errno"), mc->get_error_code());
zend_update_property_string(swoole_mysql_coro_ce, ZEND_THIS, ZEND_STRL("connect_error"), mc->get_error_msg());
RETURN_FALSE;
}
if (zserver_info && php_swoole_array_length(zserver_info) > 0)
{
php_array_merge(
Z_ARRVAL_P(sw_zend_read_and_convert_property_array(swoole_mysql_coro_ce, ZEND_THIS, ZEND_STRL("serverInfo"), 0)),
Z_ARRVAL_P(zserver_info)
);
}
zend_update_property_long(swoole_mysql_coro_ce, ZEND_THIS, ZEND_STRL("sock"), mc->get_fd());
zend_update_property_bool(swoole_mysql_coro_ce, ZEND_THIS, ZEND_STRL("connected"), 1);
RETURN_TRUE;
}
static PHP_METHOD(swoole_mysql_coro, getDefer)
{
mysql_client *mc = swoole_get_mysql_client(ZEND_THIS);
RETURN_BOOL(mc->get_defer());
}
static PHP_METHOD(swoole_mysql_coro, setDefer)
{
mysql_client *mc = swoole_get_mysql_client(ZEND_THIS);
zend_bool defer = 1;
ZEND_PARSE_PARAMETERS_START(0, 1)
Z_PARAM_OPTIONAL
Z_PARAM_BOOL(defer)
ZEND_PARSE_PARAMETERS_END_EX(RETURN_FALSE);
bool ret = mc->set_defer(defer);
if (UNEXPECTED(!ret))
{
zend_throw_exception_ex(swoole_mysql_coro_exception_ce, mc->get_error_code(), "%s", mc->get_error_msg());
}
RETURN_BOOL(ret);
}
static PHP_METHOD(swoole_mysql_coro, query)
{
mysql_client *mc = swoole_get_mysql_client(ZEND_THIS);
char *sql;
size_t sql_length;
double timeout = 0;
ZEND_PARSE_PARAMETERS_START(1, 2)
Z_PARAM_STRING(sql, sql_length)
Z_PARAM_OPTIONAL
Z_PARAM_DOUBLE(timeout)
ZEND_PARSE_PARAMETERS_END_EX(RETURN_FALSE);
mc->add_timeout_controller(timeout, SW_TIMEOUT_RDWR);
mc->query(return_value, sql, sql_length);
mc->del_timeout_controller();
swoole_mysql_coro_sync_query_result_properties(ZEND_THIS, mc, return_value);
}
static PHP_METHOD(swoole_mysql_coro, fetch)
{
mysql_client *mc = swoole_get_mysql_client(ZEND_THIS);
double timeout = 0;
ZEND_PARSE_PARAMETERS_START(0, 1)
Z_PARAM_OPTIONAL
Z_PARAM_DOUBLE(timeout)
ZEND_PARSE_PARAMETERS_END_EX(RETURN_FALSE);
mc->add_timeout_controller(timeout, SW_TIMEOUT_RDWR);
mc->fetch(return_value);
mc->del_timeout_controller();
if (sw_unlikely(Z_TYPE_P(return_value) == IS_FALSE))
{
swoole_mysql_coro_sync_error_properties(ZEND_THIS, mc->get_error_code(), mc->get_error_msg(), mc->is_connect());
}
}
static PHP_METHOD(swoole_mysql_coro, fetchAll)
{
mysql_client *mc = swoole_get_mysql_client(ZEND_THIS);
double timeout = 0;
ZEND_PARSE_PARAMETERS_START(0, 1)
Z_PARAM_OPTIONAL
Z_PARAM_DOUBLE(timeout)
ZEND_PARSE_PARAMETERS_END_EX(RETURN_FALSE);
mc->add_timeout_controller(timeout, SW_TIMEOUT_RDWR);
mc->fetch_all(return_value);
mc->del_timeout_controller();
if (sw_unlikely(Z_TYPE_P(return_value) == IS_FALSE))
{
swoole_mysql_coro_sync_error_properties(ZEND_THIS, mc->get_error_code(), mc->get_error_msg(), mc->is_connect());
}
}
static PHP_METHOD(swoole_mysql_coro, nextResult)
{
mysql_client *mc = swoole_get_mysql_client(ZEND_THIS);
double timeout = 0;
ZEND_PARSE_PARAMETERS_START(0, 1)
Z_PARAM_OPTIONAL
Z_PARAM_DOUBLE(timeout)
ZEND_PARSE_PARAMETERS_END_EX(RETURN_FALSE);
mc->add_timeout_controller(timeout, SW_TIMEOUT_RDWR);
mc->next_result(return_value);
mc->del_timeout_controller();
swoole_mysql_coro_sync_query_result_properties(ZEND_THIS, mc, return_value);
if (Z_TYPE_P(return_value) == IS_TRUE)
{
if (mc->state == SW_MYSQL_STATE_IDLE)
{
// the end of procedure
Z_TYPE_INFO_P(return_value) = mc->get_fetch_mode() ? IS_FALSE : IS_NULL;
}
}
}
static PHP_METHOD(swoole_mysql_coro, prepare)
{
mysql_client *mc = swoole_get_mysql_client(ZEND_THIS);
char *statement;
size_t statement_length;
double timeout = 0;
ZEND_PARSE_PARAMETERS_START(1, 2)
Z_PARAM_STRING(statement, statement_length)
Z_PARAM_OPTIONAL
Z_PARAM_DOUBLE(timeout)
ZEND_PARSE_PARAMETERS_END_EX(RETURN_FALSE);
mc->add_timeout_controller(timeout, SW_TIMEOUT_RDWR);
if (UNEXPECTED(!mc->send_prepare_request(statement, statement_length)))
{
_failed:
swoole_mysql_coro_sync_error_properties(ZEND_THIS, mc->get_error_code(), mc->get_error_msg(), mc->is_connect());
RETVAL_FALSE;
}
else if (UNEXPECTED(mc->get_defer()))
{
RETVAL_TRUE;
}
else
{
mysql_statement* statement = mc->recv_prepare_response();
if (UNEXPECTED(!statement))
{
goto _failed;
}
RETVAL_OBJ(swoole_mysql_coro_statement_create_object(statement, Z_OBJ_P(ZEND_THIS)));
}
mc->del_timeout_controller();
}
static PHP_METHOD(swoole_mysql_coro, recv)
{
mysql_client *mc = swoole_get_mysql_client(ZEND_THIS);
double timeout = 0;
ZEND_PARSE_PARAMETERS_START(0, 1)
Z_PARAM_OPTIONAL
Z_PARAM_DOUBLE(timeout)
ZEND_PARSE_PARAMETERS_END_EX(RETURN_FALSE);
if (UNEXPECTED(!mc->check_connection()))
{
swoole_mysql_coro_sync_error_properties(ZEND_THIS, mc->get_error_code(), mc->get_error_msg(), false);
RETURN_FALSE;
}
mc->add_timeout_controller(timeout, SW_TIMEOUT_READ);
switch(mc->state)
{
case SW_MYSQL_STATE_IDLE:
swoole_mysql_coro_sync_error_properties(ZEND_THIS, ENOMSG, "no message to receive");
RETVAL_FALSE;
break;
case SW_MYSQL_STATE_QUERY:
mc->recv_query_response(return_value);
break;
case SW_MYSQL_STATE_PREPARE:
{
mysql_statement *statement = mc->recv_prepare_response();
if (UNEXPECTED(!statement))
{
RETVAL_FALSE;
}
else
{
RETVAL_OBJ(swoole_mysql_coro_statement_create_object(statement, Z_OBJ_P(ZEND_THIS)));
}
break;
}
default:
if (UNEXPECTED(mc->state & SW_MYSQL_COMMAND_FLAG_EXECUTE))
{
swoole_mysql_coro_sync_error_properties(ZEND_THIS, EPERM, "please use statement to receive data");
}
else
{
swoole_mysql_coro_sync_error_properties(ZEND_THIS, EPERM, "please use fetch/fetchAll/nextResult to get result");
}
RETVAL_FALSE;
}
mc->del_timeout_controller();
}
static void swoole_mysql_coro_query_transcation(INTERNAL_FUNCTION_PARAMETERS, const char* command, size_t command_length)
{
mysql_client *mc = swoole_get_mysql_client(ZEND_THIS);
double timeout = 0;
ZEND_PARSE_PARAMETERS_START(0, 1)
Z_PARAM_OPTIONAL
Z_PARAM_DOUBLE(timeout)
ZEND_PARSE_PARAMETERS_END_EX(RETURN_FALSE);
if (UNEXPECTED(mc->get_defer()))
{
zend_throw_exception_ex(
swoole_mysql_coro_exception_ce, EPERM,
"you should not query transaction when defer mode is on, if you want, please use `query('%s')` instead", command
);
RETURN_FALSE;
}
mc->add_timeout_controller(timeout, SW_TIMEOUT_RDWR);
mc->query(return_value, command, command_length);
mc->del_timeout_controller();
}
static PHP_METHOD(swoole_mysql_coro, begin)
{
swoole_mysql_coro_query_transcation(INTERNAL_FUNCTION_PARAM_PASSTHRU, ZEND_STRL("BEGIN"));
}
static PHP_METHOD(swoole_mysql_coro, commit)
{
swoole_mysql_coro_query_transcation(INTERNAL_FUNCTION_PARAM_PASSTHRU, ZEND_STRL("COMMIT"));
}
static PHP_METHOD(swoole_mysql_coro, rollback)
{
swoole_mysql_coro_query_transcation(INTERNAL_FUNCTION_PARAM_PASSTHRU, ZEND_STRL("ROLLBACK"));
}
#ifdef SW_USE_MYSQLND
static PHP_METHOD(swoole_mysql_coro, escape)
{
mysql_client *mc = swoole_get_mysql_client(ZEND_THIS);
char *str;
size_t str_length;
zend_long flags = 0;
ZEND_PARSE_PARAMETERS_START(1, 2)
Z_PARAM_STRING(str, str_length)
Z_PARAM_OPTIONAL
Z_PARAM_LONG(flags)
ZEND_PARSE_PARAMETERS_END_EX(RETURN_FALSE);
char *newstr = (char *) safe_emalloc(2, str_length + 1, 1);
if (!newstr)
{
php_swoole_fatal_error(E_ERROR, "emalloc(%ld) failed", str_length + 1);
RETURN_FALSE;
}
const MYSQLND_CHARSET* cset = mysqlnd_find_charset_nr(mc->charset);
if (!cset)
{
php_swoole_fatal_error(E_ERROR, "unknown mysql charset[%d]", mc->charset);
RETURN_FALSE;
}
zend_ulong newstr_len = mysqlnd_cset_escape_slashes(cset, newstr, str, str_length);
if (newstr_len == (zend_ulong) ~0)
{
php_swoole_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)
{
mysql_client *mc = swoole_get_mysql_client(ZEND_THIS);
mc->close();
zend_update_property_bool(swoole_mysql_coro_ce, ZEND_THIS, ZEND_STRL("connected"), 0);
RETURN_TRUE;
}
static PHP_METHOD(swoole_mysql_coro_statement, execute)
{
mysql_statement *ms = swoole_get_mysql_statement(ZEND_THIS);
zval *params = nullptr;
double timeout = 0;
ZEND_PARSE_PARAMETERS_START(0, 2)
Z_PARAM_OPTIONAL
Z_PARAM_ARRAY_EX(params, 1, 0)
Z_PARAM_DOUBLE(timeout)
ZEND_PARSE_PARAMETERS_END_EX(RETURN_FALSE);
ms->add_timeout_controller(timeout, SW_TIMEOUT_RDWR);
ms->execute(return_value, params);
ms->del_timeout_controller();
swoole_mysql_coro_sync_execute_result_properties(ZEND_THIS, return_value);
}
static PHP_METHOD(swoole_mysql_coro_statement, fetch)
{
mysql_statement *ms = swoole_get_mysql_statement(ZEND_THIS);
double timeout = 0;
ZEND_PARSE_PARAMETERS_START(0, 1)
Z_PARAM_OPTIONAL
Z_PARAM_DOUBLE(timeout)
ZEND_PARSE_PARAMETERS_END_EX(RETURN_FALSE);
ms->add_timeout_controller(timeout, SW_TIMEOUT_RDWR);
ms->fetch(return_value);
ms->del_timeout_controller();
if (sw_unlikely(Z_TYPE_P(return_value) == IS_FALSE))
{
swoole_mysql_coro_sync_execute_error_properties(ZEND_THIS, ms->get_error_code(), ms->get_error_msg());
}
}
static PHP_METHOD(swoole_mysql_coro_statement, fetchAll)
{
mysql_statement *ms = swoole_get_mysql_statement(ZEND_THIS);
double timeout = 0;
ZEND_PARSE_PARAMETERS_START(0, 1)
Z_PARAM_OPTIONAL
Z_PARAM_DOUBLE(timeout)
ZEND_PARSE_PARAMETERS_END_EX(RETURN_FALSE);
ms->add_timeout_controller(timeout, SW_TIMEOUT_RDWR);
ms->fetch_all(return_value);
ms->del_timeout_controller();
if (sw_unlikely(Z_TYPE_P(return_value) == IS_FALSE))
{
swoole_mysql_coro_sync_execute_error_properties(ZEND_THIS, ms->get_error_code(), ms->get_error_msg());
}
}
static PHP_METHOD(swoole_mysql_coro_statement, nextResult)
{
mysql_statement *ms = swoole_get_mysql_statement(ZEND_THIS);
double timeout = 0;
ZEND_PARSE_PARAMETERS_START(0, 1)
Z_PARAM_OPTIONAL
Z_PARAM_DOUBLE(timeout)
ZEND_PARSE_PARAMETERS_END_EX(RETURN_FALSE);
ms->add_timeout_controller(timeout, SW_TIMEOUT_RDWR);
ms->next_result(return_value);
ms->del_timeout_controller();
swoole_mysql_coro_sync_execute_result_properties(ZEND_THIS, return_value);
if (Z_TYPE_P(return_value) == IS_TRUE)
{
mysql_client *mc = ms->get_client();
if (mc->state == SW_MYSQL_STATE_IDLE)
{
// the end of procedure
Z_TYPE_INFO_P(return_value) = mc->get_fetch_mode() ? IS_FALSE : IS_NULL;
}
}
}
static PHP_METHOD(swoole_mysql_coro_statement, recv)
{
mysql_statement *ms = swoole_get_mysql_statement(ZEND_THIS);
double timeout = 0;
enum sw_mysql_state state;
ZEND_PARSE_PARAMETERS_START(0, 1)
Z_PARAM_OPTIONAL
Z_PARAM_DOUBLE(timeout)
ZEND_PARSE_PARAMETERS_END_EX(RETURN_FALSE);
if (UNEXPECTED(!ms->is_available()))
{
swoole_mysql_coro_sync_execute_error_properties(ZEND_THIS, ms->get_error_code(), ms->get_error_msg(), false);
RETURN_FALSE;
}
ms->add_timeout_controller(timeout, SW_TIMEOUT_READ);
switch((state = ms->get_client()->state))
{
case SW_MYSQL_STATE_IDLE:
swoole_mysql_coro_sync_execute_error_properties(ZEND_THIS, ENOMSG, "no message to receive");
RETVAL_FALSE;
break;
case SW_MYSQL_STATE_EXECUTE:
ms->recv_execute_response(return_value);
break;
default:
if (UNEXPECTED(state & SW_MYSQL_COMMAND_FLAG_QUERY))
{
swoole_mysql_coro_sync_execute_error_properties(ZEND_THIS, EPERM, "please use client to receive data");
}
else
{
swoole_mysql_coro_sync_execute_error_properties(ZEND_THIS, EPERM, "please use fetch/fetchAll/nextResult to get result");
}
RETVAL_FALSE;
}
ms->del_timeout_controller();
}
static PHP_METHOD(swoole_mysql_coro_statement, close)
{
mysql_statement *ms = swoole_get_mysql_statement(ZEND_THIS);
ms->close();
RETURN_TRUE;
}
C
1
https://gitee.com/yihaibin/swoole.git
git@gitee.com:yihaibin/swoole.git
yihaibin
swoole
swoole-src
master

搜索帮助