IPub\WebSocketsZMQ\Consumer\Consumer

Complexity

Total Complexity

7

Size/Duplication

Total Lines	115
Duplicated Lines	9.57 %

Coupling/Cohesion

Components	1
Dependencies	1

Test Coverage

Coverage

15.38%

Importance

Changes

0

3 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	11	11	2
A	connect()	0	52	3
A	close()	0	4	2

<?php
/**
 * Consumer.php
 *
 * @copyright      More in license.md
 * @license        http://www.ipublikuj.eu
 * @author         Adam Kadlec http://www.ipublikuj.eu
 * @package        iPublikuj:WebSocketsZMQ!
 * @subpackage     Consumer
 * @since          1.0.0
 *
 * @date           28.02.17
 */

declare(strict_types = 1);

namespace IPub\WebSocketsZMQ\Consumer;

use Closure;
use ZMQ;
use ZMQSocketException;
use Throwable;

use Psr\Log;

use React;
use React\EventLoop;
use React\ZMQ as ReactZMQ;

use IPub;
use IPub\WebSocketsZMQ;

use IPub\WebSocketsWAMP\Application;
use IPub\WebSocketsWAMP\Entities;
use IPub\WebSocketsWAMP\PushMessages;
use IPub\WebSocketsWAMP\Serializers;

use IPub\WebSockets\Exceptions as WebSocketsExceptions;

/**
 * ZeroMQ consumer
 *
 * @package        iPublikuj:WebSocketsZMQ!
 * @subpackage     Consumer
 *
 * @author         Adam Kadlec <adam.kadlec@ipublikuj.eu>
 *
 * @method onSuccess(Consumer $consumer, $data = NULL)
 * @method onFail(Consumer $consumer, $data = NULL)
 */
final class Consumer extends PushMessages\Consumer
{
	/**
	 * @var Closure
	 */
	public $onSuccess = [];

	/**
	 * @var Closure
	 */
	public $onFail = [];

	/**
	 * @var WebSocketsZMQ\Configuration
	 */
	private $configuration;

	/**
	 * @var  Serializers\PushMessageSerializer
	 */
	private $serializer;

	/**
	 * @var ReactZMQ\SocketWrapper
	 */
	private $socket;

	/**
	 * @var  Log\LoggerInterface
	 */
	private $logger;

	/**
	 * @param WebSocketsZMQ\Configuration $configuration
	 * @param Serializers\PushMessageSerializer $serializer
	 * @param Log\LoggerInterface|NULL $logger
	 */
	public function __construct(

This method seems to be duplicated in your project.

		WebSocketsZMQ\Configuration $configuration,
		Serializers\PushMessageSerializer $serializer,
		Log\LoggerInterface $logger = NULL
	) {
		parent::__construct('zmq');

		$this->configuration = $configuration;
		$this->serializer = $serializer;
		$this->logger = $logger === NULL ? new Log\NullLogger : $logger;

It seems like $logger === NULL ? new \Psr\Log\NullLogger() : $logger can also be of type object<Psr\Log\NullLogger>. However, the property $logger is declared as type object<Psr\Log\LoggerInterface>. Maybe add an additional type check?

	}

	/**
	 * {@inheritdoc}
	 *
	 * @throws ZMQSocketException
	 */
	public function connect(EventLoop\LoopInterface $loop, Application\IApplication $application)
	{
		$context = new ReactZMQ\Context($loop);

		$this->socket = $context->getSocket(ZMQ::SOCKET_PULL);

		$this->logger->info(sprintf(
			'ZMQ transport listening on %s:%s',
			$this->configuration->getHost(),
			$this->configuration->getPort()
		));

		$this->socket->bind($this->configuration->getProtocol() . '://' . $this->configuration->getHost() . ':' . $this->configuration->getPort());

		$this->socket->on('message', function ($data) use ($application) {
			try {
				/** @var Entities\PushMessages\IMessage $message */
				$message = $this->serializer->deserialize($data);

				$application->handlePush($message, $this->getName());

Consider using $this->name. There is an issue with getName() and APC-enabled PHP versions.

				$this->onSuccess($this, $data);

			} catch (WebSocketsExceptions\TerminateException $ex) {

The class IPub\WebSockets\Exceptions\TerminateException does not exist. Did you forget a USE statement, or did you not list all dependencies?

				throw $ex;

			} catch (Throwable $ex) {
				$this->logger->error(
					'ZMQ socket failed to ack message', [
						'exception_message' => $ex->getMessage(),
						'file'              => $ex->getFile(),
						'line'              => $ex->getLine(),
						'message'           => $data,
					]
				);

				$this->onFail($this, $data);
			}
		});

		$this->socket->on('error', function (Throwable $ex) use ($application) {
			$this->logger->error(
				'ZMQ socket failed to receive message', [
					'exception_message' => $ex->getMessage(),
					'file'              => $ex->getFile(),
					'line'              => $ex->getLine(),
				]
			);

			$this->onFail($this);
		});
	}

	/**
	 * {@inheritdoc}
	 */
	public function close() : void
	{
		$this->socket ?: $this->socket->close();
	}
}