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

This adds support for the completed event. We also have to check for
stream->eof in the `stream->readfilters.head` else branch in order to
prevent double firing the completed event. The non-else branch already
has that check.

Author: nielsdos

ext/openssl/xp_ssl.c

@@ -2140,6 +2140,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 (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/streams.c

@@ -646,7 +646,7 @@ PHPAPI zend_result _php_stream_fill_read_buffer(php_stream *stream, size_t size)
 
 	} else {
 		/* is there enough data in the buffer ? */
-		if (stream->writepos - stream->readpos < (zend_off_t)size) {
+		if (!stream->eof && stream->writepos - stream->readpos < (zend_off_t)size) {
 			ssize_t justread = 0;
 
 			/* reduce buffer memory consumption if possible, to avoid a realloc */

main/streams/xp_socket.c

@@ -116,6 +116,9 @@ static ssize_t php_sockop_write(php_stream *stream, const char *buf, size_t coun
 	if (didwrite > 0) {
 		php_stream_notify_progress_increment(PHP_STREAM_CONTEXT(stream), didwrite, 0);
 	}
+	if (stream->eof) {
+		php_stream_notify_completed(PHP_STREAM_CONTEXT(stream));
+	}
 
 	return didwrite;
 }
@@ -204,6 +207,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 (stream->eof) {
+		php_stream_notify_completed(PHP_STREAM_CONTEXT(stream));
+	}
 
 	return nr_bytes;
 }