| 1 | <?php |
||
| 11 | class MessageHandler |
||
| 12 | { |
||
| 13 | /** |
||
| 14 | * @var JSONMessageFactory |
||
| 15 | */ |
||
| 16 | private $jsonMessageFactory; |
||
| 17 | |||
| 18 | /** |
||
| 19 | * @var callable |
||
| 20 | */ |
||
| 21 | private $callback; |
||
| 22 | |||
| 23 | /** |
||
| 24 | * MessageHandler constructor. |
||
| 25 | * @param JSONMessageFactory $jsonMessageFactory |
||
| 26 | */ |
||
| 27 | public function __construct(JSONMessageFactory $jsonMessageFactory) |
||
| 31 | |||
| 32 | /** |
||
| 33 | * @param AMQPMessage $message |
||
| 34 | * @throws \Exception |
||
| 35 | */ |
||
| 36 | public function handleMessage(AMQPMessage $message) |
||
| 37 | { |
||
| 38 | if (!isset($this->callback)) { |
||
| 39 | throw new ReaderException("Handling a message with no callback set"); |
||
| 40 | } |
||
| 41 | |||
| 42 | try { |
||
| 43 | $task = $this->jsonMessageFactory->create($message->body); |
||
| 44 | $response = call_user_func($this->callback, $task); |
||
| 45 | if(is_null($response) || !is_bool($response) || $response) { |
||
| 46 | $message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']); |
||
| 47 | } |
||
| 48 | } catch(InvalidJSONMessageException $e) { |
||
| 49 | throw new ParseMessageException(json_encode($message->getBody()), 0, $e); |
||
| 50 | } |
||
| 51 | } |
||
| 52 | |||
| 53 | public function setCallback(callable $callback) |
||
| 57 | } |