@@ -16,7 +16,7 @@ |
||
| 16 | 16 | protected $password; |
| 17 | 17 | protected $vHost; |
| 18 | 18 | |
| 19 | - public function __construct($host, $port, $user, $password, $vHost='/') |
|
| 19 | + public function __construct($host, $port, $user, $password, $vHost = '/') |
|
| 20 | 20 | { |
| 21 | 21 | $this->host = $host; |
| 22 | 22 | $this->port = $port; |
@@ -40,7 +40,7 @@ |
||
| 40 | 40 | */ |
| 41 | 41 | public function produce() |
| 42 | 42 | { |
| 43 | - if(!isset($this->tasks[0])) { |
|
| 43 | + if (!isset($this->tasks[0])) { |
|
| 44 | 44 | throw new TaskException('You must add at least 1 Task before producing.'); |
| 45 | 45 | } |
| 46 | 46 | $this->queueWriter->write($this->tasks); |
@@ -41,7 +41,7 @@ |
||
| 41 | 41 | */ |
| 42 | 42 | public function publish() |
| 43 | 43 | { |
| 44 | - if(!isset($this->events[0])) { |
|
| 44 | + if (!isset($this->events[0])) { |
|
| 45 | 45 | throw new DomainEventException('You must add at least 1 DomainEvent in order to publish to queue.'); |
| 46 | 46 | } |
| 47 | 47 | $this->queueWriter->write($this->events); |
@@ -17,5 +17,5 @@ |
||
| 17 | 17 | * @throws ReaderException |
| 18 | 18 | * @return void |
| 19 | 19 | */ |
| 20 | - public function read(callable $callback, $timeout=0); |
|
| 20 | + public function read(callable $callback, $timeout = 0); |
|
| 21 | 21 | } |
| 22 | 22 | \ No newline at end of file |
@@ -20,12 +20,12 @@ |
||
| 20 | 20 | $this->queueReader = $queueReader; |
| 21 | 21 | } |
| 22 | 22 | |
| 23 | - public function consume(callable $callback, $timeout=0) |
|
| 23 | + public function consume(callable $callback, $timeout = 0) |
|
| 24 | 24 | { |
| 25 | - while(true) { |
|
| 25 | + while (true) { |
|
| 26 | 26 | try { |
| 27 | 27 | $this->consumeOnce($callback, $timeout); |
| 28 | - } catch(TimeoutReaderException $e) { |
|
| 28 | + } catch (TimeoutReaderException $e) { |
|
| 29 | 29 | break; |
| 30 | 30 | } |
| 31 | 31 | } |
@@ -43,12 +43,12 @@ discard block |
||
| 43 | 43 | return $this; |
| 44 | 44 | } |
| 45 | 45 | |
| 46 | - public function start($timeout=0) |
|
| 46 | + public function start($timeout = 0) |
|
| 47 | 47 | { |
| 48 | - while(true) { |
|
| 48 | + while (true) { |
|
| 49 | 49 | try { |
| 50 | 50 | $this->processOne($timeout); |
| 51 | - } catch(TimeoutReaderException $e) { |
|
| 51 | + } catch (TimeoutReaderException $e) { |
|
| 52 | 52 | break; |
| 53 | 53 | } |
| 54 | 54 | } |
@@ -56,7 +56,7 @@ discard block |
||
| 56 | 56 | |
| 57 | 57 | public function processOne($timeout) |
| 58 | 58 | { |
| 59 | - if(!isset($this->subscriptors[0])) { |
|
| 59 | + if (!isset($this->subscriptors[0])) { |
|
| 60 | 60 | throw new DomainEventException('You must add at least 1 EventSubscriptor in order to publish start reading from queue.'); |
| 61 | 61 | } |
| 62 | 62 | $this->queueReader->read(array($this, 'notify'), $timeout); |
@@ -68,8 +68,8 @@ discard block |
||
| 68 | 68 | public function notify(DomainEvent $domainEvent) |
| 69 | 69 | { |
| 70 | 70 | $this->logger->info('Domain Event received, notifying subscribers'); |
| 71 | - foreach($this->subscriptors as $subscriptor) { |
|
| 72 | - if($subscriptor->isSubscribed($domainEvent)) { |
|
| 71 | + foreach ($this->subscriptors as $subscriptor) { |
|
| 72 | + if ($subscriptor->isSubscribed($domainEvent)) { |
|
| 73 | 73 | $subscriptor->notify($domainEvent); |
| 74 | 74 | } |
| 75 | 75 | } |
@@ -96,13 +96,13 @@ discard block |
||
| 96 | 96 | * @throws ReaderException |
| 97 | 97 | * @throws TimeoutReaderException |
| 98 | 98 | */ |
| 99 | - public function read(callable $callback, $timeout=0) |
|
| 99 | + public function read(callable $callback, $timeout = 0) |
|
| 100 | 100 | { |
| 101 | 101 | $this->initialize(); |
| 102 | 102 | $this->messageHandler->setCallback($callback); |
| 103 | 103 | try { |
| 104 | 104 | $this->channel->wait(null, false, $timeout); |
| 105 | - } catch(\Exception $e) { |
|
| 105 | + } catch (\Exception $e) { |
|
| 106 | 106 | throw new TimeoutReaderException(); |
| 107 | 107 | } |
| 108 | 108 | } |
@@ -112,7 +112,7 @@ discard block |
||
| 112 | 112 | */ |
| 113 | 113 | protected function initialize() |
| 114 | 114 | { |
| 115 | - if($this->channel) { |
|
| 115 | + if ($this->channel) { |
|
| 116 | 116 | return; |
| 117 | 117 | } |
| 118 | 118 | $this->logger->info('Connecting to RabbitMQ'); |
@@ -64,8 +64,8 @@ discard block |
||
| 64 | 64 | $this->initialize(); |
| 65 | 65 | try { |
| 66 | 66 | $messagesWithDelay = []; |
| 67 | - foreach($messages as $message) { |
|
| 68 | - if($message->getDelay() > 0) { |
|
| 67 | + foreach ($messages as $message) { |
|
| 68 | + if ($message->getDelay() > 0) { |
|
| 69 | 69 | $messagesWithDelay[$message->getDelay()][] = $message; |
| 70 | 70 | continue; |
| 71 | 71 | } |
@@ -75,8 +75,8 @@ discard block |
||
| 75 | 75 | $this->channel->batch_basic_publish($msg, $this->exchangeConfig->getName(), $message->getName()); |
| 76 | 76 | } |
| 77 | 77 | $this->channel->publish_batch(); |
| 78 | - foreach($messagesWithDelay as $delay => $delayedMessages) { |
|
| 79 | - if(!isset($this->delayedQueueWriterRegistry[$delay])) { |
|
| 78 | + foreach ($messagesWithDelay as $delay => $delayedMessages) { |
|
| 79 | + if (!isset($this->delayedQueueWriterRegistry[$delay])) { |
|
| 80 | 80 | $this->delayedQueueWriterRegistry[$delay] = new DelayedQueueWriter( |
| 81 | 81 | $this->exchangeConfig->getName(), |
| 82 | 82 | $delay, |
@@ -86,8 +86,8 @@ discard block |
||
| 86 | 86 | } |
| 87 | 87 | $this->delayedQueueWriterRegistry[$delay]->write($delayedMessages); |
| 88 | 88 | } |
| 89 | - } catch(\Exception $exception) { |
|
| 90 | - $this->logger->error('Error writing messages: '.$exception->getMessage()); |
|
| 89 | + } catch (\Exception $exception) { |
|
| 90 | + $this->logger->error('Error writing messages: ' . $exception->getMessage()); |
|
| 91 | 91 | throw new WriterException($exception->getMessage(), $exception->getCode()); |
| 92 | 92 | } |
| 93 | 93 | } |
@@ -97,7 +97,7 @@ discard block |
||
| 97 | 97 | */ |
| 98 | 98 | protected function initialize() |
| 99 | 99 | { |
| 100 | - if($this->channel) { |
|
| 100 | + if ($this->channel) { |
|
| 101 | 101 | return; |
| 102 | 102 | } |
| 103 | 103 | $this->logger->info('Connecting to RabbitMQ'); |
@@ -36,8 +36,8 @@ |
||
| 36 | 36 | public function __construct($origin, $name, $occurredOn, array $body = []) |
| 37 | 37 | { |
| 38 | 38 | $this->setOrigin($origin) |
| 39 | - ->setName($name) |
|
| 40 | - ->setOccurredOn($occurredOn) |
|
| 39 | + ->setName($name) |
|
| 40 | + ->setOccurredOn($occurredOn) |
|
| 41 | 41 | ; |
| 42 | 42 | $this->body = $body; |
| 43 | 43 | } |
@@ -91,7 +91,7 @@ discard block |
||
| 91 | 91 | */ |
| 92 | 92 | protected function setOrigin($origin) |
| 93 | 93 | { |
| 94 | - if(empty($origin)) { |
|
| 94 | + if (empty($origin)) { |
|
| 95 | 95 | throw new DomainEventException('DomainEvent origin cannot be empty'); |
| 96 | 96 | } |
| 97 | 97 | $this->origin = $origin; |
@@ -105,7 +105,7 @@ discard block |
||
| 105 | 105 | */ |
| 106 | 106 | protected function setName($name) |
| 107 | 107 | { |
| 108 | - if(empty($name)) { |
|
| 108 | + if (empty($name)) { |
|
| 109 | 109 | throw new DomainEventException('DomainEvent name cannot be empty'); |
| 110 | 110 | } |
| 111 | 111 | $this->name = $name; |
@@ -119,7 +119,7 @@ discard block |
||
| 119 | 119 | */ |
| 120 | 120 | protected function setOccurredOn($occurredOn) |
| 121 | 121 | { |
| 122 | - if(!is_null($occurredOn) && !preg_match('/^\d+(\.\d{1,4})?$/', $occurredOn)) { // accepts also microseconds |
|
| 122 | + if (!is_null($occurredOn) && !preg_match('/^\d+(\.\d{1,4})?$/', $occurredOn)) { // accepts also microseconds |
|
| 123 | 123 | throw new DomainEventException("$occurredOn is not a valid unix timestamp."); |
| 124 | 124 | } |
| 125 | 125 | $this->occurredOn = $occurredOn; |