Update Udp.php

Domraider · Jul 13, 2016 · 3ae2710 · 3ae2710
1 parent 9786d2a
commit 3ae2710
Showing 1 changed file with 7 additions and 4 deletions.
diff --git a/src/Rxnet/Connector/Udp.php b/src/Rxnet/Connector/Udp.php
@@ -1,6 +1,8 @@
 <?php
 namespace Rxnet\Connector;
 
+use Rx\Observable;
+use Rx\ObserverInterface;
 use Rxnet\Event\ConnectorEvent;
 use Rxnet\Transport\Datagram;
 
@@ -18,9 +20,10 @@ protected function createSocketForAddress()
     {
         $socket = parent::createSocketForAddress();
 
-        $stream = new Datagram($socket, $this->loop);
-        $this->notifyNext(new ConnectorEvent('/connector/connected', $stream));
-        $this->notifyCompleted();
-        return $stream;
+        return Observable::create(function(ObserverInterface $observer) use($socket) {
+            $stream = new Datagram($socket, $this->loop);
+            $observer->onNext(new ConnectorEvent('/connector/connected', $stream));
+            $observer->onCompleted();
+        });
     }
 }