<commit_before><?php

namespace Thruway\Role;

use Thruway\AbstractSession;
use Thruway\Logging\Logger;
use Thruway\Manager\ManagerDummy;
use Thruway\Manager\ManagerInterface;
use Thruway\Message\ErrorMessage;
use Thruway\Message\EventMessage;
use Thruway\Message\Message;
use Thruway\Message\PublishedMessage;
use Thruway\Message\PublishMessage;
use Thruway\Message\SubscribedMessage;
use Thruway\Message\SubscribeMessage;
use Thruway\Message\UnsubscribedMessage;
use Thruway\Message\UnsubscribeMessage;
use Thruway\Session;
use Thruway\Subscription;

/**
 * Class Broker
 *
 * @package Thruway\Role
 */
class Broker extends AbstractRole
{

    /**
     * @var \SplObjectStorage
     */
    private $subscriptions;

    /**
     * @var array
     */
    private $topics;

    /**
     * @var \Thruway\Manager\ManagerInterface
     */
    protected $manager;

    /**
     * Constructor
     *
     * @param \Thruway\Manager\ManagerInterface $manager
     */
    public function __construct(ManagerInterface $manager = null)
    {

        $this->subscriptions = new \SplObjectStorage();
        $this->topics        = [];
        $manager             = $manager ? $manager : new ManagerDummy();

        $this->setManager($manager);
        Logger::debug($this, "Broker constructor");
    }

    /**
     * Return supported features
     *
     * @return \stdClass
     */
    public function getFeatures() {
        $features = new \stdClass();

        $features->subscriber_blackwhite_listing = true;
        $features->publisher_exclusion = true;
        //$features->publisher_identification = true;
        $features->subscriber_metaevents = true;

        return $features;
    }

    /**
     * Handle received message
     *
     * @param \Thruway\AbstractSession $session
     * @param \Thruway\Message\Message $msg
     * @throws \Exception
     * @return mixed|void
     */
    public function onMessage(AbstractSession $session, Message $msg)
    {
        Logger::debug($this,
            "Broker onMessage for " . json_encode($session->getTransport()->getTransportDetails()) . ": " . json_encode($msg)
        );

        if ($msg instanceof PublishMessage):
            $this->processPublish($session, $msg);
        elseif ($msg instanceof SubscribeMessage):
            $this->processSubscribe($session, $msg);
        elseif ($msg instanceof UnsubscribeMessage):
            $this->processUnsubscribe($session, $msg);
        else:
            throw new \Exception("Unhandled message type sent to broker: " . get_class($msg));
        endif;
    }

    /**
     * Process publish message
     *
     * @param \Thruway\Session $session
     * @param \Thruway\Message\PublishMessage $msg
     */
    protected function processPublish(Session $session, PublishMessage $msg)
    {
        Logger::debug($this, "processing publish message");

        $includePublisher = false;
        $excludedSessions = [];
        $whiteList        = null;
        $options          = $msg->getOptions();

        // see if they wanted confirmation
        if (isset($options->acknowledge) && $options->acknowledge == true) {
            $publicationId = Session::getUniqueId();
            $session->sendMessage(
                new PublishedMessage($msg->getRequestId(), $publicationId)
            );
        }
        if (isset($options->exclude_me) && !$options->exclude_me) {
            $includePublisher = true;
        }
        if (isset($options->exclude) && is_array($options->exclude)) {
            // fixup exclude array - make sure it is legit
            foreach ($options->exclude as $excludedSession) {
                if (is_numeric($excludedSession)) {
                    array_push($excludedSessions, $excludedSession);
                }
            }
        }
        if (isset($options->eligible) && is_array($options->eligible)) {
            $whiteList = [];
            foreach ($options->eligible as $sessionId) {
                if (is_numeric($sessionId)) {
                    array_push($whiteList, $sessionId);
                }
            }
        }

        $this->sendEventMessages($session, $msg, $includePublisher, $excludedSessions, $whiteList);

    }

