Some updates

.idea/php.xml

@@ -15,8 +15,11 @@
       <path value="/etc/ncc" />
       <path value="/var/ncc/packages/net.nosial.optslib=1.0.0" />
       <path value="/var/ncc/packages/net.nosial.loglib=1.0.0" />
-      <path value="/var/ncc/packages/com.opis.closure=3.6.3" />
+      <path value="/var/ncc/packages/com.phpseclib.phpseclib=3.0.18" />
       <path value="/var/ncc/packages/com.php_amqplib.php_amqplib=3.5.1" />
+      <path value="/var/ncc/packages/com.paragonie.random_compat=9.99.100" />
+      <path value="/var/ncc/packages/com.paragonie.constant_time_encoding=2.6.3" />
+      <path value="/var/ncc/packages/com.opis.closure=3.6.3" />
     </include_path>
   </component>
   <component name="PhpProjectSharedConfiguration" php_language_level="8.1" />

README.md

@@ -44,7 +44,7 @@ TamerLib\Tamer::init(\TamerLib\Abstracts\ProtocolType::Gearman, [
 ## Supported Protocols
 
  * [x] Gearman
- * [x] RabbitMQ
+ * [ ] RabbitMQ (Work in progress)
  * [ ] Redis
 
 # License

src/TamerLib/Objects/Job.php

@@ -92,6 +92,7 @@
         public function toArray(): array
         {
             return [
+                'type' => 'tamer_job',
                 'id' => $this->id,
                 'name' => $this->name,
                 'data' => ($this->closure ? serialize(new SerializableClosure($this->data)) : $this->data),

src/TamerLib/Objects/JobResults.php

@@ -78,6 +78,7 @@
         public function toArray(): array
         {
             return [
+                'type' => 'tamer_job_results',
                 'id' => $this->id,
                 'data' => $this->data,
                 'status' => $this->status

src/TamerLib/Protocols/RabbitMq/Client.php

@@ -7,7 +7,9 @@
     use Closure;
     use Exception;
     use PhpAmqpLib\Message\AMQPMessage;
+    use TamerLib\Abstracts\ObjectType;
     use TamerLib\Classes\Functions;
+    use TamerLib\Classes\Validate;
     use TamerLib\Exceptions\ConnectionException;
     use TamerLib\Interfaces\ClientProtocolInterface;
     use TamerLib\Objects\Job;
@@ -326,6 +328,7 @@
             $this->preformAutoreconf();
             $correlationIds = [];
             $connection =  $this->connections[array_rand($this->connections)];
+
             if($this->automatic_reconnect)
                 $connection->preformAutoreconf();
 
@@ -348,12 +351,32 @@
             // Register callback for each task
             $callback = function($msg) use (&$correlationIds, $connection)
             {
+                var_dump(Validate::getObjectType(msgpack_unpack($msg->body)));
+                if(Validate::getObjectType(msgpack_unpack($msg->body)) !== ObjectType::JobResults)
+                {
+                    $connection->getChannel()->basic_nack($msg->delivery_info['delivery_tag'], false, true);
+                    return;
+                }
+
                 $job_result = JobResults::fromArray(msgpack_unpack($msg->body));
                 $task = $this->getTaskById($job_result->getId());
 
+                if($task == null)
+                {
+                    $connection->getChannel()->basic_nack($msg->delivery_info['delivery_tag'], false, true);
+                    return;
+                }
+
                 try
                 {
-                    $task->runCallback($job_result);
+                    if($task->isClosure())
+                    {
+                        $task->runCallback($job_result->getData());
+                    }
+                    else
+                    {
+                        $task->runCallback($job_result);
+                    }
                 }
                 catch(Exception $e)
                 {

src/TamerLib/Protocols/RabbitMq/Worker.php

@@ -8,6 +8,8 @@
     use LogLib\Log;
     use PhpAmqpLib\Message\AMQPMessage;
     use TamerLib\Abstracts\JobStatus;
+    use TamerLib\Abstracts\ObjectType;
+    use TamerLib\Classes\Validate;
     use TamerLib\Exceptions\ConnectionException;
     use TamerLib\Interfaces\WorkerProtocolInterface;
     use TamerLib\Objects\Job;
@@ -69,6 +71,7 @@
         public function __construct(?string $username = null, ?string $password = null)
         {
             $this->defined_servers = [];
+            $this->connections = [];
             $this->functions = [];
             $this->automatic_reconnect = true;
             $this->username = $username;
@@ -282,10 +285,17 @@
                 return;
 
             // Select a random connection
-            $connection =  $this->connections[array_rand($this->connections)];
+            $connection = $this->connections[array_rand($this->connections)];
 
             $callback = function($message) use ($throw_errors, $connection)
             {
+                var_dump(Validate::getObjectType(msgpack_unpack($message->body)));
+                if(Validate::getObjectType(msgpack_unpack($message->body)) !== ObjectType::Job)
+                {
+                    $connection->getChannel()->basic_nack($message->delivery_info['delivery_tag']);
+                    return;
+                }
+
                 $received_job = Job::fromArray(msgpack_unpack($message->body));
 
                 if($received_job->isClosure())
@@ -364,6 +374,7 @@
                     {
                         break;
                     }
+
                     $connection->getChannel()->wait();
                 }
             }

tests/tamer.php

@@ -8,7 +8,7 @@
 
     import('net.nosial.tamerlib', 'latest');
 
-    Tamer::init(ProtocolType::Gearman,
+    Tamer::init(ProtocolType::RabbitMQ,
         ['127.0.0.1:4730']
         //['127.0.0.1:5672'], 'guest', 'guest'
     );