Merge pull request #23 from jjok/simplify-emit-events

Simplify emitting events

Author: oliverlorenz

README.md

@@ -49,6 +49,8 @@ $loop->run();
 ```php
 <?php
 
+use oliverlorenz\reactphpmqtt\packet\Publish;
+
 require __DIR__ . '/../vendor/autoload.php';
 
 $config = include('config.php');
@@ -63,7 +65,7 @@ $connector = new oliverlorenz\reactphpmqtt\Connector($loop, $resolver, $version)
 
 $p = $connector->create($config['server'], $config['port'], $config['options']);
 $p->then(function(\React\Stream\Stream $stream) use ($connector) {
-    $stream->on('PUBLISH', function(\oliverlorenz\reactphpmqtt\packet\Publish $message) {
+    $stream->on(Publish::EVENT, function(Publish $message) {
         print_r($message);
     });
 
@@ -74,13 +76,13 @@ $p->then(function(\React\Stream\Stream $stream) use ($connector) {
 $loop->run();
 ```
 
-# Run tests
+## Run tests
 
     ./vendor/bin/phpunit -c ./tests/phpunit.xml ./tests
 
 
-# Troubleshooting
+## Troubleshooting
 
-## Why does the connect to localhost:1883 not work?
+### Why does the connect to localhost:1883 not work?
 
 The answer is simple: In the example is the DNS 8.8.8.8 configured. Your local server is not visible for them, so you can't connect.

examples/connectSubscribe.php

@@ -1,5 +1,7 @@
 <?php
 
+use oliverlorenz\reactphpmqtt\packet\Publish;
+
 require __DIR__ . '/../vendor/autoload.php';
 
 $config = include('config.php');
