<commit_before>require 'reel'

class Ssevents < Reel::Server
  include Celluloid::Logger

  def initialize data_source, ip = '127.0.0.1', port = 44444
    @connections = []
    @data_source = data_source
    @data = ''

    async.ring

    super ip, port, &method(:on_connection)
  end

  private

  def broadcast data
    @data = data

    info "Sending data to #{@connections.count} client(s)."
    @connections.each do |socket|
      async.send_sse socket, data
    end
  end

  def send_sse socket, data
    begin
      socket.data data
    rescue IOError, Errno::ECONNRESET, Errno::EPIPE
      @connections.delete socket
    end
  end

  def on_connection connection
    connection.each_request do |request|
      handle_request request
    end
  end

  def handle_request request
    body = Reel::EventStream.new do |socket|
      @connections << socket
      socket << "data: #{@data}\n\n"
    end

    headers = {
      'Content-Type'                => 'text/event-stream; charset=utf-8',
      'Cache-Control'               => 'no-cache',
      'Access-Control-Allow-Origin' => '*'
    }

    request.respond Reel::StreamResponse.new(:ok, headers, body)
  end

  def ring
    every 1 do
      broadcast @data_source.call
    end
  end
end<commit_msg>Call ::run explicitly<commit_after>require 'reel'

class Ssevents < Reel::Server
  include Celluloid::Logger

  def initialize data_source, ip = '127.0.0.1', port = 44444
    @connections = []
    @data_source = data_source
    @data = ''

    async.ring

    super ip, port, &method(:on_connection)
  end

  private

  def broadcast data
    @data = data

    info "Sending data to #{@connections.count} client(s)."
    @connections.each do |socket|
      async.send_sse socket, data
    end
  end

  def send_sse socket, data
    begin
      socket.data data
    rescue IOError, Errno::ECONNRESET, Errno::EPIPE
      @connections.delete socket
    end
  end

  def on_connection connection
    connection.each_request do |request|
      handle_request request
    end
  end

  def handle_request request
    body = Reel::EventStream.new do |socket|
      @connections << socket
      socket << "data: #{@data}\n\n"
    end

    headers = {
      'Content-Type'                => 'text/event-stream; charset=utf-8',
      'Cache-Control'               => 'no-cache',
      'Access-Control-Allow-Origin' => '*'
    }

    request.respond Reel::StreamResponse.new(:ok, headers, body)
  end

  def ring
    every 1 do
      broadcast @data_source.call
    end
  end
end

Ssevents.run -> { 'ololo' }
{'fraction_non_alphanumeric': 0.10763209393346379, 'fraction_numerical': 0.010176125244618396, 'mean_word_length': 2.477551020408163, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 10, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, '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': '2706010', 'n_tokens_mistral': 861, 'n_tokens_neox': 805, 'n_words': 211}