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

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;
 }