9935584685

Committed 15 Jul 2024 07:44AM UTC coverage: 47.259% (-18.3%) from 65.524%

Build Type

push

github

Committed by darkdarin

Commit Message

fix: tests

Run Details

4 of 4 new or added lines in 2 files covered. (100.0%)

431 existing lines in 40 files now uncovered.

612 of 1295 relevant lines covered (47.26%)

2.53 hits per line

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

0.0

/src/Core/GarbageCollector.php

<?php

namespace Tochka\Promises\Core;

use Carbon\Carbon;
use Illuminate\Support\Collection;
use Illuminate\Support\Facades\DB;
use Tochka\Promises\Core\Support\DaemonWorker;

class GarbageCollector implements GarbageCollectorInterface
{
    use DaemonWorker;

    private int $deleteOlderThen;
    /** @var array<string> */
    private array $states;

    private string $promisesTable;
    private string $promiseJobsTable;
    private string $promiseEventsTable;
    private int $promiseChunkSize;
    private int $jobsChunkSize;

    public function __construct(
        int $sleepTime,
        int $deleteOlderThen,
        array $states,
        string $promisesTable,
        string $promiseJobsTable,
        string $promiseEventsTable,
        int $promiseChunkSize = 100,
        int $jobsChunkSize = 500,
    ) {
        $this->sleepTime = $sleepTime;
        $this->deleteOlderThen = $deleteOlderThen;
        $this->states = $states;

        $this->promisesTable = $promisesTable;
        $this->promiseJobsTable = $promiseJobsTable;
        $this->promiseEventsTable = $promiseEventsTable;

        $this->promiseChunkSize = $promiseChunkSize;
        $this->jobsChunkSize = $jobsChunkSize;

        $this->lastIteration = Carbon::minValue();
    }

    /**
     * @param null|callable(): bool $shouldQuitCallback
     * @param null|callable(): bool $shouldPausedCallback
     */
    public function handle(?callable $shouldQuitCallback = null, ?callable $shouldPausedCallback = null): void
    {
        $this->daemon(function () use ($shouldQuitCallback, $shouldPausedCallback) {
            $this->clean($shouldQuitCallback, $shouldPausedCallback);
        }, $shouldQuitCallback, $shouldPausedCallback);
    }

    /**
     * @param null|callable(): bool $shouldQuitCallback
     * @param null|callable(): bool $shouldPausedCallback
     */
    public function clean(?callable $shouldQuitCallback = null, ?callable $shouldPausedCallback = null): void
    {
        if ($shouldQuitCallback === null) {
            $shouldQuitCallback = fn(): bool => false;
        }

        if ($shouldPausedCallback === null) {
            $shouldPausedCallback = fn(): bool => false;
        }

        while (!$shouldQuitCallback() && !$shouldPausedCallback()) {
            $promises = DB::table($this->promisesTable)
                ->select([$this->promiseColumn('id')])
                ->leftJoin(
                    $this->promiseJobsTable,
                    $this->promiseJobsColumn('id'),
                    '=',
                    $this->promiseColumn('parent_job_id'),
                )
                ->whereIn($this->promiseColumn('state'), $this->states)
                ->where($this->promiseColumn('updated_at'), '<', Carbon::now()->subSeconds($this->deleteOlderThen))
                ->whereNull($this->promiseJobsColumn('id'))
                ->limit($this->promiseChunkSize)
                ->pluck('id')
                ->all();

            if (empty($promises)) {
                return;
            }

            $this->handlePromiseChunks($promises, $shouldQuitCallback, $shouldPausedCallback);

            $this->sleep(0.05);
        }
    }

    /**
     * @param array<int, int> $promiseIds
     * @param callable(): bool $shouldQuitCallback
     * @param callable(): bool $shouldPausedCallback
     */
    private function handlePromiseChunks(array $promiseIds, callable $shouldQuitCallback, callable $shouldPausedCallback): void
    {
        if ($shouldQuitCallback() || $shouldPausedCallback()) {
            return;
        }

        DB::table($this->promiseJobsTable)
            ->select(['id'])
            ->whereIn('promise_id', $promiseIds)
            ->chunkById(
                $this->jobsChunkSize,
                $this->handleJobsChunks(...),
            );

        DB::table($this->promisesTable)->whereIn('id', $promiseIds)->delete();
    }

    /**
     * @param Collection<int, object{id: string}> $jobs
     * @return bool
     */
    private function handleJobsChunks(Collection $jobs): bool
    {
        $jobsIds = $jobs->pluck('id')->all();

        DB::table($this->promiseEventsTable)->whereIn('job_id', $jobsIds)->delete();
        DB::table($this->promiseJobsTable)->whereIn('id', $jobsIds)->delete();

        return true;
    }

    private function promiseColumn(string $columnName): string
    {
        return $this->promisesTable . '.' . $columnName;
    }

    private function promiseJobsColumn(string $columnName): string
    {
        return $this->promiseJobsTable . '.' . $columnName;
    }
}

