Add stream_socket_sendto and stream_socket_recvfrom which work very much

like sendto() and recvfrom() syscalls.

ext/standard/basic_functions.c

@@ -731,10 +731,12 @@ function_entry basic_functions[] = {
 	PHP_FE(stream_socket_server,				 second_and_third_args_force_ref)
 	PHP_FE(stream_socket_accept,				 		   third_arg_force_ref)
 	PHP_FE(stream_socket_get_name,											NULL)
+	PHP_FE(stream_socket_recvfrom,							fourth_arg_force_ref)
+	PHP_FE(stream_socket_sendto,											NULL)
 	PHP_FE(stream_copy_to_stream,											NULL)
 	PHP_FE(stream_get_contents,												NULL)
 	PHP_FE(fgetcsv,															NULL)
-	PHP_FE(flock,															NULL)
+	PHP_FE(flock,											 third_arg_force_ref)
 	PHP_FE(get_meta_tags,													NULL)
 	PHP_FE(stream_set_write_buffer,											NULL)
 	PHP_FALIAS(set_file_buffer, stream_set_write_buffer,					NULL)

ext/standard/file.c

@@ -201,6 +201,10 @@ PHP_MINIT_FUNCTION(file)
 
 	REGISTER_LONG_CONSTANT("STREAM_CLIENT_PERSISTENT",		PHP_STREAM_CLIENT_PERSISTENT,		CONST_CS | CONST_PERSISTENT);
 	REGISTER_LONG_CONSTANT("STREAM_CLIENT_ASYNC_CONNECT",	PHP_STREAM_CLIENT_ASYNC_CONNECT,	CONST_CS | CONST_PERSISTENT);
+	REGISTER_LONG_CONSTANT("STREAM_CLIENT_CONNECT",			PHP_STREAM_CLIENT_CONNECT,	CONST_CS | CONST_PERSISTENT);
+
+	REGISTER_LONG_CONSTANT("STREAM_PEEK", STREAM_PEEK, CONST_CS | CONST_PERSISTENT);
+	REGISTER_LONG_CONSTANT("STREAM_OOB",  STREAM_OOB, CONST_CS | CONST_PERSISTENT);
 
 	REGISTER_LONG_CONSTANT("STREAM_SERVER_BIND",			STREAM_XPORT_BIND,					CONST_CS | CONST_PERSISTENT);
 	REGISTER_LONG_CONSTANT("STREAM_SERVER_LISTEN",			STREAM_XPORT_LISTEN,				CONST_CS | CONST_PERSISTENT);

ext/standard/streamsfuncs.c

@@ -54,7 +54,7 @@ PHP_FUNCTION(stream_socket_client)
 	char *hashkey = NULL;
 	php_stream *stream = NULL;
 	int err;
-	long flags = 0;
+	long flags = PHP_STREAM_CLIENT_CONNECT;
 	char *errstr = NULL;
 	php_stream_context *context = NULL;
 
@@ -85,7 +85,7 @@ PHP_FUNCTION(stream_socket_client)
 	}
 
 	stream = php_stream_xport_create(host, host_len, ENFORCE_SAFE_MODE | REPORT_ERRORS,
-			STREAM_XPORT_CLIENT | STREAM_XPORT_CONNECT |
+			STREAM_XPORT_CLIENT | (flags & PHP_STREAM_CLIENT_CONNECT ? STREAM_XPORT_CONNECT : 0) |
 			(flags & PHP_STREAM_CLIENT_ASYNC_CONNECT ? STREAM_XPORT_CONNECT_ASYNC : 0),
 			hashkey, &tv, context, &errstr, &err);
 
@@ -136,7 +136,7 @@ PHP_FUNCTION(stream_socket_server)
 	long host_len;
 	zval *zerrno = NULL, *zerrstr = NULL, *zcontext = NULL;
 	php_stream *stream = NULL;
-	int err;
+	int err = 0;
 	long flags = STREAM_XPORT_BIND | STREAM_XPORT_LISTEN;
 	char *errstr = NULL;
 	php_stream_context *context = NULL;
@@ -266,6 +266,85 @@ PHP_FUNCTION(stream_socket_get_name)
 }
 /* }}} */
 
