<commit_before>require_relative 'version'
require_relative 'socket_store'

require 'em-hiredis'
require 'json'
require 'digest/sha2'
require 'time'
require 'logger'
require 'logger/colors'
require 'forwardable'


module PalavaMachine
  class Manager
    extend Forwardable


    attr_reader :connections


    def_delegators :@log, :debug, :info, :warn, :error, :fatal


    PAYLOAD_NEW_PEER = lambda { |connection_id, status = nil|
      payload = { event: 'new_peer', peer_id: connection_id }
      payload[:status] = status if status
      payload.to_json
    }

    PAYLOAD_PEER_LEFT = lambda { |connection_id| {
      event: 'peer_left',
      sender_id: connection_id,
    }.to_json }


    SCRIPT_JOIN_ROOM = <<-LUA
      local members = redis.call('smembers', KEYS[1])
      local count = 0
      for _, peer_id in pairs(members) do
        redis.call('publish', "ps:connection:" .. peer_id, ARGV[2])
        count = count + 1
      end
      redis.call('sadd', KEYS[1], ARGV[1])
      if count == 0 or tonumber(redis.call('get', KEYS[2])) <= count then
        redis.call('set', KEYS[2], count + 1)
      end
      redis.call('set', KEYS[3], ARGV[3])
      redis.call('set', KEYS[4], ARGV[4])
      return members
    LUA

    SCRIPT_LEAVE_ROOM = <<-LUA
      redis.call('hincrby', KEYS[7], math.floor((ARGV[3] - tonumber(redis.call('get', KEYS[3]))) / 60), 1) --stats
      redis.call('srem', KEYS[1], ARGV[1])
      redis.call('del', KEYS[3])
      redis.call('del', KEYS[4])
      redis.call('del', KEYS[5])

      if redis.call('scard', KEYS[1]) == 0 then -- also delete room if it is empty
        redis.call('hincrby', KEYS[6], redis.call('get', KEYS[2]), 1) --stats
        redis.call('del', KEYS[1])
        redis.call('del', KEYS[2])
      else -- tell others in room
        for _, peer_id in pairs(redis.call('smembers', KEYS[1])) do
          redis.call('publish', "ps:connection:" .. peer_id, ARGV[2])
        end
      end
    LUA


    def initialize(options = {})
      @redis_address = 'localhost:6379'
      @redis_db      = options[:db] || 0
      @connections   = SocketStore.new
      @log           = Logger.new(STDOUT)
      @log.level     = Logger::DEBUG
      @log.formatter = proc{ |level, datetime, _, msg|
        "#{datetime.strftime '%F %T'} | #{msg}\n"
      }
    end

    def initialize_in_em
      @redis      = EM::Hiredis.connect "redis://#{@redis_address}/#{@redis_db}"
      @publisher  = @redis.pubsub
      @subscriber = EM::Hiredis.connect("redis://#{@redis_address}/#{@redis_db}").pubsub # You need an extra connection for subs
      @redis.on :failed do
        @log.error 'Could not connect to Redis server'
      end
    end

    def announce_connection(ws)
      connection_id = @connections.register_connection(ws)
      info "#{connection_id} <open>"

      @subscriber.subscribe "ps:connection:#{connection_id}" do |payload|
        # debug "SUB payload #{payload} for <#{connection_id}>"
        ws.send_text(payload)
      end
    end

    def return_error(connection_id, message)
      raise MessageError.new(@connections[connection_id]), message
    end

    def unannounce_connection(ws, close_ws = false)
      if connection_id = @connections.unregister_connection(ws)
        info "#{connection_id} <close>"
        leave_room(connection_id)
        @subscriber.unsubscribe "ps:connection:#{connection_id}"
        if close_ws && ws.state != :closed # currently not used FIXME
          ws.close
        end
      end
    end

    def join_room(connection_id, room_id, status)
      return_error connection_id, 'no room id given' if !room_id || room_id.empty?
      return_error connection_id, 'room id too long' if room_id.size > 50

      @redis.get "store:connection:room:#{connection_id}" do |res|
        return_error connection_id, 'already joined another room' if res
        room_id = Digest::SHA512.hexdigest(room_id)
        info "#{connection_id} joins ##{room_id[0..10]}... #{status}"

        script_join_room(connection_id, room_id, status){ |members|
          return_error connection_id, 'room is full' unless members

          update_status_without_notifying_peers(connection_id, status){
            if members.empty?
              send_joined_room(connection_id, [])
            else
              get_statuses_for_members(members) do |members_with_statuses|
                send_joined_room(connection_id, members_with_statuses)
              end
            end
          }
        }
      end
    end

    def script_join_room(connection_id, room_id, status, &block)
      @redis.eval \
        SCRIPT_JOIN_ROOM,
        4,
        "store:room:members:#{room_id}",
        "store:room:peak_members:#{room_id}",
        "store:connection:joined:#{connection_id}",
        "store:connection:room:#{connection_id}",
        connection_id,
        PAYLOAD_NEW_PEER[connection_id, status],
        Time.now.getutc.to_i,
        room_id,
        &block
    end
    private :script_join_room

    def get_statuses_for_members(members)
      member_count = members.size
      members_with_statuses = []
      members.each { |peer_id|
        @redis.hgetall("store:connection:status:#{peer_id}") do |status_array|
          members_with_statuses << { peer_id: peer_id, status: Hash[status_array.each_slice(2).to_a] }
          yield members_with_statuses if members_with_statuses.size == member_count
        end
      }
    end
    private :get_statuses_for_members

    def send_joined_room(connection_id, members_with_statuses)
      @connections[connection_id].send_text({
        event: 'joined_room',
        own_id: connection_id,
        peers: members_with_statuses,
      }.to_json)
    end
    private :send_joined_room

    def leave_room(connection_id)
      @redis.get("store:connection:room:#{connection_id}") do |room_id|
        next unless room_id # return_error connection_id, 'currently not in any room'

        info "#{connection_id} leaves ##{room_id[0..10]}..."<commit_msg>fixes #1 - checking if connection is available before trying<commit_after>require_relative 'version'
