Added stream_filter_remove() to cancel a stream filter.

Register filters as resources when
instantiated by stream_filter_(ap|pre)pend().

Export php_stream_filter_flush() internal function to wind buffered data
out of a particular filter until consumed by a later filter or sent to
stream->readbuffer or stream->ops->write()

4 files changed, 94 insertions, 0 deletions

diff --git a/main/php_streams.h b/main/php_streams.h
index 290ec14e32..1c9802f93a 100755
--- a/main/php_streams.h
+++ b/main/php_streams.h
@@ -30,6 +30,7 @@
 BEGIN_EXTERN_C()
 PHPAPI int php_file_le_stream(void);
 PHPAPI int php_file_le_pstream(void);
+PHPAPI int php_file_le_stream_filter(void);
 END_EXTERN_C()
 
 /* {{{ Streams memory debugging stuff */
diff --git a/main/streams/filter.c b/main/streams/filter.c
index da61c714fa..37ab4c14d5 100644
--- a/main/streams/filter.c
+++ b/main/streams/filter.c
@@ -405,6 +405,88 @@ PHPAPI void _php_stream_filter_append(php_stream_filter_chain *chain, php_stream
 
 }
 
+PHPAPI int _php_stream_filter_flush(php_stream_filter *filter, int finish TSRMLS_DC)
+{
+	php_stream_bucket_brigade brig_a = { NULL, NULL }, brig_b = { NULL, NULL }, *inp = &brig_a, *outp = &brig_b, *brig_temp;
+	php_stream_bucket *bucket;
+	php_stream_filter_chain *chain;
+	php_stream_filter *current;
+	php_stream *stream;
+	size_t flushed_size = 0;
+	long flags = (finish ? PSFS_FLAG_FLUSH_CLOSE : PSFS_FLAG_FLUSH_INC);
+
+	if (!filter->chain || !filter->chain->stream) {
+		/* Filter is not attached to a chain, or chain is somehow not part of a stream */
+		return FAILURE;
+	}
+
+	chain = filter->chain;
+	stream = chain->stream;
+
+	for(current = filter; current; current = current->next) {
+		php_stream_filter_status_t status;
+
+		status = filter->fops->filter(stream, filter, inp, outp, NULL, flags TSRMLS_CC);
+		if (status == PSFS_FEED_ME) {
+			/* We've flushed the data far enough */
+			return SUCCESS;
+		}
+		if (status == PSFS_ERR_FATAL) {
+			return FAILURE;
+		}
+		/* Otherwise we have data available to PASS_ON
+			Swap the brigades and continue */
+		brig_temp = inp;
+		inp = outp;
+		outp = brig_temp;
+		outp->head = NULL;
+		outp->tail = NULL;
+
+		flags = PSFS_FLAG_NORMAL;
+	}
+
+	/* Last filter returned data via PSFS_PASS_ON
+		Do something with it */
+
+	for(bucket = inp->head; bucket; bucket = bucket->next) {
+		flushed_size += bucket->buflen;
+	}
+
+	if (flushed_size == 0) {
+		/* Unlikely, but possible */
+		return SUCCESS;
+	}
+
+	if (chain == &(stream->readfilters)) {
+		/* Dump any newly flushed data to the read buffer */
+		if (stream->readpos > 0) {
+			/* Back the buffer up */
+			memcpy(stream->readbuf, stream->readbuf + stream->readpos, stream->writepos - stream->readpos);
+			stream->readpos = 0;
+			stream->writepos -= stream->readpos;
+		}
+		if (flushed_size > (stream->readbuflen - stream->writepos)) {
+			/* Grow the buffer */
+			stream->readbuf = perealloc(stream->readbuf, stream->writepos + flushed_size + stream->chunk_size, stream->is_persistent);
+		}
+		while ((bucket = inp->head)) {
+			memcpy(stream->readbuf + stream->writepos, bucket->buf, bucket->buflen);
+			stream->writepos += bucket->buflen;
+			php_stream_bucket_unlink(bucket TSRMLS_CC);
+			php_stream_bucket_delref(bucket TSRMLS_CC);
+		}
+	} else if (chain == &(stream->writefilters)) {
+		/* Send flushed data to the stream */
+		while ((bucket = inp->head)) {
+			stream->ops->write(stream, bucket->buf, bucket->buflen TSRMLS_CC);
+			php_stream_bucket_unlink(bucket TSRMLS_CC);
+			php_stream_bucket_delref(bucket TSRMLS_CC);
+		}
+	}
+
+	return SUCCESS;
+}
+
 PHPAPI php_stream_filter *php_stream_filter_remove(php_stream_filter *filter, int call_dtor TSRMLS_DC)
 {
 	if (filter->prev) {
diff --git a/main/streams/php_stream_filter_api.h b/main/streams/php_stream_filter_api.h
index 733d1c5221..6ee8192923 100644
--- a/main/streams/php_stream_filter_api.h
+++ b/main/streams/php_stream_filter_api.h
@@ -123,6 +123,7 @@ struct _php_stream_filter {
 BEGIN_EXTERN_C()
 PHPAPI void _php_stream_filter_prepend(php_stream_filter_chain *chain, php_stream_filter *filter TSRMLS_DC);
 PHPAPI void _php_stream_filter_append(php_stream_filter_chain *chain, php_stream_filter *filter TSRMLS_DC);
+PHPAPI int _php_stream_filter_flush(php_stream_filter *filter, int finish TSRMLS_DC);
 PHPAPI php_stream_filter *php_stream_filter_remove(php_stream_filter *filter, int call_dtor TSRMLS_DC);
 PHPAPI void php_stream_filter_free(php_stream_filter *filter TSRMLS_DC);
 PHPAPI php_stream_filter *_php_stream_filter_alloc(php_stream_filter_ops *fops, void *abstract, int persistent STREAMS_DC TSRMLS_DC);
@@ -131,6 +132,7 @@ END_EXTERN_C()
 #define php_stream_filter_alloc_rel(fops, thisptr, persistent) _php_stream_filter_alloc((fops), (thisptr), (persistent) STREAMS_REL_CC TSRMLS_CC)
 #define php_stream_filter_prepend(chain, filter) _php_stream_filter_prepend((chain), (filter) TSRMLS_CC)
 #define php_stream_filter_append(chain, filter) _php_stream_filter_append((chain), (filter) TSRMLS_CC)
+#define php_stream_filter_flush(filter, finish) _php_stream_filter_flush((filter), (finish) TSRMLS_CC)
 
 #define php_stream_is_filtered(stream)	((stream)->readfilters.head || (stream)->writefilters.head)
 
diff --git a/main/streams/streams.c b/main/streams/streams.c
index bb2b137c32..6242afcf82 100755
--- a/main/streams/streams.c
+++ b/main/streams/streams.c
@@ -38,6 +38,7 @@
 static HashTable url_stream_wrappers_hash;
 static int le_stream = FAILURE; /* true global */
 static int le_pstream = FAILURE; /* true global */
+static int le_stream_filter = FAILURE; /* true global */
 
 PHPAPI int php_file_le_stream(void)
 {
@@ -49,6 +50,11 @@ PHPAPI int php_file_le_pstream(void)
 	return le_pstream;
 }
 
+PHPAPI int php_file_le_stream_filter(void)
+{
+	return le_stream_filter;
+}
+
 PHPAPI HashTable *_php_stream_get_url_stream_wrappers_hash(TSRMLS_D)
 {
 	return (FG(stream_wrappers) ? FG(stream_wrappers) : &url_stream_wrappers_hash);
@@ -1370,6 +1376,9 @@ int php_init_stream_wrappers(int module_number TSRMLS_DC)
 	le_stream = zend_register_list_destructors_ex(stream_resource_regular_dtor, NULL, "stream", module_number);
 	le_pstream = zend_register_list_destructors_ex(NULL, stream_resource_persistent_dtor, "persistent stream", module_number);
 
+	/* Filters are cleaned up by the streams they're attached to */
+	le_stream_filter = zend_register_list_destructors_ex(NULL, NULL, "stream filter", module_number);
+
 	return (
 			zend_hash_init(&url_stream_wrappers_hash, 0, NULL, NULL, 1) == SUCCESS
 			&&