+/* {{{ proto long stream_socket_sendto(resouce stream, string data [, long flags [, string target_addr]])
+   Send data to a socket stream.  If target_addr is specified it must be in dotted quad (or [ipv6]) format */
+PHP_FUNCTION(stream_socket_sendto)
+{
+	php_stream *stream;
+	zval *zstream;
+	long flags = 0;
+	char *data, *target_addr = NULL;
+	long datalen, target_addr_len = 0;
+	php_sockaddr_storage sa;
+	socklen_t sl = 0;
+	
+	if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "rs|ls", &zstream, &data, &datalen, &flags, &target_addr, &target_addr_len) == FAILURE) {
+		RETURN_FALSE;
+	}
+	php_stream_from_zval(stream, &zstream);
+
+	if (target_addr) {
+		/* parse the address */
+		if (FAILURE == php_network_parse_network_address_with_port(target_addr, target_addr_len, (struct sockaddr*)&sa, &sl TSRMLS_CC)) {
+			php_error_docref(NULL TSRMLS_CC, E_WARNING, "Failed to parse `%s' into a valid network address", target_addr);
+			RETURN_FALSE;
+		}
+	}
+
+	RETURN_LONG(php_stream_xport_sendto(stream, data, datalen, flags, target_addr ? &sa : NULL, sl TSRMLS_CC));
+}
+/* }}} */
+
+/* {{{ proto string stream_socket_recvfrom(resource stream, long amount [, long flags [, string &remote_addr]])
+   Receives data from a socket stream */
+PHP_FUNCTION(stream_socket_recvfrom)
+{
+	php_stream *stream;
+	zval *zstream, *zremote = NULL;
+	long to_read = 0;
+	char *read_buf;
+	long flags = 0;
+	int recvd;
+	
+	if (zend_parse_parameters(ZEND_NUM_ARGS() TSRMLS_CC, "rl|lz", &zstream, &to_read, &flags, &zremote) == FAILURE) {
+		RETURN_FALSE;
+	}
+	
+	php_stream_from_zval(stream, &zstream);
+
+	if (zremote) {
+		zval_dtor(zremote);
+		ZVAL_NULL(zremote);
+		Z_STRLEN_P(zremote) = 0;
+	}
+	
+	read_buf = emalloc(to_read + 1);
+	
+	recvd = php_stream_xport_recvfrom(stream, read_buf, to_read, flags, NULL, NULL,
+			zremote ? &Z_STRVAL_P(zremote) : NULL,
+			zremote ? &Z_STRLEN_P(zremote) : NULL
+			TSRMLS_CC);
+
+	if (recvd >= 0) {
+		if (zremote && Z_STRLEN_P(zremote)) {
+			Z_TYPE_P(zremote) = IS_STRING;
+		}
+		read_buf[recvd] = '\0';
+
+		if (PG(magic_quotes_runtime)) {
+			Z_TYPE_P(return_value) = IS_STRING;
+			Z_STRVAL_P(return_value) = php_addslashes(Z_STRVAL_P(return_value),
+					Z_STRLEN_P(return_value), &Z_STRLEN_P(return_value), 1 TSRMLS_CC);
+			return;
+		} else {
+			RETURN_STRINGL(read_buf, recvd, 0);
+		}
+	}
+
+	RETURN_FALSE;
+}
+/* }}} */
+
 /* {{{ proto long stream_get_contents(resource source [, long maxlen ])
    Reads all remaining bytes (or up to maxlen bytes) from a stream and returns them as a string. */
 PHP_FUNCTION(stream_get_contents)

ext/standard/streamsfuncs.h

@@ -21,11 +21,14 @@
 /* Flags for stream_socket_client */
 #define PHP_STREAM_CLIENT_PERSISTENT	1
 #define PHP_STREAM_CLIENT_ASYNC_CONNECT	2
+#define PHP_STREAM_CLIENT_CONNECT		4
 
 PHP_FUNCTION(stream_socket_client);
 PHP_FUNCTION(stream_socket_server);
 PHP_FUNCTION(stream_socket_accept);
 PHP_FUNCTION(stream_socket_get_name);
+PHP_FUNCTION(stream_socket_recvfrom);
+PHP_FUNCTION(stream_socket_sendto);
 
 PHP_FUNCTION(stream_copy_to_stream);
 PHP_FUNCTION(stream_get_contents);

