Browse Source
Merge branch 'PHP-5.6'
Merge branch 'PHP-5.6'
* PHP-5.6: Support async pgsql connections and non-blocking queriespull/609/merge
Daniel Lowrey
13 years ago
7 changed files with 761 additions and 98 deletions
-
479ext/pgsql/pgsql.c
-
24ext/pgsql/php_pgsql.h
-
44ext/pgsql/tests/29nb_async_connect.phpt
-
78ext/pgsql/tests/30nb_async_query_params.phpt
-
112ext/pgsql/tests/31nb_async_query_prepared.phpt
-
84ext/pgsql/tests/32nb_async_query.phpt
-
38ext/pgsql/tests/nonblocking.inc
@ -0,0 +1,44 @@ |
|||
--TEST-- |
|||
PostgreSQL non-blocking async connect |
|||
--SKIPIF-- |
|||
<?php |
|||
include("skipif.inc"); |
|||
?> |
|||
--FILE-- |
|||
<?php |
|||
|
|||
include('config.inc'); |
|||
include('nonblocking.inc'); |
|||
|
|||
if (!$db = pg_connect($conn_str, PGSQL_CONNECT_ASYNC)) { |
|||
die("pg_connect() error"); |
|||
} elseif (pg_connection_status($db) === PGSQL_CONNECTION_BAD) { |
|||
die("pg_connect() error"); |
|||
} elseif ($db_socket = pg_socket($db)) { |
|||
stream_set_blocking($db_socket, FALSE); |
|||
} else { |
|||
die("pg_socket() error"); |
|||
} |
|||
|
|||
while (TRUE) { |
|||
switch ($status = pg_connect_poll($db)) { |
|||
case PGSQL_POLLING_READING: |
|||
if (nb_is_readable($db_socket)) { break 2; } |
|||
break; |
|||
case PGSQL_POLLING_WRITING: |
|||
if (nb_is_writable($db_socket)) { break 2; } |
|||
break; |
|||
case PGSQL_POLLING_FAILED: |
|||
die("async connection failed"); |
|||
case PGSQL_POLLING_OK: |
|||
break 2; |
|||
} |
|||
} |
|||
assert(pg_connection_status($db) === PGSQL_CONNECTION_MADE); |
|||
echo "OK"; |
|||
|
|||
pg_close($db); |
|||
|
|||
?> |
|||
--EXPECT-- |
|||
OK |
|||
@ -0,0 +1,78 @@ |
|||
--TEST-- |
|||
PostgreSQL non-blocking async query params |
|||
--SKIPIF-- |
|||
<?php |
|||
include("skipif.inc"); |
|||
if (!function_exists('pg_send_query_params')) die('skip function pg_send_query_params() does not exist'); |
|||
?> |
|||
--FILE-- |
|||
<?php |
|||
|
|||
include('config.inc'); |
|||
include('nonblocking.inc'); |
|||
|
|||
$db = pg_connect($conn_str); |
|||
|
|||
$version = pg_version($db); |
|||
if ($version['protocol'] < 3) { |
|||
echo "OK"; |
|||
exit(0); |
|||
} |
|||
|
|||
$db_socket = pg_socket($db); |
|||
stream_set_blocking($db_socket, false); |
|||
|
|||
$sent = pg_send_query_params($db, "SELECT * FROM ".$table_name." WHERE num > \$1;", array(100)); |
|||
if ($sent === FALSE) { |
|||
echo "pg_send_query_params() error\n"; |
|||
} elseif ($sent === 0) { |
|||
nb_flush($db, $db_socket); |
|||
} |
|||
|
|||
nb_consume($db, $db_socket); |
|||
|
|||
if (!($result = pg_get_result($db))) { |
|||
echo "pg_get_result() error\n"; |
|||
} |
|||
if (!($rows = pg_num_rows($result))) { |
|||
echo "pg_num_rows() error\n"; |
|||
} |
|||
for ($i=0; $i < $rows; $i++) { |
|||
pg_fetch_array($result, $i, PGSQL_NUM); |
|||
} |
|||
for ($i=0; $i < $rows; $i++) { |
|||
pg_fetch_object($result); |
|||
} |
|||
for ($i=0; $i < $rows; $i++) { |
|||
pg_fetch_row($result, $i); |
|||
} |
|||
for ($i=0; $i < $rows; $i++) { |
|||
pg_fetch_result($result, $i, 0); |
|||
} |
|||
|
|||
pg_num_rows(pg_query_params($db, "SELECT * FROM ".$table_name." WHERE num > \$1;", array(100))); |
|||
pg_num_fields(pg_query_params($db, "SELECT * FROM ".$table_name." WHERE num > \$1;", array(100))); |
|||
pg_field_name($result, 0); |
|||
pg_field_num($result, $field_name); |
|||
pg_field_size($result, 0); |
|||
pg_field_type($result, 0); |
|||
pg_field_prtlen($result, 0); |
|||
pg_field_is_null($result, 0); |
|||
|
|||
$sent = pg_send_query_params($db, "INSERT INTO ".$table_name." VALUES (\$1, \$2);", array(9999, "A'BC")); |
|||
|
|||
if ($sent === FALSE) { |
|||
echo "pg_send_query_params() error\n"; |
|||
} elseif ($sent === 0) { |
|||
nb_flush($db, $db_socket); |
|||
} |
|||
|
|||
pg_last_oid($result); |
|||
pg_free_result($result); |
|||
|
|||
pg_close($db); |
|||
|
|||
echo "OK"; |
|||
?> |
|||
--EXPECT-- |
|||
OK |
|||
@ -0,0 +1,112 @@ |
|||
--TEST-- |
|||
PostgreSQL non-blocking async prepared queries |
|||
--SKIPIF-- |
|||
<?php |
|||
include("skipif.inc"); |
|||
if (!function_exists('pg_send_prepare')) die('skip function pg_send_prepare() does not exist'); |
|||
?> |
|||
--FILE-- |
|||
<?php |
|||
|
|||
include('config.inc'); |
|||
include('nonblocking.inc'); |
|||
|
|||
$db = pg_connect($conn_str); |
|||
|
|||
$version = pg_version($db); |
|||
if ($version['protocol'] < 3) { |
|||
echo "OK"; |
|||
exit(0); |
|||
} |
|||
|
|||
$db_socket = pg_socket($db); |
|||
stream_set_blocking($db_socket, false); |
|||
|
|||
$nb_send = pg_send_prepare($db, 'php_test', "SELECT * FROM ".$table_name." WHERE num > \$1;"); |
|||
if ($nb_send === FALSE) { |
|||
echo "pg_send_prepare() error\n"; |
|||
} elseif ($nb_send === 0) { |
|||
nb_flush($db, $db_socket); |
|||
} |
|||
|
|||
nb_consume($db, $db_socket); |
|||
|
|||
if (!($result = pg_get_result($db))) { |
|||
echo "pg_get_result() error\n"; |
|||
} |
|||
pg_free_result($result); |
|||
|
|||
$nb_send = pg_send_execute($db, 'php_test', array(100)); |
|||
if ($nb_send === FALSE) { |
|||
echo "pg_send_execute() error\n"; |
|||
} elseif ($nb_send === 0) { |
|||
nb_flush($db, $db_socket); |
|||
} |
|||
|
|||
nb_consume($db, $db_socket); |
|||
|
|||
if (!($result = pg_get_result($db))) { |
|||
echo "pg_get_result() error\n"; |
|||
} |
|||
|
|||
if (!($rows = pg_num_rows($result))) { |
|||
echo "pg_num_rows() error\n"; |
|||
} |
|||
for ($i=0; $i < $rows; $i++) { |
|||
pg_fetch_array($result, $i, PGSQL_NUM); |
|||
} |
|||
for ($i=0; $i < $rows; $i++) { |
|||
pg_fetch_object($result); |
|||
} |
|||
for ($i=0; $i < $rows; $i++) { |
|||
pg_fetch_row($result, $i); |
|||
} |
|||
for ($i=0; $i < $rows; $i++) { |
|||
pg_fetch_result($result, $i, 0); |
|||
} |
|||
|
|||
pg_num_rows(pg_query_params($db, "SELECT * FROM ".$table_name." WHERE num > \$1;", array(100))); |
|||
pg_num_fields(pg_query_params($db, "SELECT * FROM ".$table_name." WHERE num > \$1;", array(100))); |
|||
pg_field_name($result, 0); |
|||
pg_field_num($result, $field_name); |
|||
pg_field_size($result, 0); |
|||
pg_field_type($result, 0); |
|||
pg_field_prtlen($result, 0); |
|||
pg_field_is_null($result, 0); |
|||
|
|||
$nb_send = pg_send_prepare($db, "php_test2", "INSERT INTO ".$table_name." VALUES (\$1, \$2);"); |
|||
if ($nb_send === FALSE) { |
|||
echo "pg_send_prepare() error\n"; |
|||
} elseif ($nb_send === 0) { |
|||
nb_flush($db, $db_socket); |
|||
} |
|||
|
|||
nb_consume($db, $db_socket); |
|||
|
|||
if (!($result = pg_get_result($db))) { |
|||
echo "pg_get_result() error\n"; |
|||
} |
|||
pg_free_result($result); |
|||
|
|||
$nb_send = pg_send_execute($db, "php_test2", array(9999, "A'BC")); |
|||
if ($nb_send === FALSE) { |
|||
echo "pg_send_execute() error\n"; |
|||
} elseif ($nb_send === 0) { |
|||
nb_flush($db, $db_socket); |
|||
} |
|||
|
|||
nb_consume($db, $db_socket); |
|||
|
|||
if (!($result = pg_get_result($db))) { |
|||
echo "pg_get_result() error\n"; |
|||
} |
|||
|
|||
pg_last_oid($result); |
|||
pg_free_result($result); |
|||
pg_close($db); |
|||
|
|||
echo "OK"; |
|||
?> |
|||
--EXPECT-- |
|||
OK |
|||
|
|||
@ -0,0 +1,84 @@ |
|||
--TEST-- |
|||
PostgreSQL non-blocking async queries |
|||
--SKIPIF-- |
|||
<?php |
|||
include("skipif.inc"); |
|||
if (!function_exists('pg_send_prepare')) die('skip function pg_send_prepare() does not exist'); |
|||
?> |
|||
--FILE-- |
|||
<?php |
|||
|
|||
include('config.inc'); |
|||
include('nonblocking.inc'); |
|||
|
|||
$db = pg_connect($conn_str); |
|||
|
|||
$version = pg_version($db); |
|||
if ($version['protocol'] < 3) { |
|||
echo "OK"; |
|||
exit(0); |
|||
} |
|||
|
|||
$db_socket = pg_socket($db); |
|||
stream_set_blocking($db_socket, false); |
|||
|
|||
$nb_send = pg_send_query($db, "SELECT * FROM ".$table_name.";"); |
|||
if ($nb_send === FALSE) { |
|||
echo "pg_send_query() error\n"; |
|||
} elseif ($nb_send === 0) { |
|||
nb_flush($db, $db_socket); |
|||
} |
|||
|
|||
nb_consume($db, $db_socket); |
|||
|
|||
if (!($result = pg_get_result($db))) { |
|||
echo "pg_get_result() error\n"; |
|||
} |
|||
|
|||
if (!($rows = pg_num_rows($result))) { |
|||
echo "pg_num_rows() error\n"; |
|||
} |
|||
for ($i=0; $i < $rows; $i++) { |
|||
pg_fetch_array($result, $i, PGSQL_NUM); |
|||
} |
|||
for ($i=0; $i < $rows; $i++) { |
|||
pg_fetch_object($result); |
|||
} |
|||
for ($i=0; $i < $rows; $i++) { |
|||
pg_fetch_row($result, $i); |
|||
} |
|||
for ($i=0; $i < $rows; $i++) { |
|||
pg_fetch_result($result, $i, 0); |
|||
} |
|||
|
|||
pg_num_rows(pg_query($db, "SELECT * FROM ".$table_name.";")); |
|||
pg_num_fields(pg_query($db, "SELECT * FROM ".$table_name.";")); |
|||
pg_field_name($result, 0); |
|||
pg_field_num($result, $field_name); |
|||
pg_field_size($result, 0); |
|||
pg_field_type($result, 0); |
|||
pg_field_prtlen($result, 0); |
|||
pg_field_is_null($result, 0); |
|||
|
|||
$nb_send = pg_send_query($db, "INSERT INTO ".$table_name." VALUES (8888, 'GGG');"); |
|||
if ($nb_send === FALSE) { |
|||
echo "pg_send_query() error\n"; |
|||
} elseif ($nb_send === 0) { |
|||
nb_flush($db, $db_socket); |
|||
} |
|||
|
|||
nb_consume($db, $db_socket); |
|||
|
|||
if (!($result = pg_get_result($db))) { |
|||
echo "pg_get_result() error\n"; |
|||
} |
|||
|
|||
pg_last_oid($result); |
|||
pg_free_result($result); |
|||
pg_close($db); |
|||
|
|||
echo "OK"; |
|||
?> |
|||
--EXPECT-- |
|||
OK |
|||
|
|||
@ -0,0 +1,38 @@ |
|||
<?php |
|||
|
|||
function nb_is_readable($stream, $timeout = 1) { |
|||
$r = [$stream]; $w = []; $e = []; |
|||
return (bool) stream_select($r, $w, $e, $timeout, 0); |
|||
}; |
|||
function nb_is_writable($stream, $timeout = 1) { |
|||
$r = []; $w = [$stream]; $e = []; |
|||
return (bool) stream_select($r, $w, $e, $timeout, 0); |
|||
}; |
|||
function nb_flush($db, $db_socket) { |
|||
while (TRUE) { |
|||
if (! nb_is_writable($db_socket)) { |
|||
continue; |
|||
} |
|||
$flush = pg_flush($db); |
|||
if ($flush === TRUE) { |
|||
break; // All data flushed |
|||
} elseif ($flush === FALSE) { |
|||
echo "pg_flush() error\n"; |
|||
break; |
|||
} |
|||
} |
|||
}; |
|||
function nb_consume($db, $db_socket) { |
|||
while (TRUE) { |
|||
if (!nb_is_readable($db_socket)) { |
|||
continue; |
|||
} elseif (!pg_consume_input($db)) { |
|||
echo "pg_consume_input() error\n"; |
|||
break; |
|||
} elseif (!pg_connection_busy($db)) { |
|||
break; // All data consumed |
|||
} |
|||
|
|||
} |
|||
}; |
|||
|
|||
Write
Preview
Loading…
Cancel
Save
Reference in new issue