<commit_before><?php

namespace DoSomething\MB_Toolbox;

use DoSomething\StatHat\Client as StatHat;
use DoSomething\MB_Toolbox\MB_Toolbox;
use DoSomething\MBC_DigestEmail\MBC_RabbitMQManagementAPI;

/*
 * MBC_UserAPICampaignActivity.class.in: Used to process the transactionalQueue
 * entries that match the campaign.*.* binding.
 */
abstract class MB_Toolbox_BaseConsumer
{

   // The number of seconds to pause when throttling is triggered
   const THROTTLE_TIMEOUT = 5;

  /**
   * Singleton instance of MB_Configuration application settings and service objects
   *
   * @var object
   */
  protected $mbConfig;

  /**
   * Message Broker connection to RabbitMQ
   *
   * @var object
   */
  protected $messageBroker;

  /**
   * The channel use by the Message Broker connection to RabbitMQ
   *
   * @var object
   */
  protected $channel;

  /**
   * StatHat object for logging of activity
   *
   * @var object
   */
  protected $statHat;

  /**
   * Message Broker Toolbox - collection of utility methods used by many of the
   * Message Broker producer and consumer applications.
   *
   * @var object
   */
  protected $mbToolbox;

  /**
   * Value of message from queue to be consumed / processed.
   *
   * @var array
   */
  protected $message;

  /**
   * The number of messages that have been processed. Used to calculate the message rate
   * to trigger throttling.
   *
   * @var integer
   */
  protected $throttleMessageCount = 0;

   /**
   * A second value to track the lapsed time to calculate the massage rate.
   *
   * @var integer
   */
  protected $throttleSecondStamp = NULL;

  /**
   * startTime - The date the request message started to be generated.
   *
   * @var string $startTime
   */
  protected $startTime;

  /**
   * Constructor for MBC_BaseConsumer - all consumer applications should extend this base class.
   *
   * @param string $targetMBconfig
   *   The Message Broker object used to interface the RabbitMQ server exchanges and related queues.
   */
  public function __construct($targetMBconfig = 'messageBroker') {

    $this->mbConfig = MB_Configuration::getInstance();
    $this->messageBroker = $this->mbConfig->getProperty($targetMBconfig);
    $this->statHat = $this->mbConfig->getProperty('statHat');
    $this->mbToolbox = $this->mbConfig->getProperty('mbToolbox');
    $this->mbcRabbitMQManagementAPI = $this->mbConfig->getProperty('mbcRabbitMQManagementAPI');

    $connection = $this->messageBroker->connection;
    $this->channel = $connection->channel();
  }

  /**
   * Initial method triggered by blocked call in base mbc-??-??.php file. The $payload is the
   * contents of the message being processed from the queue.
   *
   * @param array $payload
   *   The contents of the queue entry
   */
  public function consumeQueue($payload) {

    $this->message = unserialize($payload->body);
    $this->message['original'] = $this->message;
    $this->message['payload'] = $payload;
  }

   /**
   * Throddle the rate the consumer processes messages.
   *
   * @param integer $maxMessageRate
   *   The number of messages to process per second before triggering a "pause". Higher values allow for greater
   *   processing velosity / messages per second.
   */
  protected function throttle($maxMessageRate) {

    $this->throttleMessageCount++;

    // Reset the number of processed message by each second
    if ($this->throttleSecondStamp != date('s')) {
      $this->throttleSecondStamp = date('s');
      $this->throttleMessageCount = 0;
    }

    // Trigger processing delay when max message rate is exceeded
    if ($this->throttleMessageCount > $maxMessageRate) {
      sleep(MBC_BaseConsumer::THROTTLE_TIMEOUT);

      echo '- Trottling activated, message rate: ' . $this->throttleMessageCount . '. Waiting for ' . MBC_BaseConsumer::THROTTLE_TIMEOUT . ' seconds.', PHP_EOL;
      $this->throttleMessageCount = 0;
    }
  }