ext/standard/tests/network/tcp4loop.phpt

@@ -1,7 +1,7 @@
 --TEST--
 Streams Based IPv4 TCP Loopback test
 --FILE--
-<?php
+<?php # vim:ft=php:
 	/* Setup socket server */
 	$server = stream_socket_server('tcp://127.0.0.1:31337');
 	if (!$server) {

main/network.c

@@ -478,7 +478,94 @@ bound:
 }
 /* }}} */
 
-static void populate_name(
+PHPAPI int php_network_parse_network_address_with_port(const char *addr, long addrlen, struct sockaddr *sa, socklen_t *sl TSRMLS_DC)
+{
+	char *colon;
+	char *host = NULL;
+	int is_v6;
+	char *tmp;
+	int ret = FAILURE;
+	short port;
+	struct sockaddr_in *in4 = (struct sockaddr_in*)sa;
+	struct sockaddr **sal, **psal;
+	int n;
+	char *errstr = NULL;
+#ifdef HAVE_IPV6
+	char *p;
+	struct sockaddr_in6 *in6 = (struct sockaddr_in6*)sa;
+#endif
+
+	if (*addr == '[') {
+		colon = memchr(addr + 1, ']', addrlen-1);
+		if (!colon || colon[1] != ':') {
+			return 0;
+		}
+		port = atoi(colon + 2);
+		addr++;
+	} else {
+		colon = memchr(addr, ':', addrlen);
+		port = atoi(colon + 1);
+	}
+
+	tmp = estrndup(addr, colon - addr);
+
+	/* first, try interpreting the address as a numeric address */
+
+#if HAVE_IPV6 && HAVE_INET_PTON
+	if (inet_pton(AF_INET6, tmp, &in6->sin6_addr) > 0) {
+		in6->sin6_port = htons(port);
+		in6->sin6_family = AF_INET6;
+		*sl = sizeof(struct sockaddr_in6);
+		ret = SUCCESS;
+		goto out;
+	}
+#endif
+	if (inet_aton(tmp, &in4->sin_addr) > 0) {
+		in4->sin_port = htons(port);
+		in4->sin_family = AF_INET;
+		*sl = sizeof(struct sockaddr_in);
+		ret = SUCCESS;
+		goto out;
+	}
+
+	/* looks like we'll need to resolve it */
+	n = php_network_getaddresses(tmp, SOCK_DGRAM, &psal, &errstr TSRMLS_CC);
+
+	if (n == 0) {
+		if (errstr) {
+			php_error_docref(NULL TSRMLS_CC, E_WARNING, "Failed to resolve `%s': %s", tmp, errstr);
+			STR_FREE(errstr);
+		}
+		goto out;
+	}
+
+	/* copy the details from the first item */
+	switch ((*psal)->sa_family) {
+#if HAVE_GETADDRINFO && HAVE_IPV6
+		case AF_INET6:
+			*in6 = **(struct sockaddr_in6**)psal;
+			in6->sin6_port = htons(port);
+			*sl = sizeof(struct sockaddr_in6);
+			ret = SUCCESS;
+			break;
+#endif
+		case AF_INET:
+			*in4 = **(struct sockaddr_in**)psal;
+			in4->sin_port = htons(port);
+			*sl = sizeof(struct sockaddr_in);
+			ret = SUCCESS;
+			break;
+	}
+
+	php_network_freeaddresses(psal);
+
+out:
+	STR_FREE(tmp);
+	return ret;
+}
+
+
+PHPAPI void php_network_populate_name_from_sockaddr(
 		/* input address */
 		struct sockaddr *sa, socklen_t sl,
 		/* output readable address */
@@ -556,7 +643,7 @@ PHPAPI int php_network_get_peer_name(php_socket_t sock,
 	socklen_t sl = sizeof(sa);
 	
 	if (getpeername(sock, (struct sockaddr*)&sa, &sl) == 0) {
-		populate_name((struct sockaddr*)&sa, sl,
+		php_network_populate_name_from_sockaddr((struct sockaddr*)&sa, sl,
 				textaddr, textaddrlen,
 				addr, addrlen
 				TSRMLS_CC);
@@ -575,7 +662,7 @@ PHPAPI int php_network_get_sock_name(php_socket_t sock,
 	socklen_t sl = sizeof(sa);
 	
 	if (getsockname(sock, (struct sockaddr*)&sa, &sl) == 0) {
-		populate_name((struct sockaddr*)&sa, sl,
+		php_network_populate_name_from_sockaddr((struct sockaddr*)&sa, sl,
 				textaddr, textaddrlen,
 				addr, addrlen
 				TSRMLS_CC);
@@ -625,7 +712,7 @@ PHPAPI php_socket_t php_network_accept_incoming(php_socket_t srvsock,
 		clisock = accept(srvsock, (struct sockaddr*)&sa, &sl);
 
 		if (clisock >= 0) {
-			populate_name((struct sockaddr*)&sa, sl,
+			php_network_populate_name_from_sockaddr((struct sockaddr*)&sa, sl,
 					textaddr, textaddrlen,
 					addr, addrlen
 					TSRMLS_CC);

main/php_network.h

@@ -173,6 +173,18 @@ PHPAPI php_stream *_php_stream_sock_open_from_socket(php_socket_t socket, const
 /* open a connection to a host using php_hostconnect and return a stream */
 PHPAPI php_stream *_php_stream_sock_open_host(const char *host, unsigned short port,
 		int socktype, struct timeval *timeout, const char *persistent_id STREAMS_DC TSRMLS_DC);
+PHPAPI void php_network_populate_name_from_sockaddr(
+		/* input address */
+		struct sockaddr *sa, socklen_t sl,
+		/* output readable address */
+		char **textaddr, long *textaddrlen,
+		/* output address */
+		struct sockaddr **addr,
+		socklen_t *addrlen
+		TSRMLS_DC);
+
+PHPAPI int php_network_parse_network_address_with_port(const char *addr,
+		long addrlen, struct sockaddr *sa, socklen_t *sl TSRMLS_DC);
 
 #define php_stream_sock_open_from_socket(socket, persistent)	_php_stream_sock_open_from_socket((socket), (persistent) STREAMS_CC TSRMLS_CC)
 #define php_stream_sock_open_host(host, port, socktype, timeout, persistent)	_php_stream_sock_open_host((host), (port), (socktype), (timeout), (persistent) STREAMS_CC TSRMLS_CC)

main/streams/php_stream_transport.h

@@ -88,6 +88,22 @@ PHPAPI int php_stream_xport_get_name(php_stream *stream, int want_peer,
 		void **addr, socklen_t *addrlen
 		TSRMLS_DC);
 
+enum php_stream_xport_send_recv_flags {
+	STREAM_OOB = 1,
+	STREAM_PEEK = 2
+};
+
+/* Similar to recv() system call; read data from the stream, optionally
+ * peeking, optionally retrieving OOB data */
+PHPAPI int php_stream_xport_recvfrom(php_stream *stream, char *buf, size_t buflen,
+		long flags, void **addr, socklen_t *addrlen,
+		char **textaddr, int *textaddrlen TSRMLS_DC);
+
+/* Similar to send() system call; send data to the stream, optionally
+ * sending it as OOB data */
+PHPAPI int php_stream_xport_sendto(php_stream *stream, const char *buf, size_t buflen,
+		long flags, void *addr, socklen_t addrlen TSRMLS_DC);
+
 /* Structure definition for the set_option interface that the above functions wrap */
 
 typedef struct _php_stream_xport_param {
@@ -96,7 +112,9 @@ typedef struct _php_stream_xport_param {
 		STREAM_XPORT_OP_LISTEN, STREAM_XPORT_OP_ACCEPT,
 		STREAM_XPORT_OP_CONNECT_ASYNC,
 		STREAM_XPORT_OP_GET_NAME,
-		STREAM_XPORT_OP_GET_PEER_NAME
+		STREAM_XPORT_OP_GET_PEER_NAME,
+		STREAM_XPORT_OP_RECV,
+		STREAM_XPORT_OP_SEND
 	} op;
 	unsigned int want_addr:1;
 	unsigned int want_textaddr:1;
@@ -107,6 +125,11 @@ typedef struct _php_stream_xport_param {
 		long namelen;
 		int backlog;
 		struct timeval *timeout;
+		struct sockaddr *addr;
+		socklen_t addrlen;
+		char *buf;
+		size_t buflen;
+		long flags;
 	} inputs;
 	struct {
 		php_stream *client;

main/streams/transports.c

@@ -77,6 +77,10 @@ PHPAPI php_stream *_php_stream_xport_create(const char *name, long namelen, int
 				if (PHP_STREAM_OPTION_RETURN_OK == php_stream_set_option(stream, PHP_STREAM_OPTION_CHECK_LIVENESS, 0, NULL)) {
 					return stream;
 				}
+				/* dead - kill it */
+				php_stream_close(stream);
+				stream = NULL;
+
 				/* fall through */
 
 			case PHP_STREAM_PERSISTENT_FAILURE:
@@ -366,6 +370,112 @@ PHPAPI int php_stream_xport_crypto_enable(php_stream *stream, int activate TSRML
 	return ret;
 }
 
+/* Similar to recv() system call; read data from the stream, optionally
+ * peeking, optionally retrieving OOB data */
+PHPAPI int php_stream_xport_recvfrom(php_stream *stream, char *buf, size_t buflen,
+		long flags, void **addr, socklen_t *addrlen, char **textaddr, int *textaddrlen
+		TSRMLS_DC)
+{
+	php_stream_xport_param param;
+	int ret = 0;
+	int recvd_len = 0;
+	int oob;
+
+	if (flags == 0 && addr == NULL) {
+		return php_stream_read(stream, buf, buflen);
+	}
+
+	if (stream->readfilters.head) {
+		php_error_docref(NULL TSRMLS_CC, E_WARNING, "cannot peek or fetch OOB data from a filtered stream");
+		return -1;
+	}
+	
+	oob = (flags & STREAM_OOB) == STREAM_OOB;
+
+	if (!oob && addr == NULL) {
+		/* must be peeking at regular data; copy content from the buffer
+		 * first, then adjust the pointer/len before handing off to the
+		 * stream */
+		recvd_len = stream->writepos - stream->readpos;
+		if (recvd_len > buflen) {
+			recvd_len = buflen;
+		}
+		if (recvd_len) {
+			memcpy(buf, stream->readbuf, recvd_len);
+			buf += recvd_len;
+			buflen -= recvd_len;
+		}
+		/* if we filled their buffer, return */
+		if (buflen == 0) {
+			return recvd_len;
+		}
+	}
+
+	/* otherwise, we are going to bypass the buffer */
+	
+	memset(&param, 0, sizeof(param));
+
+	param.op = STREAM_XPORT_OP_RECV;
+	param.want_addr = addr ? 1 : 0;
+	param.want_textaddr = textaddr ? 1 : 0;
+	param.inputs.buf = buf;
+	param.inputs.buflen = buflen;
+	param.inputs.flags = flags;
+	
+	ret = php_stream_set_option(stream, PHP_STREAM_OPTION_XPORT_API, 0, &param);
+
+	if (ret == PHP_STREAM_OPTION_RETURN_OK) {
+		if (addr) {
+			*addr = param.outputs.addr;
+			*addrlen = param.outputs.addrlen;
+		}
+		if (textaddr) {
+			*textaddr = param.outputs.textaddr;
+			*textaddrlen = param.outputs.textaddrlen;
+		}
+		return recvd_len + param.outputs.returncode;
+	}
+	return recvd_len ? recvd_len : -1;
+}
+
+/* Similar to send() system call; send data to the stream, optionally
+ * sending it as OOB data */
+PHPAPI int php_stream_xport_sendto(php_stream *stream, const char *buf, size_t buflen,
+		long flags, void *addr, socklen_t addrlen TSRMLS_DC)
+{
+	php_stream_xport_param param;
+	int ret = 0;
+	int oob;
+
+	if (flags == 0 && addr == NULL) {
+		return php_stream_write(stream, buf, buflen);
+	}
+	
+	oob = (flags & STREAM_OOB) == STREAM_OOB;
+
+	if ((oob || addr) && stream->writefilters.head) {
+		php_error_docref(NULL TSRMLS_CC, E_WARNING, "cannot write OOB data, or data to a targeted address on a filtered stream");
+		return -1;
+	}
+	
+	memset(&param, 0, sizeof(param));
+
+	param.op = STREAM_XPORT_OP_SEND;
+	param.want_addr = addr ? 1 : 0;
+	param.inputs.buf = (char*)buf;
+	param.inputs.buflen = buflen;
+	param.inputs.flags = flags;
+	param.inputs.addr = addr;
+	param.inputs.addrlen = addrlen;
+	
+	ret = php_stream_set_option(stream, PHP_STREAM_OPTION_XPORT_API, 0, &param);
+
+	if (ret == PHP_STREAM_OPTION_RETURN_OK) {
+		return param.outputs.returncode;
+	}
+	return -1;
+}
+
 
 /*
  * Local variables:

main/streams/xp_socket.c

@@ -193,9 +193,30 @@ static int php_sockop_stat(php_stream *stream, php_stream_statbuf *ssb TSRMLS_DC
 	return fstat(sock->socket, &ssb->sb);
 }
 
+static inline int sock_recvfrom(php_netstream_data_t *sock, char *buf, size_t buflen, int flags,
+		char **textaddr, long *textaddrlen,
+		struct sockaddr **addr, socklen_t *addrlen
+		TSRMLS_DC)
+{
+	php_sockaddr_storage sa;
+	socklen_t sl = sizeof(sa);
+	int ret;
+	int want_addr = textaddr || addr;
+
+	if (want_addr) {
+		ret = recvfrom(sock->socket, buf, buflen, flags, (struct sockaddr*)&sa, &sl);
+		php_network_populate_name_from_sockaddr((struct sockaddr*)&sa, sl,
+			textaddr, textaddrlen, addr, addrlen TSRMLS_CC);
+	} else {
+		ret = recv(sock->socket, buf, buflen, flags);
+	}
+
+	return ret;
+}
+
 static int php_sockop_set_option(php_stream *stream, int option, int value, void *ptrparam TSRMLS_DC)
 {
-	int oldmode;
+	int oldmode, flags;
 	php_netstream_data_t *sock = (php_netstream_data_t*)stream->abstract;
 	php_stream_xport_param *xparam;
 	
@@ -280,6 +301,42 @@ static int php_sockop_set_option(php_stream *stream, int option, int value, void
 							TSRMLS_CC);
 					return PHP_STREAM_OPTION_RETURN_OK;
 
+				case STREAM_XPORT_OP_SEND:
+					flags = 0;
+					if ((xparam->inputs.flags & STREAM_OOB) == STREAM_OOB) {
+						flags |= MSG_OOB;
+					}
+					xparam->outputs.returncode = sendto(sock->socket,
+							xparam->inputs.buf, xparam->inputs.buflen,
+							flags,
+							xparam->inputs.addr,
+							xparam->inputs.addrlen);
+					if (xparam->outputs.returncode == -1) {
+						char *err = php_socket_strerror(php_socket_errno(), NULL, 0);
+						php_error_docref(NULL TSRMLS_CC, E_WARNING,
+						   	"%s\n", err);
+						efree(err);
+					}
+					return PHP_STREAM_OPTION_RETURN_OK;
+
+				case STREAM_XPORT_OP_RECV:
+					flags = 0;
+					if ((xparam->inputs.flags & STREAM_OOB) == STREAM_OOB) {
+						flags |= MSG_OOB;
+					}
+					if ((xparam->inputs.flags & STREAM_PEEK) == STREAM_PEEK) {
+						flags |= MSG_PEEK;
+					}
+					xparam->outputs.returncode = sock_recvfrom(sock,
+							xparam->inputs.buf, xparam->inputs.buflen,
+							flags,
+							xparam->want_textaddr ? &xparam->outputs.textaddr : NULL,
+							xparam->want_textaddr ? &xparam->outputs.textaddrlen : NULL,
+							xparam->want_addr ? &xparam->outputs.addr : NULL,
+							xparam->want_addr ? &xparam->outputs.addrlen : NULL
+							TSRMLS_CC);
+					return PHP_STREAM_OPTION_RETURN_OK;
+
 
 				default:
 					return PHP_STREAM_OPTION_RETURN_NOTIMPL;