1	<?php
2
3	namespace Tochka\Promises\Core;
4
5	use Carbon\Carbon;
6	use Illuminate\Support\Collection;
7	use Illuminate\Support\Facades\DB;
8	use Tochka\Promises\Core\Support\DaemonWorker;
9
10	class GarbageCollector implements GarbageCollectorInterface
11	{
12	use DaemonWorker;
13
14	private int $deleteOlderThen;
15	/** @var array<string> */
16	private array $states;
17
18	private string $promisesTable;
19	private string $promiseJobsTable;
20	private string $promiseEventsTable;
21	private int $promiseChunkSize;
22	private int $jobsChunkSize;
23
24	public function __construct(
25	int $sleepTime,
26	int $deleteOlderThen,
27	array $states,
28	string $promisesTable,
29	string $promiseJobsTable,
30	string $promiseEventsTable,
31	int $promiseChunkSize = 100,
32	int $jobsChunkSize = 500,
33	) {
34	$this->sleepTime = $sleepTime;	×
35	$this->deleteOlderThen = $deleteOlderThen;	×
36	$this->states = $states;	×
37
38	$this->promisesTable = $promisesTable;	×
39	$this->promiseJobsTable = $promiseJobsTable;	×
40	$this->promiseEventsTable = $promiseEventsTable;	×
41
42	$this->promiseChunkSize = $promiseChunkSize;	×
43	$this->jobsChunkSize = $jobsChunkSize;	×
44
45	$this->lastIteration = Carbon::minValue();	×
46	}
47
48	/**
49	* @param null\|callable(): bool $shouldQuitCallback
50	* @param null\|callable(): bool $shouldPausedCallback
51	*/
52	public function handle(?callable $shouldQuitCallback = null, ?callable $shouldPausedCallback = null): void
53	{
UNCOV 54	$this->daemon(function () use ($shouldQuitCallback, $shouldPausedCallback) {	×
UNCOV 55	$this->clean($shouldQuitCallback, $shouldPausedCallback);	×
UNCOV 56	}, $shouldQuitCallback, $shouldPausedCallback);	×
57	}
58
59	/**
60	* @param null\|callable(): bool $shouldQuitCallback
61	* @param null\|callable(): bool $shouldPausedCallback
62	*/
63	public function clean(?callable $shouldQuitCallback = null, ?callable $shouldPausedCallback = null): void
64	{
65	if ($shouldQuitCallback === null) {	×
66	$shouldQuitCallback = fn(): bool => false;	×
67	}
68
69	if ($shouldPausedCallback === null) {	×
70	$shouldPausedCallback = fn(): bool => false;	×
71	}
72
73	while (!$shouldQuitCallback() && !$shouldPausedCallback()) {	×
74	$promises = DB::table($this->promisesTable)	×
UNCOV 75	->select([$this->promiseColumn('id')])	×
UNCOV 76	->leftJoin(	×
77	$this->promiseJobsTable,	×
UNCOV 78	$this->promiseJobsColumn('id'),	×
79	'=',	×
UNCOV 80	$this->promiseColumn('parent_job_id'),	×
UNCOV 81	)	×
UNCOV 82	->whereIn($this->promiseColumn('state'), $this->states)	×
UNCOV 83	->where($this->promiseColumn('updated_at'), '<', Carbon::now()->subSeconds($this->deleteOlderThen))	×
UNCOV 84	->whereNull($this->promiseJobsColumn('id'))	×
85	->limit($this->promiseChunkSize)	×
UNCOV 86	->pluck('id')	×
UNCOV 87	->all();	×
88
UNCOV 89	if (empty($promises)) {	×
90	return;	×
91	}
92
UNCOV 93	$this->handlePromiseChunks($promises, $shouldQuitCallback, $shouldPausedCallback);	×
94
UNCOV 95	$this->sleep(0.05);	×
96	}
97	}
98
99	/**
100	* @param array<int, int> $promiseIds
101	* @param callable(): bool $shouldQuitCallback
102	* @param callable(): bool $shouldPausedCallback
103	*/
104	private function handlePromiseChunks(array $promiseIds, callable $shouldQuitCallback, callable $shouldPausedCallback): void
105	{
106	if ($shouldQuitCallback() \|\| $shouldPausedCallback()) {	×
107	return;	×
108	}
109
UNCOV 110	DB::table($this->promiseJobsTable)	×
111	->select(['id'])	×
UNCOV 112	->whereIn('promise_id', $promiseIds)	×
113	->chunkById(	×
UNCOV 114	$this->jobsChunkSize,	×
UNCOV 115	$this->handleJobsChunks(...),	×
UNCOV 116	);	×
117
UNCOV 118	DB::table($this->promisesTable)->whereIn('id', $promiseIds)->delete();	×
119	}
120
121	/**
122	* @param Collection<int, object{id: string}> $jobs
123	* @return bool
124	*/
125	private function handleJobsChunks(Collection $jobs): bool
126	{
UNCOV 127	$jobsIds = $jobs->pluck('id')->all();	×
128
129	DB::table($this->promiseEventsTable)->whereIn('job_id', $jobsIds)->delete();	×
UNCOV 130	DB::table($this->promiseJobsTable)->whereIn('id', $jobsIds)->delete();	×
131
UNCOV 132	return true;	×
133	}
134
135	private function promiseColumn(string $columnName): string
136	{
UNCOV 137	return $this->promisesTable . '.' . $columnName;	×
138	}
139
140	private function promiseJobsColumn(string $columnName): string
141	{
UNCOV 142	return $this->promiseJobsTable . '.' . $columnName;	×
143	}
144	}

tochka-developers / queue-promises / 9935584685

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