mgdigital / BusQue

    public function queueCommand(string $queueName, string $id, string $serialized)

    {

        if (!$this->storeCommandAndCheckIfIdReserved($queueName, $id, $serialized)) {

            $this->client->pipeline(function (ClientContextInterface $client) use ($queueName, $id) {

                self::cReserveCommandId($client, $queueName, $id);

                self::cUpdateCommandStatus($client, $queueName, $id, self::STATUS_QUEUED);

                $client->lpush(":{$queueName}:queue", [ $id ]);

            });

        }

    }

            return $this->awaitCommand($queueName, $timeout);

        }

        /* @var $id string */

        list(, $serialized) = $this->client->pipeline(function (ClientContextInterface $client) use ($queueName, $id) {

            self::cUpdateCommandStatus($client, $queueName, $id, self::STATUS_IN_PROGRESS);

            self::cRetrieveCommand($client, $queueName, $id);

            self::cReleaseReservedCommandIds($client, $queueName, [ $id ]);

        });

        return new ReceivedCommand($queueName, $id, $serialized);

    }