<commit_before><?php
/*
 * This file is part of the prooph/service-bus.
 * (c) 2014-2016 prooph software GmbH <<EMAIL>>
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 *
 * Date: 2016-08-11T08:00:00Z
 */

namespace Prooph\ServiceBus\Plugin\Router;

use Prooph\Common\Event\ActionEvent;
use Prooph\Common\Event\ActionEventEmitter;
use Prooph\Common\Event\ActionEventListenerAggregate;
use Prooph\Common\Event\DetachAggregateHandlers;
use Prooph\ServiceBus\Async\AsyncMessage;
use Prooph\ServiceBus\Async\MessageProducer;
use Prooph\ServiceBus\Exception;
use Prooph\ServiceBus\MessageBus;

/**
 * Class AsyncSwitchMessageRouter
 *
 * @package Prooph\ServiceBus\Router
 * @author <NAME> <<EMAIL>>
 */
class AsyncSwitchMessageRouter implements ActionEventListenerAggregate
{
    use DetachAggregateHandlers;

    /**
     * @var ActionEventListenerAggregate
     */
    protected $router;

    /**
     * @var MessageProducer
     */
    protected $asyncMessageProducer;


    /**
     * @param ActionEventListenerAggregate $router
     * @param MessageProducer $asyncMessageProducer
     */
    public function __construct(ActionEventListenerAggregate $router, MessageProducer $asyncMessageProducer)
    {
        $this->router = $router;
        $this->asyncMessageProducer = $asyncMessageProducer;
    }

    /**
     * @param ActionEventEmitter $events
     * @return void
     */
    public function attach(ActionEventEmitter $events)
    {
        $this->trackHandler($events->attachListener(MessageBus::EVENT_ROUTE, [$this, "onRouteMessage"]));
    }


    /**
     * @param ActionEvent $actionEvent
     */
    public function onRouteMessage(ActionEvent $actionEvent)
    {
        $messageName = (string)$actionEvent->getParam(MessageBus::EVENT_PARAM_MESSAGE_NAME);

        if (empty($messageName)) {
            return;
        }

        $message = $actionEvent->getParam(MessageBus::EVENT_PARAM_MESSAGE);

//        if (is_object($message) && $message instanceof Message)
//           $messageMetadata = $message->metadata();


        //if the message is marked as async, but had not yet been sent via async then send to async producer
        if ($message instanceof AsyncMessage && !($message->metadata()['handled-async'] == false)){
//            if ($message instanceof AsyncMessage && !(isset($messageMetadata['handled-by-async-queue']) && $messageMetadata['handled-by-async-queue'] === true)){

            //apply meta data, this is need to we can identify that the message has already been send via the async bus
            $message = $message->withAddedMetadata('handled-async', true);

            // update
            $actionEvent->setParam(MessageBus::EVENT_PARAM_MESSAGE, $message);
            $actionEvent->setParam(MessageBus::EVENT_PARAM_MESSAGE_HANDLER, $this->asyncMessageProducer);

            return;
        }


        // pass ActionEvent to decorated router
        return $this->router->onRouteMessage($actionEvent);

    }
}<commit_msg>fixed for php 5.5 & tidy up<commit_after><?php
/*
 * This file is part of the prooph/service-bus.
 * (c) 2014-2016 prooph software GmbH <<EMAIL>>
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 *
 * Date: 2016-08-11T08:00:00Z
 */

namespace Prooph\ServiceBus\Plugin\Router;

use Prooph\Common\Event\ActionEvent;
use Prooph\Common\Event\ActionEventEmitter;
use Prooph\Common\Event\ActionEventListenerAggregate;
use Prooph\Common\Event\DetachAggregateHandlers;
use Prooph\ServiceBus\Async\AsyncMessage;
use Prooph\ServiceBus\Async\MessageProducer;
use Prooph\ServiceBus\Exception;
use Prooph\ServiceBus\MessageBus;

/**
 * Class AsyncSwitchMessageRouter
 *
 * @package Prooph\ServiceBus\Router
 * @author <NAME> <<EMAIL>>
 */
class AsyncSwitchMessageRouter implements ActionEventListenerAggregate
{
    use DetachAggregateHandlers;

    /**
     * @var ActionEventListenerAggregate
     */
    protected $router;

    /**
     * @var MessageProducer
     */
    protected $asyncMessageProducer;


    /**
     * @param ActionEventListenerAggregate $router
     * @param MessageProducer $asyncMessageProducer
     */
    public function __construct(ActionEventListenerAggregate $router, MessageProducer $asyncMessageProducer)
    {
        $this->router = $router;
        $this->asyncMessageProducer = $asyncMessageProducer;
    }

    /**
     * @param ActionEventEmitter $events
     * @return void
     */
    public function attach(ActionEventEmitter $events)
    {
        $this->trackHandler($events->attachListener(MessageBus::EVENT_ROUTE, [$this, "onRouteMessage"]));
    }


    /**
     * @param ActionEvent $actionEvent
     */
    public function onRouteMessage(ActionEvent $actionEvent)
    {
        $messageName = (string)$actionEvent->getParam(MessageBus::EVENT_PARAM_MESSAGE_NAME);

        if (empty($messageName)) {
            return;
        }

        $message = $actionEvent->getParam(MessageBus::EVENT_PARAM_MESSAGE);

        //if the message is marked with AsyncMessage, but had not yet been sent via async then sent to async producer
        if ($message instanceof AsyncMessage && !(isset($messageMetadata['handled-async']) && $messageMetadata['handled-async'] === true)){

            //apply meta data, this is need to we can identify that the message has already been send via the async producer
            $message = $message->withAddedMetadata('handled-async', true);

            // update ActionEvent
            $actionEvent->setParam(MessageBus::EVENT_PARAM_MESSAGE, $message);
            $actionEvent->setParam(MessageBus::EVENT_PARAM_MESSAGE_HANDLER, $this->asyncMessageProducer);

            return;
        }

        // pass ActionEvent to decorated router
        return $this->router->onRouteMessage($actionEvent);
    }
}
{'fraction_non_alphanumeric': 0.10917400135409613, 'fraction_numerical': 0.007786052809749492, 'mean_word_length': 3.357669616519174, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 39, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '666909', 'n_tokens_mistral': 1653, 'n_tokens_neox': 1605, 'n_words': 379}