Update Pheanstalk class

Author: vcorre

src/Pheanstalk.php

@@ -264,27 +264,11 @@ private function _dispatch($command)
     public function create(Workflow $workflow, $force = false): Workflow
     {
         try {
-            $tubes = [];
-            /** @var Job $job */
-            foreach ($workflow->getJobs() as $job) {
-                /** @var Task $task */
-                foreach ($job->getTasks() as $task) {
-                    $tubes = array_merge($tubes, [$task->getQueue()]);
-                }
-            }
-            foreach ($tubes as $tube) {
-                if (!$this->getCurrentClass()->tubeExists($tube)) {
-                    $this->getCurrentClass()->createTube(new Tube($tube, 1));
-                };
-            }
+            $this->checkAndCreateTubes($workflow);
             $workflow = $this->_dispatch(new Command\CreateCommand($workflow));
         } catch (ServerDuplicateEntryException $e) {
             if ($force) {
-                $workflows = $this->_dispatch(new Command\ListWorkflowsCommand());
-                $workflowToDelete = $workflows->filter(function(Workflow $listedWorkflow) use ($workflow) {
-                    return $listedWorkflow->getName() === $workflow->getName()
-                        && $listedWorkflow->getGroup() === $workflow->getGroup();
-                })->first();
+                $workflowToDelete = $this->findWorkflow($workflow);
                 $this->getCurrentClass()->delete($workflowToDelete);
 
                 return $this->getCurrentClass()->create($workflow);
@@ -295,6 +279,43 @@ public function create(Workflow $workflow, $force = false): Workflow
         return $workflow;
     }
 
+    /**
+     * @param Workflow $workflow
+     *
+     * @return Workflow|bool
+     * @throws Exception\ClientException
+     */
+    public function findWorkflow(Workflow $workflow)
+    {
+        $workflows = $this->_dispatch(new Command\ListWorkflowsCommand());
+        return $workflows->filter(function(Workflow $listedWorkflow) use ($workflow) {
+            return $listedWorkflow->getName() === $workflow->getName()
+                && $listedWorkflow->getGroup() === $workflow->getGroup();
+        })->first();
+    }
+
+    /**
+     * @param Workflow $workflow
+     *
+     * @throws Exception\ClientException
+     */
+    public function checkAndCreateTubes(Workflow $workflow)
+    {
+        $tubes = [];
+        /** @var Job $job */
+        foreach ($workflow->getJobs() as $job) {
+            /** @var Task $task */
+            foreach ($job->getTasks() as $task) {
+                $tubes = array_merge($tubes, [$task->getQueue()]);
+            }
+        }
+        foreach ($tubes as $tube) {
+            if (!$this->getCurrentClass()->tubeExists($tube)) {
+                $this->getCurrentClass()->createTube(new Tube($tube, 1));
+            };
+        }
+    }
+
     /**
      * {@inheritdoc}
      */

src/PheanstalkInterface.php

@@ -186,6 +186,7 @@ public function stats();
      * @param Workflow  $data     The workflow to create
      * @param null|bool (optional) $force Will erase already existent old workflow if already exists
      *
+     * @throws Exception\ClientException
      * @return Workflow The newly created workflow
      */
     public function create(Workflow $data, $force = false): Workflow;