GithubStar/workflow
1
0
Fork 0
mirror of https://github.com/sogou/workflow.git synced 2026-10-26 01:30:47 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' of https://github.com/sogou/workflow

Browse Source
This commit is contained in:
XieHan
2020-10-01 22:41:22 +08:00
parent a8f3b09c00 bdae28dbf3
commit 3ad0053537
2 changed files with 730 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/en/about-upstream.md
+410
View File
@@ -0,0 +1,410 @@
# About Upstream
In nginx, Upstream represents the load balancing configuration of the reverse proxy. Here, we expand the meaning of Upstream so that it has the following characteristics:
1. Each Upstream is an independent reverse proxy
2. Accessing an Upstream is equivalent to using an appropriate strategy to select one in a group of services/targets/upstream and downstream for access
3. Upstream has load balancing, error handling and service governance capabilities.
### Advantages of Upstream over domain name DNS resolution
Both upstream and domain name DNS resolution can configure a group of ip to a host, but
1. DNS domain name resolution doesn’t address port number. The service DNS domain names with the same IP and different ports cannot be configured together; but it is possible for Upstream.
2. The set of addresses corresponding to DNS domain name resolution must be ip; while the set of addresses corresponding to Upstream can be ip, domain name or unix-domain-socket
3. Normally, DNS domain name resolution will be cached by operating system or DNS server on the network, and the update time is limited by ttl; while Upstream can be updated in real time and take effect in real time
4. The consumption of DNS domain name is much greater than that of Upstream resolution and selection
### Upstream of Workflow
This is a local reverse proxy module, and the proxy configuration is effective for both server and client.
Proxy supports dynamic configuration. Proxy name does not include port, but proxy request supports specified port.
Each Upstream is configured with its own independent name UpstreamName, and a set of Addresses is added and set. These Addresses can be:
1. ip4
2. ip6
3. Domain name
4. unix-domain-socket
### Why to replace nginx's Upstream
#### Upstream working mode of nginx
1. Supports http/https protocol only
2. Needs to build a nginx service, start the start process occupies socket and other resources
3. The request is sent to nginx first, and nginx forwards the request to remote end, which will increase one more network communication overhead
#### Local Upstream working method of workflow
1. Protocol irrelevant, you can even access mysql, redis, mongodb, etc. through upstream
2. You can directly simulate the function of reverse proxy in the process, no need to start other processes or ports
3. The selection process is basic calculation and table lookup, no additional network communication overhead
# Use Upstream
### Common interfaces
~~~cpp
class UpstreamManager
{
public:
static int upstream_create_consistent_hash(const std::string& name,
upstream_route_t consitent_hash);
static int upstream_create_weighted_random(const std::string& name,
bool try_another);
static int upstream_create_manual(const std::string& name,
upstream_route_t select,
bool try_another,
upstream_route_t consitent_hash);
static int upstream_delete(const std::string& name);
public:
static int upstream_add_server(const std::string& name,
const std::string& address);
static int upstream_add_server(const std::string& name,
const std::string& address,
const struct AddressParams *address_params);
static int upstream_remove_server(const std::string& name,
const std::string& address);
...
}
~~~
### Example 1 Random access in multiple targets
Configure a local reverse proxy to evenly send all the local requests for **my_proxy.name** to 6 target servers
~~~cpp
UpstreamManager::upstream_create_weighted_random(
"my_proxy.name",
true); // In case of fusing, retry till the available is found or all fuses are blown
UpstreamManager::upstream_add_server("my_proxy.name", "192.168.2.100:8081");
UpstreamManager::upstream_add_server("my_proxy.name", "192.168.2.100:8082");
UpstreamManager::upstream_add_server("my_proxy.name", "192.168.10.10");
UpstreamManager::upstream_add_server("my_proxy.name", "test.sogou.com:8080");
UpstreamManager::upstream_add_server("my_proxy.name", "abc.sogou.com");
UpstreamManager::upstream_add_server("my_proxy.name", "abc.sogou.com");
UpstreamManager::upstream_add_server("my_proxy.name", "/dev/unix_domain_scoket_sample");
auto *http_task = WFTaskFactory::create_http_task("http://my_proxy.name/somepath?a=10", 0, 0, nullptr);
http_task->start();
~~~
Basic principles
1. Select a target randomly
2. If try_another is configured as true, one of all surviving targets will be selected randomly
3. Select in the main servers only, the mains and backups of the group where the selected target is located and the backup without group are regarded as valid optional objects
### Example 2 Random access among multiple targets based on weights
Configure a local reverse proxy, send all **weighted.random** requests to the 3 target servers based on the weight distribution of 5/20/1
~~~cpp
UpstreamManager::upstream_create_weighted_random(
"weighted.random",
false); // If you don’t retry in case of fusing, the request will surely fail
AddressParams address_params = ADDRESS_PARAMS_DEFAULT;
address_params.weight = 5; //weight is 5
UpstreamManager::upstream_add_server("weighted.random", "192.168.2.100:8081", &address_params); // weight is 5
address_params.weight = 20; // weight is 20
UpstreamManager::upstream_add_server("weighted.random", "192.168.2.100:8082", &address_params); // weight is 20
UpstreamManager::upstream_add_server("weighted.random", "abc.sogou.com"); // weight is 1
auto *http_task = WFTaskFactory::create_http_task("http://weighted.random:9090", 0, 0, nullptr);
http_task->start();
~~~
Basic principles
1. According to the weight distribution, randomly select a target, the greater the weight is, the greater the probability is
2. If try_another is configured as true, one of all surviving targets will be selected randomly as per weights.
3. Select in the main servers only, the main and backup of the group where the selected target is located and the backup without group are regarded as valid optional objects
### Example 3 Access among multiple targets based on the framework's default consistent hash
~~~cpp
UpstreamManager::upstream_create_consistent_hash(
"abc.local",
nullptr); // nullptr represents using the default consistent hash function of the framework
UpstreamManager::upstream_add_server("abc.local", "192.168.2.100:8081");
UpstreamManager::upstream_add_server("abc.local", "192.168.2.100:8082");
UpstreamManager::upstream_add_server("abc.local", "192.168.10.10");
UpstreamManager::upstream_add_server("abc.local", "test.sogou.com:8080");
UpstreamManager::upstream_add_server("abc.local", "abc.sogou.com");
auto *http_task = WFTaskFactory::create_http_task("http://abc.local/service/method", 0, 0, nullptr);
http_task->start();
~~~
Basic principles
1. Each main server is regarded as 16 virtual nodes
2. The framework will use std::hash to calculate the address + virtual index of all nodes as the node value of the consistent hash
3. The framework will use std::hash to calculate path + query + fragment as a consistent hash data value
4. Choose the value nearest to the surviving node as the target each time
5. For each main, as long as there is a main in surviving group, or there is a backup in surviving group, or there is a surviving no group backup, it is regarded as surviving
### Example 4 User-defined consistent hash function
~~~cpp
UpstreamManager::upstream_create_consistent_hash(
"abc.local",
[](const char *path, const char *query, const char *fragment) -> unsigned int {
unsigned int hash = 0;
while (*path)
hash = (hash * 131) + (*path++);
while (*query)
hash = (hash * 131) + (*query++);
while (*fragment)
hash = (hash * 131) + (*fragment++);
return hash;
});
UpstreamManager::upstream_add_server("abc.local", "192.168.2.100:8081");
UpstreamManager::upstream_add_server("abc.local", "192.168.2.100:8082");
UpstreamManager::upstream_add_server("abc.local", "192.168.10.10");
UpstreamManager::upstream_add_server("abc.local", "test.sogou.com:8080");
UpstreamManager::upstream_add_server("abc.local", "abc.sogou.com");
auto *http_task = WFTaskFactory::create_http_task("http://abc.local/sompath?a=1#flag100", 0, 0, nullptr);
http_task->start();
~~~
Basic principles
1. The framework will use a user-defined consistent hash function as the data value
2. The rest is the same as the above principles
### Example 5 User-defined selection strategy
~~~cpp
UpstreamManager::upstream_create_manual(
"xyz.cdn",
[](const char *path, const char *query, const char *fragment) -> unsigned int {
return atoi(fragment);
},
true, // If a blown target is selected, a second selection will be made
nullptr); // nullptr represents using the default consistent hash function of the framework in the second selection
UpstreamManager::upstream_add_server("xyz.cdn", "192.168.2.100:8081");
UpstreamManager::upstream_add_server("xyz.cdn", "192.168.2.100:8082");
UpstreamManager::upstream_add_server("xyz.cdn", "192.168.10.10");
UpstreamManager::upstream_add_server("xyz.cdn", "test.sogou.com:8080");
UpstreamManager::upstream_add_server("xyz.cdn", "abc.sogou.com");
auto *http_task = WFTaskFactory::create_http_task("http://xyz.cdn/sompath?key=somename#3", 0, 0, nullptr);
http_task->start();
~~~
Basic principles
1. The framework first determines the selection in the main server list according to the normal selection function provided by the user and then get the modulo
2. For each main server, as long as there is a main server in surviving group, or there is a backup in surviving group, or there is a surviving no group backup, it is regarded as surviving
3. If the selected target no longer survives and try_another is set as true, a second selection will be made using consistent hash function
4. If the second selection is triggered, the consistent hash will ensure that a survival target will be selected, unless all machines are blown
### Example 6 Simple main-backup mode
~~~cpp
UpstreamManager::upstream_create_weighted_random(
"simple.name",
true);//One main, one backup, nothing is different in this item
AddressParams address_params = ADDRESS_PARAMS_DEFAULT;
address_params.server_type = 0; /* 1 for main server */
UpstreamManager::upstream_add_server("simple.name", "main01.test.ted.bj.sogou", &address_params); // main
address_params.server_type = 1; /* 0 for backup server */
UpstreamManager::upstream_add_server("simple.name", "backup01.test.ted.gd.sogou", &address_params); //backup
auto *http_task = WFTaskFactory::create_http_task("http://simple.name/request", 0, 0, nullptr);
auto *redis_task = WFTaskFactory::create_redis_task("redis://simple.name/2", 0, nullptr);
redis_task->get_req()->set_query("MGET", {"key1", "key2", "key3", "key4"});
(*http_task * redis_task).start();
~~~
Basic principles
1. The main-backup mode does not conflict with any of the modes shown above, and it can take effect at the same time
2. The number of main/backup is independent of each other and there is no limit. All main servers are coequal to each other, and all backup servers are coequal to each others, but main and backup are not coequal to each other.
3. As long as a main server is alive, the request will always use a main server.
4. If all main servers are blown, backup server will take over the request as a substitute target until any main server works well again
5. In every strategy, surviving backup can be used as the basis for the survival of main
### Example 7 Main-backup + consistent hash + grouping
~~~cpp
UpstreamManager::upstream_create_consistent_hash(
"abc.local",
nullptr);//nullptr represents using the default consistent hash function of the framework
AddressParams address_params = ADDRESS_PARAMS_DEFAULT;
address_params.server_type = 0;
address_params.group_id = 1001;
UpstreamManager::upstream_add_server("abc.local", "192.168.2.100:8081", &address_params);//main in group 1001
address_params.server_type = 1;
address_params.group_id = 1001;
UpstreamManager::upstream_add_server("abc.local", "192.168.2.100:8082", &address_params);//backup for group 1001
address_params.server_type = 0;
address_params.group_id = 1002;
UpstreamManager::upstream_add_server("abc.local", "backup01.test.ted.bj.sogou", &address_params);//main in group 1002
address_params.server_type = 1;
address_params.group_id = 1002;
UpstreamManager::upstream_add_server("abc.local", "backup01.test.ted.gd.sogou", &address_params);//backup for group 1002
address_params.server_type = 1;
address_params.group_id = -1;
UpstreamManager::upstream_add_server("abc.local", "test.sogou.com:8080", &address_params);//backup with no group mean backup for all groups and no group
UpstreamManager::upstream_add_server("abc.local", "abc.sogou.com");//main, no group
auto *http_task = WFTaskFactory::create_http_task("http://abc.local/service/method", 0, 0, nullptr);
http_task->start();
~~~
Basic principles
1. Group number -1 means no group, this kind of target does not belong to any group
2. The main servers without a group are coequal to each other, and they can even be regarded as one group. But they are isolated from the other main servers with a group
3. A backup without a group can serve as a backup for any group target of Global/any target without a group
4. The group number can identify which main and backup are working together
5. The backups of different groups are isolated from each other, and they serve the main servers of their own group only
6. Add the default group number -1 of the target, and the type is main
# Upstream selection strategy
When the URIHost of the url that initiates the request is filled with UpstreamName, it is regarded as a request to the Upstream corresponding to the name, and then it will be selected from the set of Addresses recorded by the Upstream:
1. Weight random strategy: selection randomly according to weight
2. Consistent hashing strategy: The framework uses a standard consistent hashing algorithm, and users can define the consistent hash function consistent_hash for the requested uri
3. Manual strategy: make definite selection according to the select function that user provided for the requested uri, if the blown target is selected: **a.** If try_another is false, this request will return to failure **b.** If try_another is true, the framework uses standard consistent hash algorithm to make a second selection, and the user can define the consistent hash function consistent_hash for the requested uri
4. Main-backup strategy: According to the priority of main first, backup next, select a main server as long as it can be used. This strategy can take effect concurrently with any of [1], [2], and [3], and they influence each other.
Round-robin/weighted-round-robin: regarded as equivalent to [1], not available for now
The framework recommends common users to use strategy [2], which can ensure that the cluster has good fault tolerance and scalability
For complex scenarios, advanced users can use strategy [3] to customize complex selection logic
# Address attribute
~~~cpp
struct EndpointParams
{
size_t max_connections;
int connect_timeout;
int response_timeout;
int ssl_connect_timeout;
};
static constexpr struct EndpointParams ENDPOINT_PARAMS_DEFAULT =
{
.max_connections = 200,
.connect_timeout = 10 * 1000,
.response_timeout = 10 * 1000,
.ssl_connect_timeout = 10 * 1000,
};
struct AddressParams
{
struct EndpointParams endpoint_params;
unsigned int dns_ttl_default;
unsigned int dns_ttl_min;
unsigned int max_fails;
unsigned short weight;
int server_type; /* 0 for main and 1 for backup. */
int group_id;
};
static constexpr struct AddressParams ADDRESS_PARAMS_DEFAULT =
{
.endpoint_params = ENDPOINT_PARAMS_DEFAULT,
.dns_ttl_default = 12 * 3600,
.dns_ttl_min = 180,
.max_fails = 200,
.weight = 1, // only for main of UPSTREAM_WEIGHTED_RANDOM
.server_type = 0,
.group_id = -1,
};
~~~
Each address can be configured with custom parameters:
* Max_connections, connect_timeout, response_timeout, ssl_connect_timeout of EndpointParams: connection-related parameters
* dns_ttl_default: The default ttl in the dns cache in seconds, and the default value is 12 hours. The dns cache is for the current process, that is, the process will disappear after exiting, and the configuration is only valid for the current process
* dns_ttl_min: The shortest effective time of dns in seconds, and the default value is 3 minutes. It is used to decide whether to perform dns again when communication fails and retry.
* max_fails: the number of [continuous] failures that triggered fusing (Note: each time the communication is successful, the count will be cleared)
* Weight: weight, the default value is 1, which is only valid for main. It is used for Upstream random strategy selection, the larger the weight is, the easier it is to be selected; this parameter is meaningless under other strategies
* server_type: main/backup configuration, main by default (server_type=0). At any time, the main servers in the same group are always at higher priority than backups
* group_id: basis for grouping, the default value is -1. -1 means no grouping (free). A free backup can be regarded as backup to any main server. Any backup with group is always at higher priority than any free backup.
# About fuse
## MTTR
Mean time to repair (MTTR) is the average value of the repair time when the product changes from a fault state to a working state.
## Service avalanche effect
Service avalanche effect is a phenomenon in which "service caller failure" (result) is caused by "service provider's failure" (cause), and the unavailability is amplified gradually/level by level
If it is not controlled effectively, the effect will not converge, but will be amplified geometrically, just like an avalanche, that’s why it is called avalanche effect
Description of the phenomenon: at first it is just a small service or module abnormality/timeout, causing abnormality/timeout of other downstream dependent services, then causing a chain reaction, eventually leading to paralysis of most or all services
As the fault is repaired, the effect will disappear, so the duration of the effect is usually equal to MTTR
## Fuse mechanism
When the error or abnormal touch of a certain target meets the preset threshold condition, the target is temporarily considered unavailable, and the target is removed, namely fuse is started and enters the fuse period
After the fuse duration reaches MTTR duration, the fuse is closed, (attempt to) restore the target
Fuse mechanism strategy can effectively prevent avalanche effect
## Upstream fuse protection mechanism
MTTR=30 seconds, which is temporarily not configurable, but we will consider opening it to be configured by users in the future.
When the number of consecutive failures of a certain Address reaches the set upper limit (200 times by default), this Address will be blown, MTTR=30 seconds.
During the fusing period, once the Address is selected by the strategy, Upstream will decide whether to try other Addresses and how to try according to the specific configuration
Please note that if one of the following 1-4 scenarios is met, the communication task will get an error of WFT_ERR_UPSTREAM_UNAVAILABLE = 1004:
1. Weight random strategy, all targets are in the fusing period
2. Consistent hash strategy, all targets are in the fusing period
3. Manual strategy and try_another==true, all targets are in the fusing period
4. Manual strategy and try_another==false, and all the following three conditions shall meet at the same time:
1). The main selected by the select function is in the fusing period, and all free devices are in the fusing period
2). The main is a free main, or other targets in the group where the main is located are all in the fusing period
3). All free devices are in the fusing period
# Upstream port priority
1. Priority is given to the port number explicitly configured on the Upstream Address
2. If not, select the port number explicitly configured in the request url
3. If none, use the default port number of the protocol
~~~cpp
Configure UpstreamManager::upstream_add_server("my_proxy.name", "192.168.2.100:8081");
Request http://my_proxy.name:456/test.html => http://192.168.2.100:8081/test.html
Request http://my_proxy.name/test.html => http://192.168.2.100:8081/test.html
~~~
~~~cpp
Configure UpstreamManager::upstream_add_server("my_proxy.name", "192.168.10.10");
Request http://my_proxy.name:456/test.html => http://192.168.10.10:456/test.html
Request http://my_proxy.name/test.html => http://192.168.10.10:80/test.html
~~~
docs/en/tutorial-12-mysql_cli.md
+320
View File
@@ -0,0 +1,320 @@
# Asynchronous MySQL client: mysql\_cli
# Sample code
[tutorial-12-mysql\_cli.cc](../tutorial/tutorial-12-mysql_cli.cc)
# About mysql\_cli
The usage of mysql\_cli in the tutorial is similar to that of the official client. It is an asynchronous MySQL client with an interactive command line interface.
To start the program, run the command: ./mysql_cli \<URL\>
After startup, you can directly enter MySQL command in the terminal to interact with db, or enter `quit` or `Ctrl-C` to exit.
# Format of MySQL URL
mysql://username:password@host:port/dbname?character\_set=charset&character\_set\_results=charset
- fill in the username and the password for the MySQL database;
- the default port number is 3306;
- **dbname** is the name of the database to be used. It is recommended to provide a dbname if SQL statements only operates on one database;
- If you have upstream selection requirements for MySQL, please see [upstream documents](/docs/en/about-upstream.md).
- **character_set** indicates a character set used for the client, with the same meaning of --default-character-set in official client. The default value is utf8. For details, please see official MySQL documents [character-set.html](https://dev.mysql.com/doc/internals/en/character-set.html).
- **character_set_results** indicates a character set for client, connection and results. If you wants to use `SET NAMES ` in SQL statements, please set it here.
Sample MySQL URL:
mysql://root:password@127.0.0.1
mysql://@test.mysql.com:3306/db1?character\_set=utf8
# Creating and starting a MySQL task
You can use WFTaskFactory to create a MySQL task. The usage of creating interface and callback functions are similar to other tasks in workflow:
~~~cpp
using mysql_callback_t = std::function<void (WFMySQLTask *)>;
WFMySQLTask *create_mysql_task(const std::string& url, int retry_max, mysql_callback_t callback);
void set_query(const std::string& query);
~~~
You can call **set\_query()** on the request to write SQL statements after creating a WFMySQLTask.
According to the MySQL protocol, if an empty packet is sent after the connection is established, the server will wait instead of returning a packet, so the user will get a timeout. Therefore, the framework makes special check on those tasks that do not call `set_query()` and will immediately return **WFT\_ERR\_MYSQL\_QUERY\_NOT\_SET**.
Other functions, including callback, series and user\_data are used in a way similar to other tasks in workflow.
The following codes show some general usage:
~~~cpp
int main(int argc, char *argv[])
{
...
WFMySQLTask *task = WFTaskFactory::create_mysql_task(url, RETRY_MAX, mysql_callback);
task->get_req()->set_query("SHOW TABLES;");
...
task->start();
...
}
~~~
# Supported commands
Currently the supported command is **COM\_QUERY**, which can cover the basic requirements for adding, deleting, modifying and querying data, creating and deleting databases, creating and deleting tables, prepare, using stored procedures and using transactions.
Because the program doesn't support the selection of databases (**USE** command) in our interactive commands, if there are **cross-database** operations in SQL statements, you can specify the database and table with **db\_name.table\_name**.
**Multiple commands** can be joined together and then passed to WFMySQLTask with `set_query()`. Generally speaking, multiple statements can get all the results back at one time. However, as the packet return method in the MySQL protocol is not compatible with question and answer communication under some provisions, please read the following cautions before you add the SQL statements in `set_query()`:
- For the statement that gets a single result set, multiple statements can be spliced (ordinary INSERT/UPDATE/SELECT/PREPARE)
- One statement that returns multiple result sets (such as CALL a stored procedure) is also supported.
- In other cases, it is recommended to divide SQL statements into individual requests
For example:
~~~cpp
// Correct!
// This will get multiple result sets for mutiple ordinary statements.
req->set_query("SELECT * FROM table1; SELECT * FROM table2; INSERT INTO table3 (id) VALUES (1);");
// Correct!
// This will get single or multiple result sets for some multi-result-set statements.
req->set_query("CALL procedure1();");
// Incorrect!
// This contains a multi-result-set statement and other statements.
// Cannot get all the result sets correctly.
req->set_query("CALL procedure1(); SELECT * FROM table1;");
~~~
# Parsing results
Similar to other tasks in workflow, you can use **task->get\_resp()** to get **MySQLResponse**, and you can use **MySQLResultCursor** to traverse the result set, the infomation of each column of the result set (**MySQLField**), each row and each **MySQLCell**. For details on the interfaces, please see [MySQLResult.h](/src/protocol/MySQLResult.h).
The specific steps should be:
1. checking the task state (state at communication): you can check whether the task is successfully executed by checking whether **task->get\_state()** is equal to WFT\_STATE\_SUCCESS;
2. determining the type of the reply packet (state at parsing the return packet): call **resp->get\_packet\_type()** to check the type of the MySQL return packet. The common types include:
- MYSQL\_PACKET\_OK: non-result-set requests: parsed successfully;
- MYSQL\_PACKET\_EOF: result-set requests: parsed successfully;
- MYSQL\_PACKET\_ERROR: requests: failed;
3. checking the result set state (state at reading the result sets): you can use MySQLResultCursor to read the content in the result set. Because the data returned by a MySQL server contains multiple result sets, the cursor will **automatically point to the reading position of the first result set** at first. **cursor->get\_cursor\_status()** returns the following states:
- MYSQL\_STATUS\_GET\_RESULT: the data are available to read;
- MYSQL\_STATUS\_END: the last record of the current result set has been read;
- MYSQL\_STATUS\_EOF: all result-sets are fetched;
- MYSQL\_STATUS\_OK: the reply packet is a non-result-set packet, so you do not need to read data through the result set interface;
- MYSQL\_STATUS\_ERROR: parsing error;
4. reading each field of the columns:
- `int get_field_count() const;`
- `const MySQLField *fetch_field();`
- `const MySQLField *const *fetch_fields() const;`
5. reading each line: you can use **cursor->fetch\_row()** to read by row until the return value is false, in which the offset within the cursor that points to the row in the current result set will be moved:
- `int get_rows_count() const;`
- `bool fetch_row(std::vector<MySQLCell>& row_arr);`
- `bool fetch_row(std::map<std::string, MySQLCell>& row_map);`
- `bool fetch_row(std::unordered_map<std::string, MySQLCell>& row_map);`
- `bool fetch_row_nocopy(const void **data, size_t *len, int *data_type);`
6. taking out all the rows in the current result set directly: you can use **cursor->fetch\_all()** to read all rows, and the cursor that is used to record the rows internally will be moved directly to the end; The cursor state changes to MYSQL\_STATUS\_END:
- `bool fetch_all(std::vector<std::vector<MySQLCell>>& rows);`
7. returning to the head of the current result set: if it is necessary to read this result set again, you can use **cursor->rewind()** to return to the head of the current result set, and then read it via the operations in Step 5 or Step 6;
8. getting the next result set: because the data packet returned by MySQL server may contains multiple result sets (for example, each select statement gets a result set; or the multiple result sets returned by calling a procedure). Therefore, you can use **cursor->next\_result\_set()** to jump to the next result set. If the return value is false, it means that all result sets have been taken.
9. returning to the first result set: use **cursor->first\_result\_set()** to return to the heads of all result sets, and then you can repeat the operations from Step 3.
10. getting the data of each column (MySQLCell): the row read in Step 5 is composed of multiple columns, and the result of each column is one MySQLCell. It mainly uses the following interfaces:
- `int get_data_type();` returns MYSQL\_TYPE\_LONG, MYSQL\_TYPE\_STRING, and etc. For the details, please see [mysql\_types.h](../src/protocol/mysql_types.h).
- `bool is_TYPE() const;` . The TYPE is int, string or ulonglong. It is used to check the data type.
- `TYPE as_TYPE() const;` Same as the above. It reads the data from MySQLCell in a certain type.
- `void get_cell_nocopy(const void **data, size_t *len, int *data_type) const;` nocopy interface.
The whole example is shown below:
~~~cpp
void task_callback(WFMySQLTask *task)
{
// step-1. Check the status of the task
if (task->get_state() != WFT_STATE_SUCCESS)
{
fprintf(stderr, "task error = %d\n", task->get_error());
return;
}
MySQLResultCursor cursor(task->get_resp());
bool test_first_result_set_flag = false;
bool test_rewind_flag = false;
begin:
// step-2. Check the status of reply packet
switch (resp->get_packet_type())
{
case MYSQL_PACKET_OK:
fprintf(stderr, "OK. %llu rows affected. %d warnings. insert_id=%llu.\n",
task->get_resp()->get_affected_rows(),
task->get_resp()->get_warnings(),
task->get_resp()->get_last_insert_id());
break;
case MYSQL_PACKET_EOF:
do {
fprintf(stderr, "cursor_status=%d field_count=%u rows_count=%u ",
cursor.get_cursor_status(), cursor.get_field_count(), cursor.get_rows_count());
// step-3. Check the status of the result set
if (cursor.get_cursor_status() != MYSQL_STATUS_GET_RESULT)
break;
// step-4. Read each fields. This is a nocopy api
const MySQLField *const *fields = cursor.fetch_fields();
for (int i = 0; i < cursor.get_field_count(); i++)
{
fprintf(stderr, "db=%s table=%s name[%s] type[%s]\n",
fields[i]->get_db().c_str(), fields[i]->get_table().c_str(),
fields[i]->get_name().c_str(), datatype2str(fields[i]->get_data_type()));
}
// step-6. Read all the rows. You may use while (cursor.fetch_row(map/vector)) to get each rows accoding to step-5.
std::vector<std::vector<MySQLCell>> rows;
cursor.fetch_all(rows);
for (unsigned int j = 0; j < rows.size(); j++)
{
// step-10. Read each cell
for (unsigned int i = 0; i < rows[j].size(); i++)
{
fprintf(stderr, "[%s][%s]", fields[i]->get_name().c_str(),
datatype2str(rows[j][i].get_data_type()));
// step-10. Check the type wih is_string()and transform the type with as_string()
if (rows[j][i].is_string())
{
std::string res = rows[j][i].as_string();
fprintf(stderr, "[%s]\n", res.c_str());
} else if (rows[j][i].is_int()) {
fprintf(stderr, "[%d]\n", rows[j][i].as_int());
} // else if ...
}
}
// step-8. Get the next result set
} while (cursor.next_result_set());
if (test_first_result_set_flag == false)
{
test_first_result_set_flag = true;
// step-9. Go back to the first result set
cursor.first_result_set();
goto begin;
}
if (test_rewind_flag == false)
{
test_rewind_flag = true;
// step-7. Go back to the first of the current result set
cursor.rewind();
goto begin;
}
break;
default:
fprintf(stderr, "Abnormal packet_type=%d\n", resp->get_packet_type());
break;
}
return;
}
~~~
# WFMySQLConnection
Since it is a highly concurrent asynchronous client, this means that the client may have more than one connection to the server. As both MySQL transactions and preparation are stateful, in order to ensure that one transaction or preparation ocupies one connection exclusively, you can use our encapsulated secondary factory WFMySQLConnection to create a task. Each WFMySQLConnection guarantees that one connection is occupied exclusively. For the details, please see [WFMySQLConnection.h](/src/client/WFMySQLConnection.h).
### 1\. Creating and initializing WFMySQLConnection
When creating a WFMySQLConnection, you need to pass in globally unique **id**, and the subsequent calls on this WFMySQLConnection will use this id to find the corresponding unique connection.
When initializing a WFMySQLConnection, you need to pass a URL, and then you do not need to set the URL for the task created on this connection.
~~~cpp
class WFMySQLConnection
{
public:
WFMySQLConnection(int id);
int init(const std::string& url);
...
};
~~~
### 2\. Creating a task and closing a connection
With **create\_query\_task()**, you can create a task by writing an SQL request and a callback function. The task is garuanteed to be sent on this connection.
Sometimes you need to close this connection manually. Because when you stop using it, this connection will be kept until MySQL server time out. During this period, if you use the same id and url to create a WFMySQLConnection, you may reuse the connection.
Therefore, we suggest that if you do not want to reuse the connection, you should use **create\_disconnect\_task()** to create a task and manually close the connection.
~~~cpp
class WFMySQLConnection
{
public:
...
WFMySQLTask *create_query_task(const std::string& query,
mysql_callback_t callback);
WFMySQLTask *create_disconnect_task(mysql_callback_t callback);
}
~~~
WFMySQLConnection is equivalent to a secondary factory. In the framework, we arrange that the life cycle of any factory object does not need to be maintained until the task ends. The following code is completely legal:
~~~cpp
WFMySQLConnection *conn = new WFMySQLConnection(1234);
conn->init(url);
auto *task = conn->create_query_task("SELECT * from table", my_callback);
conn->deinit();
delete conn;
task->start();
~~~
### 3\. Cautions
If you have started `BEGIN` but have not `COMMIT` or `ROLLBACK` during the transaction and the connection has been interrupted during the transaction, the connection will be automatically reconnected internally by the framework, and you will get **ECONNRESET** error in the next task request. In this case, the transaction statements those have not been `COMMIT` would be expired and you may need to send them again.
### 4\. Preparation
You can also use the WFMySQLConnection for **PREPARE**. And you can easily use it to **defend against SQL injection**. If the connection is reconnected, you also get an **ECONNRESET** error.
### 5\. Complete example
~~~cpp
WFMySQLConnection conn(1);
conn.init("mysql://root@127.0.0.1/test");
// test transaction
const char *query = "BEGIN;";
WFMySQLTask *t1 = conn.create_query_task(query, task_callback);
query = "SELECT * FROM check_tiny FOR UPDATE;";
WFMySQLTask *t2 = conn.create_query_task(query, task_callback);
query = "INSERT INTO check_tiny VALUES (8);";
WFMySQLTask *t3 = conn.create_query_task(query, task_callback);
query = "COMMIT;";
WFMySQLTask *t4 = conn.create_query_task(query, task_callback);
WFMySQLTask *t5 = conn.create_disconnect_task(task_callback);
((*t1) > t2 > t3 > t4 > t5).start();
~~~