require_relative 'socket_store'

require 'em-hiredis'
require 'json'
require 'digest/sha2'
require 'time'
require 'logger'
require 'logger/colors'
require 'forwardable'


module PalavaMachine
  class Manager
    extend Forwardable


    attr_reader :connections


    def_delegators :@log, :debug, :info, :warn, :error, :fatal


    PAYLOAD_NEW_PEER = lambda { |connection_id, status = nil|
      payload = { event: 'new_peer', peer_id: connection_id }
      payload[:status] = status if status
      payload.to_json
    }

    PAYLOAD_PEER_LEFT = lambda { |connection_id| {
      event: 'peer_left',
      sender_id: connection_id,
    }.to_json }


    SCRIPT_JOIN_ROOM = <<-LUA
      local members = redis.call('smembers', KEYS[1])
      local count = 0
      for _, peer_id in pairs(members) do
        redis.call('publish', "ps:connection:" .. peer_id, ARGV[2])
        count = count + 1
      end
      redis.call('sadd', KEYS[1], ARGV[1])
      if count == 0 or tonumber(redis.call('get', KEYS[2])) <= count then
        redis.call('set', KEYS[2], count + 1)
      end
      redis.call('set', KEYS[3], ARGV[3])
      redis.call('set', KEYS[4], ARGV[4])
      return members
    LUA

    SCRIPT_LEAVE_ROOM = <<-LUA
      redis.call('hincrby', KEYS[7], math.floor((ARGV[3] - tonumber(redis.call('get', KEYS[3]))) / 60), 1) --stats
      redis.call('srem', KEYS[1], ARGV[1])
      redis.call('del', KEYS[3])
      redis.call('del', KEYS[4])
      redis.call('del', KEYS[5])

      if redis.call('scard', KEYS[1]) == 0 then -- also delete room if it is empty
        redis.call('hincrby', KEYS[6], redis.call('get', KEYS[2]), 1) --stats
        redis.call('del', KEYS[1])
        redis.call('del', KEYS[2])
      else -- tell others in room
        for _, peer_id in pairs(redis.call('smembers', KEYS[1])) do
          redis.call('publish', "ps:connection:" .. peer_id, ARGV[2])
        end
      end
    LUA


    def initialize(options = {})
      @redis_address = 'localhost:6379'
      @redis_db      = options[:db] || 0
      @connections   = SocketStore.new
      @log           = Logger.new(STDOUT)
      @log.level     = Logger::DEBUG
      @log.formatter = proc{ |level, datetime, _, msg|
        "#{datetime.strftime '%F %T'} | #{msg}\n"
      }
    end

    def initialize_in_em
      @redis      = EM::Hiredis.connect "redis://#{@redis_address}/#{@redis_db}"
      @publisher  = @redis.pubsub
      @subscriber = EM::Hiredis.connect("redis://#{@redis_address}/#{@redis_db}").pubsub # You need an extra connection for subs
      @redis.on :failed do
        @log.error 'Could not connect to Redis server'
      end
    end

    def announce_connection(ws)
      connection_id = @connections.register_connection(ws)
      info "#{connection_id} <open>"

      @subscriber.subscribe "ps:connection:#{connection_id}" do |payload|
        # debug "SUB payload #{payload} for <#{connection_id}>"
        ws.send_text(payload)
      end
    end

    def return_error(connection_id, message)
      raise MessageError.new(@connections[connection_id]), message
    end

    def unannounce_connection(ws, close_ws = false)
      if connection_id = @connections.unregister_connection(ws)
        info "#{connection_id} <close>"
        leave_room(connection_id)
        @subscriber.unsubscribe "ps:connection:#{connection_id}"
        if close_ws && ws.state != :closed # currently not used FIXME
          ws.close
        end
      end
    end

    def join_room(connection_id, room_id, status)
      return_error connection_id, 'no room id given' if !room_id || room_id.empty?
      return_error connection_id, 'room id too long' if room_id.size > 50

      @redis.get "store:connection:room:#{connection_id}" do |res|
        return_error connection_id, 'already joined another room' if res
        room_id = Digest::SHA512.hexdigest(room_id)
        info "#{connection_id} joins ##{room_id[0..10]}... #{status}"

        script_join_room(connection_id, room_id, status){ |members|
          return_error connection_id, 'room is full' unless members

          update_status_without_notifying_peers(connection_id, status){
            if members.empty?
              send_joined_room(connection_id, [])
            else
              get_statuses_for_members(members) do |members_with_statuses|
                send_joined_room(connection_id, members_with_statuses)
              end
            end
          }
        }
      end
    end

    def script_join_room(connection_id, room_id, status, &block)
      @redis.eval \
        SCRIPT_JOIN_ROOM,
        4,
        "store:room:members:#{room_id}",
        "store:room:peak_members:#{room_id}",
        "store:connection:joined:#{connection_id}",
        "store:connection:room:#{connection_id}",
        connection_id,
        PAYLOAD_NEW_PEER[connection_id, status],
        Time.now.getutc.to_i,
        room_id,
        &block
    end
    private :script_join_room

    def get_statuses_for_members(members)
      member_count = members.size
      members_with_statuses = []
      members.each { |peer_id|
        @redis.hgetall("store:connection:status:#{peer_id}") do |status_array|
          members_with_statuses << { peer_id: peer_id, status: Hash[status_array.each_slice(2).to_a] }
          yield members_with_statuses if members_with_statuses.size == member_count
        end
      }
    end
    private :get_statuses_for_members

    def send_joined_room(connection_id, members_with_statuses)
      @connections[connection_id] and @connections[connection_id].send_text({
        event: 'joined_room',
        own_id: connection_id,
        peers: members_with_statuses,
      }.to_json)
    end
    private :send_joined_room

    def leave_room(connection_id)
      @redis.get("store:connection:room:#{connection_id}") do |room_id|
        next unless room_id # return_error connection_id, 'currently not in any room'

        info "#{connection_id} leaves ##{room_id[0..10]}..."
{'fraction_non_alphanumeric': 0.12442855955448424, 'fraction_numerical': 0.008893691297481507, 'mean_word_length': 2.8039835599114764, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 11, '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': '97646', 'n_tokens_mistral': 4065, 'n_tokens_neox': 3925, 'n_words': 829}