Consumer.php

<?php
namespace Magenest\Salesforce\Model\Queue;

use Magento\Framework\Exception\LocalizedException;
use Magenest\Salesforce\Model\Queue;
use Magento\Framework\Filesystem;
use Magento\Framework\Notification\NotifierInterface;

/**
 * Class Consumer
 * @package Magenest\Salesforce\Model\Queue
 */
class Consumer
{
    /* @var \Psr\Log\LoggerInterface  /
    protected $_logger;

    /* @var \Magento\Sales\Api\OrderRepositoryInterface  /
    protected $_orderRepository;

    /* @var NotifierInterface  /
    protected $_notifier;

    /* @var \Magenest\Salesforce\Model\QueueFactory  /
    protected $_queueFactory;

    /**
     * @var Filesystem
     */
    private $filesystem;

    /* @var \Magento\Framework\Serialize\Serializer\Json  /
    protected $_json;

    /* @var string  /
    protected $_type = null;

    /**
     * Consumer constructor.
     *
     * @param \Psr\Log\LoggerInterface $logger
     * @param \Magento\Sales\Api\OrderRepositoryInterface $orderRepository
     * @param NotifierInterface $notifier
     * @param \Magenest\Salesforce\Model\QueueFactory $queueFactory
     * @param Filesystem $filesystem
     * @param \Magento\Framework\Serialize\Serializer\Json $json
     */
    public function __construct(
        \Psr\Log\LoggerInterface $logger,
        \Magento\Sales\Api\OrderRepositoryInterface $orderRepository,
        NotifierInterface $notifier,
        \Magenest\Salesforce\Model\QueueFactory $queueFactory,
        Filesystem $filesystem,
        \Magento\Framework\Serialize\Serializer\Json $json
    ){
        $this->_logger = $logger;
        $this->_orderRepository = $orderRepository;
        $this->_notifier = $notifier;
        $this->_queueFactory = $queueFactory;
        $this->filesystem = $filesystem;
        $this->_json = $json;
    }

    /**
     * @param string $orderSearchResult
     */
    public function process($orderSearchResult)
    {
        try{
            $this->execute($orderSearchResult);
            $this->_notifier->addMajor(
                __('Your queue are ready'),
                __('You can check your orders at Salesforce Queue page')
            );
        }catch (\Exception $e){
            $errorCode = $e->getCode();
            $message = __('Sorry, something went wrong during add order to queue. Please see log for details.');
            $this->_notifier->addCritical(
                $errorCode,
                $message
            );
            $this->_logger->critical($errorCode .": ". $message);
        }
    }

    /**
     * @param $orderItems
     *
     * @throws LocalizedException
     */
    private function execute($orderItems)
    {
        $orderCollectionArr = [];
        /* @var \Magenest\Salesforce\Model\Queue $queue /
        $queue = $this->_queueFactory->create();
        $orderItems = $this->_json->unserialize($orderItems);
        if(is_array($orderItems)){
            foreach ($orderItems as $type => $orderId) {
                $this->_type = $type;
                $orderCollectionArr[] = [
                    'type' => $type,
                    'entity_id' => $orderId,
                    'priority' => 1,
                ];
            }
            $queue->deleteQueueByType($this->_type);
            $queue->enqueueMultiRecords($orderCollectionArr);
        }
    }
}