Implement GH-8641: [Stream] STREAM_NOTIFY_COMPLETED over HTTP never emitted

This adds support for the completed event. Since the read handler could
be entered twice towards the end of the stream we remember what the eof
flag was before reading so we can emit the completed event when the flag
changes to true.

Author: nielsdos

ext/openssl/xp_ssl.c

@@ -2035,6 +2035,7 @@ static ssize_t php_openssl_sockop_io(int read, php_stream *stream, char *buf, si
 		}
 
 		/* Main IO loop. */
+		bool old_eof = stream->eof;
 		do {
 			struct timeval cur_time, elapsed_time, left_time;
 
@@ -2143,6 +2144,9 @@ static ssize_t php_openssl_sockop_io(int read, php_stream *stream, char *buf, si
 		if (nr_bytes > 0) {
 			php_stream_notify_progress_increment(PHP_STREAM_CONTEXT(stream), nr_bytes, 0);
 		}
+		if (old_eof != stream->eof) {
+			php_stream_notify_completed(PHP_STREAM_CONTEXT(stream));
+		}
 
 		/* And if we were originally supposed to be blocking, let's reset the socket to that. */
 		if (began_blocked && php_set_sock_blocking(sslsock->s.socket, 1) == SUCCESS) {

ext/standard/tests/http/gh8641.phpt

@@ -0,0 +1,34 @@
+--TEST--
+GH-8641 ([Stream] STREAM_NOTIFY_COMPLETED over HTTP never emitted)
+--SKIPIF--
+<?php require 'server.inc'; http_server_skipif(); ?>
+--INI--
+allow_url_fopen=1
+--FILE--
+<?php
+require 'server.inc';
+
+function stream_notification_callback($notification_code, $severity, $message, $message_code, $bytes_transferred, $bytes_max)
+{
+    if ($notification_code === STREAM_NOTIFY_COMPLETED) {
+        echo $notification_code, ' ', $bytes_transferred, ' ', $bytes_max, PHP_EOL;
+    }
+}
+
+$ctx = stream_context_create();
+stream_context_set_params($ctx, array("notification" => "stream_notification_callback"));
+
+$responses = array(
+    "data://text/plain,HTTP/1.0 200 Ok\r\nContent-Length: 11\r\n\r\nHello world",
+);
+
+['pid' => $pid, 'uri' => $uri] = http_server($responses, $output);
+
+$f = file_get_contents($uri, 0, $ctx);
+
+http_server_kill($pid);
+var_dump($f);
+?>
+--EXPECTF--
+8 11 11
+string(11) "Hello world"

main/streams/php_stream_context.h

@@ -94,6 +94,10 @@ END_EXTERN_C()
 	php_stream_notification_notify((context), PHP_STREAM_NOTIFY_PROGRESS, PHP_STREAM_NOTIFY_SEVERITY_INFO, \
 			NULL, 0, (bsofar), (bmax), NULL); } } while(0)
 
+#define php_stream_notify_completed(context) do { if ((context) && (context)->notifier) { \
+	php_stream_notification_notify((context), PHP_STREAM_NOTIFY_COMPLETED, PHP_STREAM_NOTIFY_SEVERITY_INFO, \
+			NULL, 0, (context)->notifier->progress, (context)->notifier->progress_max, NULL); } } while(0)
+
 #define php_stream_notify_progress_init(context, sofar, bmax) do { if ((context) && (context)->notifier) { \
 	(context)->notifier->progress = (sofar); \
 	(context)->notifier->progress_max = (bmax); \

main/streams/xp_socket.c

@@ -189,9 +189,11 @@ static ssize_t php_sockop_read(php_stream *stream, char *buf, size_t count)
 	}
 
 	ssize_t nr_bytes = recv(sock->socket, buf, XP_SOCK_BUF_SIZE(count), recv_flags);
-	int err = php_socket_errno();
+
+	bool old_eof = stream->eof;
 
 	if (nr_bytes < 0) {
+		int err = php_socket_errno();
 		if (PHP_IS_TRANSIENT_ERROR(err)) {
 			nr_bytes = 0;
 		} else {
@@ -204,6 +206,9 @@ static ssize_t php_sockop_read(php_stream *stream, char *buf, size_t count)
 	if (nr_bytes > 0) {
 		php_stream_notify_progress_increment(PHP_STREAM_CONTEXT(stream), nr_bytes, 0);
 	}
+	if (old_eof != stream->eof) {
+		php_stream_notify_completed(PHP_STREAM_CONTEXT(stream));
+	}
 
 	return nr_bytes;
 }