PASTEBINGO  API TOOLS FAQ DEALS
Search...
pasteSIGN IN SIGN UP
Public Pastes
Untitled
PHP | 0 sec ago
Untitled
2 sec ago
Untitled
2 sec ago
Untitled
4 sec ago
Untitled
5 sec ago
Untitled
6 sec ago
Untitled
Python | 6 sec ago
Il Piatto Piange TAS
12 sec ago

Untitled
A GUEST  MAR 26TH, 2020  0  NEVER
NOTE: Your guest paste has been posted. If you sign up for a free account, you can edit and delete your pastes!
rawdownloadcloneembedreportprintPHP 1.41 KB
<?php

class Message {
    public function __construct($job) {
        //@todo: set data accordingly
    }

    public $timestamp = 123;
    public $data      = [];
}

class Subscriber {
    private $run = true;

    public function __construct() {
        declare(ticks = 1);
        pcntl_signal(SIGTERM, function () {
            $this->run = false;
        });
    }

    public function Subscribe(Callable $callback): void {
        while ($this->run) {
            $jobs = $this->selectJobs();
            foreach ($jobs as $job) {
                $Message = new Message($job);
                if ($callback($Message)) {
                    $this->markJobDone($job['id']);
                } else {
                    $this->markJobFailed($job['id']);
                }
            }
        }
    }

    private function selectJobs(): array {
        //@todo: select && lock jobs from musql
        return [];
    }

    private function markJobDone(int $job_id) : void {
        //@todo: delete from mysql
    }

    private function markJobFailed(int $job_id) : void {
        //@todo: log error, send job TO THE END OF THE QUEUE (otherwise failed jobs may stop queue to being processed)
        //@todo: increase fail_counter. if fail_counter > threshold_value, do not select this job anymore => need manual processing
    }
}

class Worker {
    public function run(): void {
        (new Subscriber())->Subscribe([$this, 'processMessage']);
    }

    private function processMessage(Message $Message): bool {
        var_dump($Message->data);
        // process message here
        return true;
    }
}

(new Worker)->run();
RAW Paste Data
<?php

class Message {
    public function __construct($job) {
        //@todo: set data accordingly
    }

    public $timestamp = 123;
    public $data      = [];
}

class Subscriber {
    private $run = true;

    public function __construct() {
        declare(ticks = 1);
        pcntl_signal(SIGTERM, function () {
            $this->run = false;
        });
    }

    public function Subscribe(Callable $callback): void {
        while ($this->run) {
            $jobs = $this->selectJobs();
            foreach ($jobs as $job) {
                $Message = new Message($job);
                if ($callback($Message)) {
                    $this->markJobDone($job['id']);
                } else {
                    $this->markJobFailed($job['id']);
                }
                if (!$this->run) break;
            }
        }
    }

    private function selectJobs(): array {
        //@todo: select && lock jobs from musql
        return [];
    }

    private function markJobDone(int $job_id) : void {
        //@todo: delete from mysql
    }

    private function markJobFailed(int $job_id) : void {
        //@todo: log error, send job TO THE END OF THE QUEUE (otherwise failed jobs may stop queue to being processed)
        //@todo: increase fail_counter. if fail_counter > threshold_value, do not select this job anymore => need manual processing
    }
}

class Worker {
    public function run(): void {
        (new Subscriber())->Subscribe([$this, 'processMessage']);
    }

    private function processMessage(Message $Message): bool {
        var_dump($Message->data);
        // process message here
        return true;
    }
}

(new Worker)->run();
We use cookies for various purposes including analytics. By continuing to use Pastebin, you agree to our use of cookies as described in the Cookies Policy. OK, I Understand

create new paste  /  dealsnew!  /  syntax languages  /  archive  /  faq  /  tools  /  night mode  /  api  /  scraping api
privacy statement  /  cookies policy  /  terms of service  /  security disclosure  /  dmca  /  contact

By using Pastebin.com you agree to our cookies policy to enhance your experience.
Site design & logo © 2020 Pastebin; user contributions (pastes) licensed under cc by-sa 3.0 -- FavPNG -- Dedicated Server Hosting by Steadfast