7032442828

Committed 29 Nov 2023 11:03AM UTC coverage: 81.707% (-1.0%) from 82.683%

Build # 7032442828

Build Type

push

github

Committed by

web-flow

Commit Message

Update README.md

Run Details

335 of 410 relevant lines covered (81.71%)

10.2 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

77.78

/src/Queues/ServiceBusQueue.php

<?php

declare(strict_types=1);

namespace Daycry\Queues\Queues;

use DateTime;
use DateTimeZone;
use Daycry\Queues\Exceptions\QueueException;
use Daycry\Queues\Interfaces\QueueInterface;
use Daycry\Queues\Interfaces\WorkerInterface;
use Daycry\Queues\Libraries\ServiceBusHeaders as LibrariesServiceBusHeaders;
use GuzzleHttp\Client;
use GuzzleHttp\Psr7\Request;
use GuzzleHttp\Exception\RequestException;
use Daycry\Queues\Job as QueuesJob;

class ServiceBusQueue extends BaseQueue implements QueueInterface, WorkerInterface
{
    private string $url;
    private LibrariesServiceBusHeaders $serviceBusHeaders;
    private ?array $job = null;

    public function __construct()
    {
        $config = service('settings')->get('Queue.serviceBus');
        $this->url = $config['url'];
        $this->serviceBusHeaders = (new LibrariesServiceBusHeaders())->generateMessageId()->generateSasToken($this->url, $config['issuer'], $config['secret']);
    }
    
    public function enqueue(object $data, string $queue = 'default')
    {
        $this->calculateDelay($data);

        if($this->getDelay() > 0)
        {
            $datetime = new DateTime($data->schedule->date, new DateTimeZone($data->schedule->timezone));
            $this->serviceBusHeaders->schedule($datetime);
        }
        
        $response = $this->request($queue . '/messages', 'post', $data, $this->serviceBusHeaders->getHeaders());

        return $this->serviceBusHeaders->getMessageId();
    }

    public function setLabel(string $label)
    {
        $this->serviceBusHeaders->setLabel($label);
    }

    public function watch(string $queue)
    {
        $url = $queue . '/messages/head';

        $response = $this->request($url, 'delete', [], $this->serviceBusHeaders->getHeaders());

        if($response) {
            $this->job = $response;

            return $this->job;
        }

        return false;
    }

    public function removeJob(QueuesJob $job, bool $recreate = false): bool
    {
        if($recreate === true)
        {
            $job->addAttempt();
            $job->enqueue($job->getQueue());
        }

        $this->job = null;

        return true;
    }

    public function getDataJob()
    {
        return $this->job['body'];
    }

    protected function request(string $url, string $method, object|array $body = [], array $headers = []): array
    {
        if(strpos($url, 'https://') !== 0) {
            $url = $this->url . $url;
        }

        $client  = new Client(['verify' => false]);
        $request = new Request($method, $url, $headers, \json_encode($body));

        try {
            $response = $client->send($request, ['timeout' => 10]);

            if( $response )
            {
                return [
                    'body' => json_decode($response->getBody()->getContents()),
                    'status' => $response->getStatusCode(),
                    'headers' => $response->getHeaders()
                ];
            }

            return null;

        } catch (RequestException $e) {
            throw QueueException::forInvalidConnection($e->getMessage());
        }
    }
}

1	<?php
2
3	declare(strict_types=1);
4
5	namespace Daycry\Queues\Queues;
6
7	use DateTime;
8	use DateTimeZone;
9	use Daycry\Queues\Exceptions\QueueException;
10	use Daycry\Queues\Interfaces\QueueInterface;
11	use Daycry\Queues\Interfaces\WorkerInterface;
12	use Daycry\Queues\Libraries\ServiceBusHeaders as LibrariesServiceBusHeaders;
13	use GuzzleHttp\Client;
14	use GuzzleHttp\Psr7\Request;
15	use GuzzleHttp\Exception\RequestException;
16	use Daycry\Queues\Job as QueuesJob;
17
18	class ServiceBusQueue extends BaseQueue implements QueueInterface, WorkerInterface
19	{
20	private string $url;
21	private LibrariesServiceBusHeaders $serviceBusHeaders;
22	private ?array $job = null;
23
24	public function __construct()
25	{
26	$config = service('settings')->get('Queue.serviceBus');	11✔
27	$this->url = $config['url'];	11✔
28	$this->serviceBusHeaders = (new LibrariesServiceBusHeaders())->generateMessageId()->generateSasToken($this->url, $config['issuer'], $config['secret']);	11✔
29	}
30
31	public function enqueue(object $data, string $queue = 'default')
32	{
33	$this->calculateDelay($data);	6✔
34
35	if($this->getDelay() > 0)	6✔
36	{
37	$datetime = new DateTime($data->schedule->date, new DateTimeZone($data->schedule->timezone));	×
38	$this->serviceBusHeaders->schedule($datetime);	×
39	}
40
41	$response = $this->request($queue . '/messages', 'post', $data, $this->serviceBusHeaders->getHeaders());	6✔
42
43	return $this->serviceBusHeaders->getMessageId();	6✔
44	}
45
46	public function setLabel(string $label)
47	{
48	$this->serviceBusHeaders->setLabel($label);	1✔
49	}
50
51	public function watch(string $queue)
52	{
53	$url = $queue . '/messages/head';	5✔
54
55	$response = $this->request($url, 'delete', [], $this->serviceBusHeaders->getHeaders());	5✔
56
57	if($response) {	5✔
58	$this->job = $response;	5✔
59
60	return $this->job;	5✔
61	}
62
63	return false;	×
64	}
65
66	public function removeJob(QueuesJob $job, bool $recreate = false): bool
67	{
68	if($recreate === true)	5✔
69	{
70	$job->addAttempt();	×
71	$job->enqueue($job->getQueue());	×
72	}
73
74	$this->job = null;	5✔
75
76	return true;	5✔
77	}
78
79	public function getDataJob()
80	{
81	return $this->job['body'];	5✔
82	}
83
84	protected function request(string $url, string $method, object\|array $body = [], array $headers = []): array
85	{
86	if(strpos($url, 'https://') !== 0) {	11✔
87	$url = $this->url . $url;	11✔
88	}
89
90	$client = new Client(['verify' => false]);	11✔
91	$request = new Request($method, $url, $headers, \json_encode($body));	11✔
92
93	try {
94	$response = $client->send($request, ['timeout' => 10]);	11✔
95
96	if( $response )	11✔
97	{
98	return [	11✔
99	'body' => json_decode($response->getBody()->getContents()),	11✔
100	'status' => $response->getStatusCode(),	11✔
101	'headers' => $response->getHeaders()	11✔
102	];	11✔
103	}
104
105	return null;	×
106
107	} catch (RequestException $e) {	×
108	throw QueueException::forInvalidConnection($e->getMessage());	×
109	}
110	}
111	}

daycry / queues / 7032442828

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous