add tests, allow to close in write_func

ext-src/swoole_http_client_coro.cc

@@ -123,10 +123,12 @@ class Client {
  bool websocket_compression = false; // allow to compress websocket messages
  bool accept_websocket_compression = false; // websocket server accepts compression
 #endif
+ bool in_callback;
+ bool has_upload_files = false;
+
  File *download_file = nullptr; // save http response to file
  zend::String download_file_name; // unlink the file on error
  zend_long download_offset = 0;
- bool has_upload_files = false;
 
  /* safety zval */
  zval _zobject;
@@ -479,7 +481,9 @@ static int http_parser_on_body(swoole_http_parser *parser, const char *at, size_
  zval zargv[2];
  zargv[0] = *http->zobject;
  ZVAL_STRINGL(&zargv[1], at, length);
+ http->in_callback = true;
  bool success = http->write_func->call(2, zargv, nullptr);
+ http->in_callback = false;
  zval_ptr_dtor(&zargv[1]);
  return success ? 0 : -1;
  }
@@ -740,7 +744,10 @@ void Client::apply_setting(zval *zset, const bool check_all) {
  if (!write_func->is_callable()) {
  delete write_func;
  write_func = nullptr;
- zend_throw_exception_ex(swoole_exception_ce, SW_ERROR_INVALID_PARAMS, "write_func must be of type callable, %s given", zend_zval_type_name(ztmp));
+ zend_throw_exception_ex(swoole_exception_ce,
+ SW_ERROR_INVALID_PARAMS,
+ "write_func must be of type callable, %s given",
+ zend_zval_type_name(ztmp));
  }
  }
  }
@@ -1417,6 +1424,10 @@ bool Client::recv_response(double timeout) {
  retval,
  total_bytes,
  parser.state == s_start_res);
+ if (socket->get_socket()->close_wait) {
+ success = false;
+ break;
+ }
  if (parser.state == s_start_res) {
  // handle redundant data (websocket packet)
  if (parser.upgrade && (size_t) retval > parsed_n + SW_WEBSOCKET_HEADER_LEN) {
@@ -1600,6 +1611,10 @@ bool Client::close(const bool should_be_reset) {
  if (!_socket) {
  return false;
  }
+ if (in_callback) {
+ _socket->get_socket()->close_wait = 1;
+ return true;
+ }
  zend_update_property_bool(Z_OBJCE_P(zobject), SW_Z8_OBJ_P(zobject), ZEND_STRL("connected"), 0);
  if (!_socket->close()) {
  php_swoole_socket_set_error_properties(zobject, _socket);

tests/swoole_http_client_coro/write_func_1.phpt

@@ -0,0 +1,48 @@
+--TEST--
+swoole_http_client_coro: write func 1
+--SKIPIF--
+<?php require __DIR__ . '/../include/skipif.inc'; ?>
+--FILE--
+<?php
+require __DIR__ . '/../include/bootstrap.php';
+
+const N = 8;
+$chunks = [];
+$n = N;
+while ($n--) {
+ $chunks[] = base64_encode(random_bytes(random_int(256, 4096)));
+}
+
+$pm = new ProcessManager;
+$pm->parentFunc = function ($pid) use ($pm, $chunks) {
+ Co\run(function () use ($pm, $chunks) {
+ $cli = new Swoole\Coroutine\Http\Client('127.0.0.1', $pm->getFreePort());
+ $index = 0;
+ $cli->set(['write_func' => function ($client, $data) use ($chunks, &$index) {
+ Assert::eq($chunks[$index], $data);
+ $index++;
+ }]);
+ Assert::assert($cli->get('/'));
+ });
+ $pm->kill();
+ echo "DONE\n";
+};
+
+$pm->childFunc = function () use ($pm, $chunks) {
+ Co\run(function () use ($pm, $chunks) {
+ $server = new Swoole\Coroutine\Http\Server('127.0.0.1', $pm->getFreePort());
+ $server->handle('/', function ($req, $resp) use ($server, $chunks) {
+ foreach ($chunks as $chunk) {
+ $resp->write($chunk);
+ usleep(mt_rand(10, 50) * 1000);
+ }
+ });
+ $server->start();
+ });
+};
+
+$pm->childFirst();
+$pm->run();
+?>
+--EXPECT--
+DONE

tests/swoole_http_client_coro/write_func_2.phpt

@@ -0,0 +1,54 @@
+--TEST--
+swoole_http_client_coro: write func 1
+--SKIPIF--
+<?php require __DIR__ . '/../include/skipif.inc'; ?>
+--FILE--
+<?php
+require __DIR__ . '/../include/bootstrap.php';
+
+const N = 16;
+$chunks = [];
+$n = N;
+while ($n--) {
+ $chunks[] = base64_encode(random_bytes(random_int(256, 4096)));
+}
+
+$pm = new ProcessManager;
+$pm->parentFunc = function ($pid) use ($pm, $chunks) {
+ Co\run(function () use ($pm, $chunks) {
+ $cli = new Swoole\Coroutine\Http\Client('127.0.0.1', $pm->getFreePort());
+ $index = 0;
+ $cli->set(['write_func' => function ($client, $data) use ($chunks, &$index) {
+ Assert::eq($chunks[$index], $data);
+ $index++;
+ if ($index == N / 2) {
+ // reset connection
+ $client->close();
+ }
+ }]);
+ Assert::false($cli->get('/'));
+ Assert::eq($cli->getStatusCode(), SWOOLE_HTTP_CLIENT_ESTATUS_SERVER_RESET);
+ });
+ $pm->kill();
+ echo "DONE\n";
+};
+
+$pm->childFunc = function () use ($pm, $chunks) {
+ Co\run(function () use ($pm, $chunks) {
+ $server = new Swoole\Coroutine\Http\Server('127.0.0.1', $pm->getFreePort());
+ $server->handle('/', function ($req, $resp) use ($server, $chunks) {
+ foreach ($chunks as $chunk) {
+ $resp->write($chunk);
+ usleep(mt_rand(10, 50) * 1000);
+ }
+ $resp->end();
+ });
+ $server->start();
+ });
+};
+
+$pm->childFirst();
+$pm->run();
+?>
+--EXPECT--
+DONE