  /**
   * queueStatus(): Lookup the current message ready and unacked count of the queue that the
   * applicaiton is connected to.
   *
   * @param string $targetQueue
   *   The name of the queue to gather the stats from.
   *
   * return array $messageCount
   *   The numnber of messages in the "ready" and "unacked" state.
   */
  protected function queueStatus($targetQueue) {

    $queueStatus = $this->mbcRabbitMQManagementAPI->queueStatus($targetQueue);
    return $queueStatus;<commit_msg>Comment cleanup and move of mbRabbitMQManagementAPI into MB_Toolbox library<commit_after><?php

namespace DoSomething\MB_Toolbox;

use DoSomething\StatHat\Client as StatHat;
use DoSomething\MB_Toolbox\MB_Toolbox;

/*
 * MBC_UserAPICampaignActivity.class.in: Used to process the transactionalQueue
 * entries that match the campaign.*.* binding.
 */
abstract class MB_Toolbox_BaseConsumer
{

   // The number of seconds to pause when throttling is triggered
   const THROTTLE_TIMEOUT = 5;

  /**
   * Singleton instance of MB_Configuration application settings and service objects
   * @var object $mbConfig
   */
  protected $mbConfig;

  /**
   * Message Broker connection to RabbitMQ
   * @var object $messageBroker
   */
  protected $messageBroker;

  /**
   * The channel use by the Message Broker connection to RabbitMQ
   * @var object $channel
   */
  protected $channel;

  /**
   * StatHat object for logging of activity
   * @var object $statHat
   */
  protected $statHat;

  /**
   * Message Broker Toolbox - collection of utility methods used by many of the
   * Message Broker producer and consumer applications.
   * @var object $mbToolbox
   */
  protected $mbToolbox;

  /**
   * Value of message from queue to be consumed / processed.
   * @var array $message
   */
  protected $message;

  /**
   * The number of messages that have been processed. Used to calculate the message rate
   * to trigger throttling.
   * @var integer $throttleMessageCount
   */
  protected $throttleMessageCount = 0;

   /**
   * A second value to track the lapsed time to calculate the massage rate.
   * @var integer $throttleSecondStamp
   */
  protected $throttleSecondStamp = NULL;

  /**
   * The date the request message started to be generated.
   * @var string $startTime
   */
  protected $startTime;

  /**
   * A connection object to the RabbitMQ Management API.
   * @var object $mbRabbitMQManagementAPI
   */
  protected $mbRabbitMQManagementAPI;


  /**
   * Constructor for MBC_BaseConsumer - all consumer applications should extend this base class.
   *
   * @param string $targetMBconfig
   *   The Message Broker object used to interface the RabbitMQ server exchanges and related queues.
   */
  public function __construct($targetMBconfig = 'messageBroker') {

    $this->mbConfig = MB_Configuration::getInstance();
    $this->messageBroker = $this->mbConfig->getProperty($targetMBconfig);
    $this->statHat = $this->mbConfig->getProperty('statHat');
    $this->mbToolbox = $this->mbConfig->getProperty('mbToolbox');
    $this->mbRabbitMQManagementAPI = $this->mbConfig->getProperty('mbRabbitMQManagementAPI');

    $connection = $this->messageBroker->connection;
    $this->channel = $connection->channel();
  }

  /**
   * Initial method triggered by blocked call in base mbc-??-??.php file. The $payload is the
   * contents of the message being processed from the queue.
   *
   * @param array $payload
   *   The contents of the queue entry
   */
  public function consumeQueue($payload) {

    $this->message = unserialize($payload->body);
    $this->message['original'] = $this->message;
    $this->message['payload'] = $payload;
  }

   /**
   * Throddle the rate the consumer processes messages.
   *
   * @param integer $maxMessageRate
   *   The number of messages to process per second before triggering a "pause". Higher values allow for greater
   *   processing velosity / messages per second.
   */
  protected function throttle($maxMessageRate) {

    $this->throttleMessageCount++;

    // Reset the number of processed message by each second
    if ($this->throttleSecondStamp != date('s')) {
      $this->throttleSecondStamp = date('s');
      $this->throttleMessageCount = 0;
    }

    // Trigger processing delay when max message rate is exceeded
    if ($this->throttleMessageCount > $maxMessageRate) {
      sleep(MBC_BaseConsumer::THROTTLE_TIMEOUT);

      echo '- Trottling activated, message rate: ' . $this->throttleMessageCount . '. Waiting for ' . MBC_BaseConsumer::THROTTLE_TIMEOUT . ' seconds.', PHP_EOL;
      $this->throttleMessageCount = 0;
    }
  }

  /**
   * queueStatus(): Lookup the current message ready and unacked count of the queue that the
   * applicaiton is connected to.
   *
   * @param string $targetQueue
   *   The name of the queue to gather the stats from.
   *
   * return array $messageCount
   *   The numnber of messages in the "ready" and "unacked" state.
   */
  protected function queueStatus($targetQueue) {

    $queueStatus = $this->mbRabbitMQManagementAPI->queueStatus($targetQueue);
    return $queueStatus;
{'fraction_non_alphanumeric': 0.10202492211838006, 'fraction_numerical': 0.0008900756564307966, 'mean_word_length': 3.336227689339122, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 67, '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': '3605385', 'n_tokens_mistral': 2660, 'n_tokens_neox': 2568, 'n_words': 807}