<commit_before>require 'pubnub/configuration'
require 'pubnub/parser'
require 'pubnub/envelope'
require 'pubnub/crypto'
require 'pubnub/uuid'
require 'pubnub/exceptions'
require 'pubnub/event'
require 'pubnub/formatter'
require 'pubnub/validator'
require 'pubnub/pam'

module Pubnub
  class Client
    include Configuration

    attr_reader :env
    attr_accessor :single_event_connections_pool, :subscribe_event_connections_pool, :uuid

    EVENTS = %w(publish subscribe presence leave history here_now audit grant revoke time)
    VERSION = Pubnub::VERSION

    EVENTS.each do |event_name|
      require File.join('pubnub', 'events', event_name)
    end

    EVENTS.each do |event_name|
      define_method event_name do |params, &block|
        params[:callback] = block if params[:callback].nil?
        event = Pubnub.const_get(classify_method(event_name)).new(params, self)
        $logger.debug('Created event ' + event.class.to_s)
        if params[:http_sync]
          event.fire(self)
        else
          EM.next_tick {
            begin
              event.fire(self)
            rescue => e
              puts e
            end
          }
        end
        #event.fire(self)
      end
    end
    alias_method :unsubscribe, :leave

    def initialize(options)
      validate!(options)
      setup_app(options)
      # From this moment we have to use @env in that method instead of options
      create_connections_pools(@env)
      create_subscriptions_pools(@env)
      start_event_machine(@env)

    end

    def shutdown(stop_em = false)
      @single_event_connections_pool.each do |origin, conn|
        conn.shutdown_in_all_threads
        conn = nil
      end
      @single_event_connections_pool = Hash.new

      @subscribe_event_connections_pool.each do |origin, conn|
        conn.shutdown_in_all_threads
        conn = nil
      end
      @subscribe_event_connections_pool = Hash.new

      @env[:callback_pool] = Hash.new
      @env[:subscriptions] = Hash.new

      EM.stop if stop_em

      $logger.info('Bye!')
    end

    def start_subscribe
      @env[:wait_for_response] = Hash.new unless @wait_for_response
      @env[:railgun] = EM.add_periodic_timer(PERIODIC_TIMER_INTERVAL) do
        begin
          @env[:subscriptions].each do |origin, subscribe|
            unless @env[:wait_for_response][origin] == true
              @env[:wait_for_response][origin] = true

              $logger.debug('Async subscription running')
              $logger.debug("ORIGIN #{origin}")
              $logger.debug("SUBSCRIBE #{subscribe}")
              subscribe.start_event(self) if subscribe

              @env[:wait_for_response][origin] = false<commit_msg>Allow to override old sub<commit_after>require 'pubnub/configuration'
require 'pubnub/parser'
require 'pubnub/envelope'
require 'pubnub/crypto'
require 'pubnub/uuid'
require 'pubnub/exceptions'
require 'pubnub/event'
require 'pubnub/formatter'
require 'pubnub/validator'
require 'pubnub/pam'

module Pubnub
  class Client
    include Configuration

    attr_reader :env
    attr_accessor :single_event_connections_pool, :subscribe_event_connections_pool, :uuid

    EVENTS = %w(publish subscribe presence leave history here_now audit grant revoke time)
    VERSION = Pubnub::VERSION

    EVENTS.each do |event_name|
      require File.join('pubnub', 'events', event_name)
    end

    EVENTS.each do |event_name|
      define_method event_name do |params, &block|
        params[:callback] = block if params[:callback].nil?
        event = Pubnub.const_get(classify_method(event_name)).new(params, self)
        $logger.debug('Created event ' + event.class.to_s)
        if params[:http_sync]
          event.fire(self)
        else
          EM.next_tick {
            begin
              event.fire(self)
            rescue => e
              puts e
            end
          }
        end
        #event.fire(self)
      end
    end
    alias_method :unsubscribe, :leave

    def initialize(options)
      validate!(options)
      setup_app(options)
      # From this moment we have to use @env in that method instead of options
      create_connections_pools(@env)
      create_subscriptions_pools(@env)
      start_event_machine(@env)

    end

    def shutdown(stop_em = false)
      @single_event_connections_pool.each do |origin, conn|
        conn.shutdown_in_all_threads
        conn = nil
      end
      @single_event_connections_pool = Hash.new

      @subscribe_event_connections_pool.each do |origin, conn|
        conn.shutdown_in_all_threads
        conn = nil
      end
      @subscribe_event_connections_pool = Hash.new

      @env[:callback_pool] = Hash.new
      @env[:subscriptions] = Hash.new

      EM.stop if stop_em

      $logger.info('Bye!')
    end

    def start_subscribe(overwrite = false)

      if overwrite
        @env[:railgun].cancel
        @env[:railgun] = nil
        @env[:wait_for_response].each do |k,v|
          @env[:wait_for_response][k] = false
        end
      end

      @env[:wait_for_response] = Hash.new unless @wait_for_response
      @env[:railgun] = EM.add_periodic_timer(PERIODIC_TIMER_INTERVAL) do
        begin
          @env[:subscriptions].each do |origin, subscribe|
            unless @env[:wait_for_response][origin] == true
              @env[:wait_for_response][origin] = true

              $logger.debug('Async subscription running')
              $logger.debug("ORIGIN #{origin}")
              $logger.debug("SUBSCRIBE #{subscribe}")
              subscribe.start_event(self) if subscribe

              @env[:wait_for_response][origin] = false
{'fraction_non_alphanumeric': 0.09311452015697468, 'fraction_numerical': 0.0, 'mean_word_length': 2.619754680438993, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, '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': '193820', 'n_tokens_mistral': 1722, 'n_tokens_neox': 1664, 'n_words': 366}