Use stream based API

README.md

@@ -41,17 +41,11 @@ for the required multicast socket options and constants.
 
 These options are only available to the low level socket API (ext-sockets), not
 to the newer stream based networking API.
-Because of this, this library depends on sockets based on `ext-sockets`, provided
-via [clue/socket-react](https://github.com/clue/php-socket-react)
-and [clue/socket-raw](https://github.com/clue/php-socket-raw).
-
 For the most part, React PHP is built around the general purpose stream based API
 and has only somewhat limited support for the low level socket API.
-The package [clue/socket-react](https://github.com/clue/php-socket-react)
-works around this for the most part.
-Simply put, you should try to avoid using the default `StreamSelectLoop`,
-as it requires a workaround to poll the socket resources via a periodic timer
-every 10ms.
+Because of this, this library uses a workaround to create stream based sockets
+and then sets the required socket options on its underlying low level socket
+resource.
 
 This library also provides somewhat limited support for PHP 5.3.
 While this version lacks the required socket options and constants for listening

composer.json

@@ -16,13 +16,13 @@
     "require": {
         "php": ">=5.3",
         "react/event-loop": "~0.3.0|~0.4.0",
-        "react/promise": "~1.0|~2.0",
-        "clue/socket-react": "~0.3.0"
+        "react/datagram": "~1.0"
     },
     "require-dev": {
         "clue/hexdump": "0.2.*"
     },
     "suggest": {
-        "php": "PHP 5.4+ is required for listening on multicast addresses and IGMP announcements"
+        "php": "PHP 5.4+ is required for listening on multicast addresses (socket options to send IGMP announcements)",
+        "ext-sockets": "Low level socket API required for listening on multicast addresses (socket options to send IGMP announcements)"
     }
 }

src/Factory.php

@@ -3,57 +3,67 @@
 namespace Clue\React\Multicast;
 
 use React\EventLoop\LoopInterface;
-use Socket\React\Datagram\Factory as DatagramFactory;
-use Socket\Raw\Factory as RawFactory;
+use React\Datagram\Socket as DatagramSocket;
 use BadMethodCallException;
+use RuntimeException;
 
 class Factory
 {
     private $loop;
-    private $rawFactory;
-    private $datagramFactory;
 
-    public function __construct(LoopInterface $loop, RawFactory $rawFactory = null, DatagramFactory $datagramFactory = null)
+    public function __construct(LoopInterface $loop)
     {
-        if ($rawFactory === null) {
-            $rawFactory = new RawFactory();
-        }
-
-        if ($datagramFactory === null) {
-            $datagramFactory = new DatagramFactory($loop);
-        }
-
-        $this->rawFactory = $rawFactory;
-        $this->datagramFactory = $datagramFactory;
+        $this->loop = $loop;
     }
 
     public function createSender()
     {
-        $socket = $this->rawFactory->createUdp4();
-        return $this->datagramFactory->createFromRaw($socket);
+        $stream = @stream_socket_server('udp://0.0.0.0:0', $errno, $errstr, STREAM_SERVER_BIND);
+        if ($stream === false) {
+            throw new RuntimeException('Unable to create sending socket: ' . $errstr, $errno);
+        }
+
+        return new DatagramSocket($this->loop, $stream);
     }
 
     public function createReceiver($address)
     {
         if (!defined('MCAST_JOIN_GROUP')) {
-            throw new BadMethodCallException('MCAST_JOIN_GROUP not defined');
+            throw new BadMethodCallException('MCAST_JOIN_GROUP not defined (requires PHP 5.4+)');
+        }
+        if (!function_exists('socket_import_stream')) {
+            throw new BadMethodCallException('Function socket_import_stream missing (requires ext-sockets and PHP 5.4+)');
         }
 
         $parts = parse_url('udp://' . $address);
 
-        $socket = $this->rawFactory->createUdp4();
+        $stream = @stream_socket_server('udp://0.0.0.0:' . $parts['port'], $errno, $errstr, STREAM_SERVER_BIND);
+        if ($stream === false) {
+            throw new RuntimeException('Unable to create receiving socket: ' . $errstr, $errno);
+        }
+
+        $socket = socket_import_stream($stream);
+        if ($stream === false) {
+            throw new RuntimeException('Unable to access underlying socket resource');
+        }
 
         // allow multiple processes to bind to the same address
-        $socket->setOption(SOL_SOCKET, SO_REUSEADDR, 1);
+        $ret = socket_set_option($socket, SOL_SOCKET, SO_REUSEADDR, 1);
+        if ($ret === false) {
+            throw new RuntimeException('Unable to enable SO_REUSEADDR');
+        }
 
         // join multicast group and bind to port
-        $socket->setOption(
+        $ret = socket_set_option(
+            $socket,
             IPPROTO_IP,
             MCAST_JOIN_GROUP,
             array('group' => $parts['host'], 'interface' => 0)
         );
-        $socket->bind('0.0.0.0:' . $parts['port']);
+        if ($ret === false) {
+            throw new RuntimeException('Unable to join multicast group');
+        }
 
-        return $this->datagramFactory->createFromRaw($socket);
+        return new DatagramSocket($this->loop, $stream);
     }
 }