    /**
     * Send an Event Message for each subscription
     * @param Session $session
     * @param PublishMessage $msg
     * @param $includePublisher
     * @param $excludedSessions
     * @param $whiteList
     */
    private function sendEventMessages(Session $session, PublishMessage $msg, $includePublisher, $excludedSessions, $whiteList)
    {
        $arrayOfSubscriptions = [];

        /* @var $subscription \Thruway\Subscription */
        foreach ($this->subscriptions as $subscription) {
            array_push($arrayOfSubscriptions, $subscription);
        }

        foreach ($this->subscriptions as $subscription) {
        //foreach ($arrayOfSubscriptions as $subscription) {
            if ($msg->getTopicName() == $subscription->getTopic() &&
                ($includePublisher || $subscription->getSession() != $session)
            ) {
                if (!in_array($subscription->getSession()->getSessionId(), $excludedSessions)) {
                    if ($whiteList === null || in_array($subscription->getSession()->getSessionId(), $whiteList)) {
                        $eventMsg = EventMessage::createFromPublishMessage($msg, $subscription->getId());
                        $this->disclosePublisherOption($session, $eventMsg, $subscription);
                        $subscription->getSession()->sendMessage($eventMsg);
                    }
                }
            }
        }<commit_msg>Commented out the wrong thing.<commit_after><?php

namespace Thruway\Role;

use Thruway\AbstractSession;
use Thruway\Logging\Logger;
use Thruway\Manager\ManagerDummy;
use Thruway\Manager\ManagerInterface;
use Thruway\Message\ErrorMessage;
use Thruway\Message\EventMessage;
use Thruway\Message\Message;
use Thruway\Message\PublishedMessage;
use Thruway\Message\PublishMessage;
use Thruway\Message\SubscribedMessage;
use Thruway\Message\SubscribeMessage;
use Thruway\Message\UnsubscribedMessage;
use Thruway\Message\UnsubscribeMessage;
use Thruway\Session;
use Thruway\Subscription;

/**
 * Class Broker
 *
 * @package Thruway\Role
 */
class Broker extends AbstractRole
{

    /**
     * @var \SplObjectStorage
     */
    private $subscriptions;

    /**
     * @var array
     */
    private $topics;

    /**
     * @var \Thruway\Manager\ManagerInterface
     */
    protected $manager;

    /**
     * Constructor
     *
     * @param \Thruway\Manager\ManagerInterface $manager
     */
    public function __construct(ManagerInterface $manager = null)
    {

        $this->subscriptions = new \SplObjectStorage();
        $this->topics        = [];
        $manager             = $manager ? $manager : new ManagerDummy();

        $this->setManager($manager);
        Logger::debug($this, "Broker constructor");
    }

    /**
     * Return supported features
     *
     * @return \stdClass
     */
    public function getFeatures() {
        $features = new \stdClass();

        $features->subscriber_blackwhite_listing = true;
        $features->publisher_exclusion = true;
        //$features->publisher_identification = true;
        $features->subscriber_metaevents = true;

        return $features;
    }

    /**
     * Handle received message
     *
     * @param \Thruway\AbstractSession $session
     * @param \Thruway\Message\Message $msg
     * @throws \Exception
     * @return mixed|void
     */
    public function onMessage(AbstractSession $session, Message $msg)
    {
        Logger::debug($this,
            "Broker onMessage for " . json_encode($session->getTransport()->getTransportDetails()) . ": " . json_encode($msg)
        );

        if ($msg instanceof PublishMessage):
            $this->processPublish($session, $msg);
        elseif ($msg instanceof SubscribeMessage):
            $this->processSubscribe($session, $msg);
        elseif ($msg instanceof UnsubscribeMessage):
            $this->processUnsubscribe($session, $msg);
        else:
            throw new \Exception("Unhandled message type sent to broker: " . get_class($msg));
        endif;
    }

    /**
     * Process publish message
     *
     * @param \Thruway\Session $session
     * @param \Thruway\Message\PublishMessage $msg
     */
    protected function processPublish(Session $session, PublishMessage $msg)
    {
        Logger::debug($this, "processing publish message");

        $includePublisher = false;
        $excludedSessions = [];
        $whiteList        = null;
        $options          = $msg->getOptions();

        // see if they wanted confirmation
        if (isset($options->acknowledge) && $options->acknowledge == true) {
            $publicationId = Session::getUniqueId();
            $session->sendMessage(
                new PublishedMessage($msg->getRequestId(), $publicationId)
            );
        }
        if (isset($options->exclude_me) && !$options->exclude_me) {
            $includePublisher = true;
        }
        if (isset($options->exclude) && is_array($options->exclude)) {
            // fixup exclude array - make sure it is legit
            foreach ($options->exclude as $excludedSession) {
                if (is_numeric($excludedSession)) {
                    array_push($excludedSessions, $excludedSession);
                }
            }
        }
        if (isset($options->eligible) && is_array($options->eligible)) {
            $whiteList = [];
            foreach ($options->eligible as $sessionId) {
                if (is_numeric($sessionId)) {
                    array_push($whiteList, $sessionId);
                }
            }
        }

        $this->sendEventMessages($session, $msg, $includePublisher, $excludedSessions, $whiteList);

    }

    /**
     * Send an Event Message for each subscription
     * @param Session $session
     * @param PublishMessage $msg
     * @param $includePublisher
     * @param $excludedSessions
     * @param $whiteList
     */
    private function sendEventMessages(Session $session, PublishMessage $msg, $includePublisher, $excludedSessions, $whiteList)
    {
        $arrayOfSubscriptions = [];

        /* @var $subscription \Thruway\Subscription */
        foreach ($this->subscriptions as $subscription) {
            array_push($arrayOfSubscriptions, $subscription);
        }

        //foreach ($this->subscriptions as $subscription) {
        foreach ($arrayOfSubscriptions as $subscription) {
            if ($msg->getTopicName() == $subscription->getTopic() &&
                ($includePublisher || $subscription->getSession() != $session)
            ) {
                if (!in_array($subscription->getSession()->getSessionId(), $excludedSessions)) {
                    if ($whiteList === null || in_array($subscription->getSession()->getSessionId(), $whiteList)) {
                        $eventMsg = EventMessage::createFromPublishMessage($msg, $subscription->getId());
                        $this->disclosePublisherOption($session, $eventMsg, $subscription);
                        $subscription->getSession()->sendMessage($eventMsg);
                    }
                }
            }
        }
{'fraction_non_alphanumeric': 0.12385687649826867, 'fraction_numerical': 0.0, 'mean_word_length': 2.2935672514619885, 'pattern_counts': {'":': 2, '<': 5, '<?xml version=': 0, '>': 81, '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': '3367651', 'n_tokens_mistral': 3066, 'n_tokens_neox': 2982, 'n_words': 602}