igusev/phpnsq

PHP Client for nsq

dev-master 2021-01-08 06:14 UTC

Requires

• bramus/monolog-colored-line-formatter: ^2.0
• monolog/monolog: ^2.2
• react/event-loop: ^v0.4|^v0.5|^v1.1

MIT

• stackerzzq

• queue
• nsq

README

Install

composer require okstuff/phpnsq

Subscribe one topic and channel

php examples/console phpnsq:sub <topic> <channel>

Notice

Before try this, you should install nsq by yourself.

Examples

1. Publish

<?php

require __DIR__ . "/../vendor/autoload.php";

use OkStuff\PhpNsq\Message\Message;
use OkStuff\PhpNsq\PhpNsq;

$config = require __DIR__ . '/../src/config/phpnsq.php';
$phpnsq = new PhpNsq($config);

//normal publish
$message = new Message();
$message->setBody("Hello nsq.");
$phpnsq->setTopic("sample_topic")->publish(json_encode($message));

//defered publish
$message = [
    "title"   => "hello",
    "content" => "this is a nsq php client.",
];
$phpnsq->setTopic("sample_topic")->publishDefer(json_encode($message), 10);

//multiple publish
$messages = [
    "Hello, I am nsq client",
    "There are so many libraries developed by PHP",
    "Oh, no, PHP is not so good and slowly",
];
$phpnsq->setTopic("sample_topic")->publishMulti(...$messages);

1. Subscribe

<?php

use OkStuff\PhpNsq\Message\Message;
use OkStuff\PhpNsq\Command\Base;
use Symfony\Component\Console\Input\InputArgument;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Output\OutputInterface;

class Subscribe extends Base
{
    CONST COMMAND_NAME = 'phpnsq:sub';

    public function configure()
    {
        $this->setName(self::COMMAND_NAME)
            ->addArgument("topic", InputArgument::REQUIRED, "The topic you want to subscribe")
            ->addArgument("channel", InputArgument::REQUIRED, "The channel you want to subscribe")
            ->setDescription('subscribe new notification.')
            ->setHelp("This command allows you to subscribe notifications...");
    }

    public function execute(InputInterface $input, OutputInterface $output)
    {
        $phpnsq = self::$phpnsq;
        $phpnsq->setTopic($input->getArgument("topic"))
            ->setChannel($input->getArgument("channel"))
            ->subscribe($this, function (Message $message) use ($phpnsq, $output) {
                $phpnsq->getLogger()->info("READ", $message);
            });
        $this->addPeriodicTimer(5, function () use ($output) {
            $memory    = memory_get_usage() / 1024;
            $formatted = number_format($memory, 3) . 'K';
            $output->writeln("############ Current memory usage: {$formatted} ############");
        });
        $this->runLoop();
    }
}

1. Console

#!/usr/bin/env php
<?php

require __DIR__.'/../vendor/autoload.php';

use Symfony\Component\Console\Application;
//use OkStuff\PhpNsq\Command\Subscribe;

require __DIR__.'/subscribe.php';

$application = new Application();

$config = require __DIR__.'/../src/config/phpnsq.php';

$application->add(new Subscribe($config));

$application->run();