Merge pull request #4 from clue/ref

Refactor to ease overwriting socket operations

Author: clue

Datagram/Buffer.php

@@ -8,8 +8,9 @@
 
 class Buffer extends EventEmitter
 {
-    private $loop;
-    private $socket;
+    protected $loop;
+    protected $socket;
+
     private $listening = false;
     private $outgoing = array();
     private $writable = true;
@@ -29,32 +30,27 @@ public function send($data, $remoteAddress = null)
         $this->outgoing []= array($data, $remoteAddress);
 
         if (!$this->listening) {
-            $this->loop->addWriteStream($this->socket, array($this, 'handleWrite'));
+            $this->handleResume();
             $this->listening = true;
         }
     }
 
-    public function handleWrite()
+    public function onWritable()
     {
         list($data, $remoteAddress) = array_shift($this->outgoing);
 
-        if ($remoteAddress === null) {
-            // do not use fwrite() as it obeys the stream buffer size and
-            // packets are not to be split at 8kb
-            $ret = @stream_socket_sendto($this->socket, $data);
-        } else {
-            $ret = @stream_socket_sendto($this->socket, $data, 0, $remoteAddress);
+        try {
+            $this->handleWrite($data, $remoteAddress);
         }
-
-        if ($ret < 0) {
-            $error = error_get_last();
-            $message = 'Unable to send packet: ' . trim($error['message']);
-            $this->emit('error', array(new Exception($message)));
+        catch (Exception $e) {
+            $this->emit('error', array($e, $this));
         }
 
         if (!$this->outgoing) {
-            $this->loop->removeWriteStream($this->socket);
-            $this->listening = false;
+            if ($this->listening) {
+                $this->handlePause();
+                $this->listening = false;
+            }
 
             if (!$this->writable) {
                 $this->close();
@@ -71,7 +67,7 @@ public function close()
         $this->emit('close', array($this));
 
         if ($this->listening) {
-            $this->loop->removeWriteStream($this->socket);
+            $this->handlePause();
             $this->listening = false;
         }
 
@@ -89,8 +85,34 @@ public function end()
 
         $this->writable = false;
 
-        if (!$this->listening) {
+        if (!$this->outgoing) {
             $this->close();
         }
     }
+
+    protected function handlePause()
+    {
+        $this->loop->removeWriteStream($this->socket);
+    }
+
+    protected function handleResume()
+    {
+        $this->loop->addWriteStream($this->socket, array($this, 'onWritable'));
+    }
+
+    protected function handleWrite($data, $remoteAddress)
+    {
+        if ($remoteAddress === null) {
+            // do not use fwrite() as it obeys the stream buffer size and
+            // packets are not to be split at 8kb
+            $ret = @stream_socket_sendto($this->socket, $data);
+        } else {
+            $ret = @stream_socket_sendto($this->socket, $data, 0, $remoteAddress);
+        }
+
+        if ($ret < 0) {
+            $error = error_get_last();
+            throw new Exception('Unable to send packet: ' . trim($error['message']));
+        }
+    }
 }

Datagram/Socket.php

@@ -14,12 +14,16 @@ class Socket extends EventEmitter implements SocketInterface
 
     public $bufferSize = 65536;
 
-    public function __construct(LoopInterface $loop, $socket)
+    public function __construct(LoopInterface $loop, $socket, Buffer $buffer = null)
     {
         $this->loop = $loop;
         $this->socket = $socket;
 
-        $this->buffer = new Buffer($loop, $socket);
+        if ($buffer === null) {
+            $buffer = new Buffer($loop, $socket);
+        }
+        $this->buffer = $buffer;
+
         $that = $this;
         $this->buffer->on('error', function ($error) use ($that) {
             $that->emit('error', array($error, $that));
@@ -65,21 +69,18 @@ public function resume()
         }
     }
 
-    public function onReceive($message)
+    public function onReceive()
     {
-        $data = stream_socket_recvfrom($this->socket, $this->bufferSize, 0, $peer);
-
-        if ($data === false) {
-            // receiving data failed => remote side rejected one of our packets
-            // due to the nature of UDP, there's no way to tell which one exactly
-            // $peer is not filled either
-
+        try {
+            $data = $this->handleReceive($peer);
+        }
+        catch (Exception $e) {
             // emit error message and local socket
-            $this->emit('error', array(new \Exception('Invalid message'), $this));
+            $this->emit('error', array($e, $this));
             return;
         }
 
-        $this->emit('message', array($data, $this->sanitizeAddress($peer), $this));
+        $this->emit('message', array($data, $peer, $this));
     }
 
     public function close()
@@ -91,7 +92,7 @@ public function close()
         $this->emit('close', array($this));
         $this->pause();
 
-        fclose($this->socket);
+        $this->handleClose();
         $this->socket = false;
         $this->buffer->close();
 
@@ -118,4 +119,26 @@ private function sanitizeAddress($address)
         }
         return $address;
     }
+
+    protected function handleReceive(&$peerAddress)
+    {
+        $data = stream_socket_recvfrom($this->socket, $this->bufferSize, 0, $peerAddress);
+
+        if ($data === false) {
+            // receiving data failed => remote side rejected one of our packets
+            // due to the nature of UDP, there's no way to tell which one exactly
+            // $peer is not filled either
+
+            throw new \Exception('Invalid message');
+        }
+
+        $peerAddress = $this->sanitizeAddress($peerAddress);
+
+        return $data;
+    }
+
+    protected function handleClose()
+    {
+        fclose($this->socket);
+    }
 }