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