@@ -14,7 +16,7 @@
 
 $p = $connector->create($config['server'], $config['port'], $config['options']);
 $p->then(function(\React\Stream\Stream $stream) use ($connector) {
-    $stream->on('PUBLISH', function(\oliverlorenz\reactphpmqtt\packet\Publish $message) {
+    $stream->on(Publish::EVENT, function(Publish $message) {
         printf(
             'Received payload "%s" for topic "%s"%s',
             $message->getPayload(),

src/Connector.php

@@ -15,10 +15,7 @@
 use oliverlorenz\reactphpmqtt\packet\Factory;
 use oliverlorenz\reactphpmqtt\packet\MessageHelper;
 use oliverlorenz\reactphpmqtt\packet\PingRequest;
-use oliverlorenz\reactphpmqtt\packet\PingResponse;
 use oliverlorenz\reactphpmqtt\packet\Publish;
-use oliverlorenz\reactphpmqtt\packet\PublishReceived;
-use oliverlorenz\reactphpmqtt\packet\PublishRelease;
 use oliverlorenz\reactphpmqtt\packet\Subscribe;
 use oliverlorenz\reactphpmqtt\packet\SubscribeAck;
 use oliverlorenz\reactphpmqtt\packet\Unsubscribe;
@@ -88,31 +85,17 @@ private function listenForPackets(Stream $stream)
             foreach ($messages as $data) {
                 try {
                     $message = Factory::getByMessage($this->version, $data);
-                    echo "received:\t" . get_class($message) . "\n";
+                    $stream->emit($message::EVENT, [$message]);
 
-                    if ($message instanceof ConnectionAck) {
-                        $stream->emit('CONNECTION_ACK', array($message));
-                    } elseif ($message instanceof PingResponse) {
-                        $stream->emit('PING_RESPONSE', array($message));
-                    } elseif ($message instanceof Publish) {
-                        $stream->emit('PUBLISH', array($message));
-                    } elseif ($message instanceof PublishReceived) {
-                        $stream->emit('PUBLISH_RECEIVED', array($message));
-                    } elseif ($message instanceof PublishRelease) {
-                        $stream->emit('PUBLISH_RELEASE', array($message));
-                    } elseif ($message instanceof UnsubscribeAck) {
-                        $stream->emit('UNSUBSCRIBE_ACK', array($message));
-                    } elseif ($message instanceof SubscribeAck) {
-                        $stream->emit('SUBSCRIBE_ACK', array($message));
-                    }
+                    echo "received:\t" . get_class($message) . "\n";
                 } catch (\InvalidArgumentException $ex) {
 
                 }
             }
         });
 
         $deferred = new Deferred();
-        $stream->on('CONNECTION_ACK', function($message) use ($stream, $deferred) {
+        $stream->on(ConnectionAck::EVENT, function($message) use ($stream, $deferred) {
             $deferred->resolve($stream);
         });
 
@@ -183,7 +166,7 @@ public function subscribe(Stream $stream, $topic, $qos = 0)
         $this->sendPacketToStream($stream, $packet);
 
         $deferred = new Deferred();
-        $stream->on('SUBSCRIBE_ACK', function($message) use ($stream, $deferred) {
+        $stream->on(SubscribeAck::EVENT, function($message) use ($stream, $deferred) {
             $deferred->resolve($stream);
         });
 
@@ -202,7 +185,7 @@ public function unsubscribe(Stream $stream, $topic)
         $this->sendPacketToStream($stream, $packet);
 
         $deferred = new Deferred();
-        $stream->on('UNSUBSCRIBE_ACK', function($message) use ($stream, $deferred) {
+        $stream->on(UnsubscribeAck::EVENT, function($message) use ($stream, $deferred) {
             $deferred->resolve($stream);
         });
 

src/packet/ConnectionAck.php

@@ -11,7 +11,9 @@
  * The CONNACK Packet is the packet sent by the Server in response to
  * a CONNECT Packet received from a Client.
  */
-class ConnectionAck extends ControlPacket {
+class ConnectionAck extends ControlPacket
+{
+    const EVENT = 'CONNECTION_ACK';
 
     public static function getControlPacketType()
     {

src/packet/Factory.php

@@ -13,7 +13,7 @@ class Factory {
      * @param $version
      * @param $input
      * @throws \InvalidArgumentException
-     * @return ControlPacket
+     * @return ConnectionAck|PingResponse|SubscribeAck|Publish|PublishComplete|PublishRelease|PublishReceived
      */
     public static function getByMessage($version, $input)
     {

src/packet/PingResponse.php

@@ -11,7 +11,9 @@
  * A PINGRESP Packet is sent by the Server to the Client in response
  * to a PINGREQ Packet. It indicates that the Server is alive.
  */
-class PingResponse extends ControlPacket {
+class PingResponse extends ControlPacket
+{
+    const EVENT = 'PING_RESPONSE';
 
     public static function getControlPacketType()
     {

src/packet/Publish.php

@@ -13,7 +13,9 @@
  * A PUBLISH Control Packet is sent from a Client to a Server or from
  * Server to a Client to transport an Application Message.
  */
-class Publish extends ControlPacket {
+class Publish extends ControlPacket
+{
+    const EVENT = 'PUBLISH';
 
     protected $messageId;
 

src/packet/PublishReceived.php

@@ -11,7 +11,9 @@
  * A PUBREC Packet is the response to a PUBLISH Packet with QoS 2.
  * It is the second packet of the QoS 2 protocol exchange.
  */
-class PublishReceived extends ControlPacket {
+class PublishReceived extends ControlPacket
+{
+    const EVENT = 'PUBLISH_RECEIVED';
 
     public static function getControlPacketType()
     {

src/packet/PublishRelease.php

@@ -11,7 +11,9 @@
  * A PUBREL Packet is the response to a PUBREC Packet.
  * It is the third packet of the QoS 2 protocol exchange.
  */
-class PublishRelease extends ControlPacket {
+class PublishRelease extends ControlPacket
+{
+    const EVENT = 'PUBLISH_RELEASE';
 
     public static function getControlPacketType()
     {

src/packet/SubscribeAck.php

@@ -11,7 +11,9 @@
  * A SUBACK Packet is sent by the Server to the Client to confirm receipt
  * and processing of a SUBSCRIBE Packet.
  */
-class SubscribeAck extends ControlPacket {
+class SubscribeAck extends ControlPacket
+{
+    const EVENT = 'SUBSCRIBE_ACK';
 
     public static function getControlPacketType()
     {

src/packet/UnsubscribeAck.php

@@ -11,7 +11,9 @@
  * The UNSUBACK Packet is sent by the Server to the Client to confirm
  * receipt of an UNSUBSCRIBE Packet.
  */
-class UnsubscribeAck extends ControlPacket {
+class UnsubscribeAck extends ControlPacket
+{
+    const EVENT = 'UNSUBSCRIBE_ACK';
 
     public static function getControlPacketType()
     {