<commit_before># -*- coding: utf-8 -*-
#
# Revolution!
#

require 'timeout'

Module.new do

  @thread = nil

  plugin = Plugin::create(:streaming)

  plugin.add_event(:boot){ |service|
    @service = service
    start if UserConfig[:realtime_rewind] }

  UserConfig.connect(:realtime_rewind){ |key, new_val, before_val, id|
    if new_val
      Delayer.new{ self.start }
    else
      if @thread
        Plugin.call(:rewindstatus, 'UserStream: disconnected')
        @thread.kill
      else
        Plugin.call(:rewindstatus, 'UserStream: already disconnected. nothing to do.')
      end
    end
  }

  class << self

    def self.define_event(event_name)
      type_strict event_name => tcor(Symbol, String)
      queue = Queue.new
      service = nil
      Thread.new{
        sleep(1) while not service
        loop{
          sleep((UserConfig[:"#{event_name}_queue_delay"] || 100).to_f / 1000)
          datum = queue.pop
          yield(service, datum) } }
      define_method("event_#{event_name}"){ |json|
        type_strict json => tcor(Array, Hash)
        service ||= @service
        queue << json } end

    def self.define_together_event(event_name)
      type_strict event_name => tcor(Symbol, String)
      lock = Mutex.new
      events = Set.new
      service = nil
      thread = Thread.new{
        begin
        sleep(1) while not service
        loop{
          sleep((UserConfig[:"#{event_name}_queue_delay"] || 100).to_f / 1000)
          Thread.stop if events.empty?
          yield(service, lock.synchronize{ data = events; events = Set.new; data.freeze }) }
        rescue => e
          error e
          Plugin.activity :error, e.to_s, :exception => e end }
      define_method("event_#{event_name}"){ |json|
        type_strict json => tcor(Array, Hash)
        service ||= @service
        lock.synchronize{ events << json }
        if thread.alive?
          thread.wakeup
        else
          error "event_#{event_name}: event processing thread was dead."
          thread.join end } end

    def start
      unless @thread and @thread.alive?
        @thread = Thread.new{
          while(UserConfig[:realtime_rewind])
            sleep(3)
            catch(:streaming_break){
              start_streaming{ |q|
                throw(:streaming_break) unless(UserConfig[:realtime_rewind])<commit_msg>イベントを処理するスレッドをTimeLimitedQueueに変更 refs #424<commit_after># -*- coding: utf-8 -*-
#
# Revolution!
#

require 'timeout'

Module.new do

  @thread = nil

  plugin = Plugin::create(:streaming)

  plugin.add_event(:boot){ |service|
    @service = service
    start if UserConfig[:realtime_rewind] }

  UserConfig.connect(:realtime_rewind){ |key, new_val, before_val, id|
    if new_val
      Delayer.new{ self.start }
    else
      if @thread
        Plugin.call(:rewindstatus, 'UserStream: disconnected')
        @thread.kill
      else
        Plugin.call(:rewindstatus, 'UserStream: already disconnected. nothing to do.')
      end
    end
  }

  class << self

    def self.define_event(event_name)
      type_strict event_name => tcor(Symbol, String)
      queue = Queue.new
      service = nil
      Thread.new{
        sleep(1) while not service
        loop{
          sleep((UserConfig[:"#{event_name}_queue_delay"] || 100).to_f / 1000)
          datum = queue.pop
          yield(service, datum) } }
      define_method("event_#{event_name}"){ |json|
        type_strict json => tcor(Array, Hash)
        service ||= @service
        queue << json } end

    def self.define_together_event(event_name)
      type_strict event_name => tcor(Symbol, String)
      speed_key = "#{event_name}_queue_delay".to_sym
      queue = TimeLimitedQueue.new(HYDE, (UserConfig[speed_key] || 100).to_f / 1000) { |data|
        yield @service, data }
      UserConfig.connect(speed_key) { |key, val|
        notice "change #{key} #{val}"
        queue.expire = (val || 100).to_f / 1000 }
      define_method("event_#{event_name}"){ |json|
        type_strict json => tcor(Array, Hash)
        queue.push json } end

    def start
      unless @thread and @thread.alive?
        @thread = Thread.new{
          while(UserConfig[:realtime_rewind])
            sleep(3)
            catch(:streaming_break){
              start_streaming{ |q|
                throw(:streaming_break) unless(UserConfig[:realtime_rewind])
{'fraction_non_alphanumeric': 0.11799953906430052, 'fraction_numerical': 0.010371053238073288, 'mean_word_length': 2.409269442262372, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 13, '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': '4589908', 'n_tokens_mistral': 1395, 'n_tokens_neox': 1336, 'n_words': 298}