<commit_before>require 'zmq'

module RZ
  module Context

  HALLO = ['HALLO'].freeze
  NOOP  = ['NOOP'].freeze
  DELIM = [''].freeze

  private

    # NOOP override this if you whant to log messages
    def log(level,&block)
    end

    def debug(&block)
      log :debug,&block 
    end

    def warn(&block)
      log :warn,&block 
    end

    def info(&block)
      log :info,&block 
    end

    def error(&block)
      log :error,&block 
    end

    def zmq_cleanup
      zmq_named_sockets.each_key do |name|
        zmq_named_socket_close name
      end
      zmq_sockets.each do |socket|
        zmq_socket_close socket
      end
      raise unless zmq_sockets.length.zero?
      zmq_context.close 
      @zmq_named_sockets = @zmq_context = @zmq_sockets = nil
    end

    def zmq_context
      @zmq_context ||= ZMQ::Context.new
    end

    def zmq_split(message)
      delim = nil
      message.each_with_index do |part,idx|
        if part.length == 0 
          delim = idx 
          break
        end
      end
      raise 'undelimited message' unless delim
      [message[0...delim],message[(delim+1)..-1]]
    end

    def zmq_recv(socket,flags=0)
      debug { "#{zmq_identity(socket)} recv message" }
      message = []
      loop do
        part = socket.recv(flags)
        return unless part
        message << part
        more = socket.getsockopt ZMQ::RCVMORE
        opts = 0
        opts |= ZMQ::SNDMORE if more
        break unless more
      end
      debug { "#{zmq_identity(socket)} recved message: #{message.inspect}" }
      message
    end

    def zmq_identity(socket)
      socket.getsockopt(ZMQ::IDENTITY) || 'UNKOWN'
    end

    def zmq_send(socket,message)
      debug { "#{zmq_identity(socket)} send message: #{message.inspect}" }
      message.each_with_index do |part,idx|
        last = message.length == idx+1
        flags = 0
        flags |= ZMQ::SNDMORE unless last
        unless socket.send part,flags
          raise
        end
      end
      debug { "#{zmq_identity(socket)} send finished" }
      self
    end

    def zmq_named_socket(name,type)
      zmq_named_sockets[name] ||= begin
                                    socket = zmq_socket(type)
                                    yield socket if block_given?
                                    socket
                                  end
    end

    def zmq_named_socket_close(name)
      socket = zmq_named_sockets.delete name
      unless socket
        warn { "requested to close inexistend named socket: #{name.inspect}" } 
      end
      zmq_socket_close socket if socket
    end

    def zmq_named_sockets
      @zmq_named_sockets ||= {}
    end

    def zmq_socket(type)
      p type
      socket = zmq_context.socket(type)
      p socket
      zmq_sockets << socket
      socket
    end

    def zmq_socket_close(socket)
      socket.close
      zmq_sockets.delete socket
    end

    def zmq_sockets
      @zmq_sockets ||= []
    end
  end
end
<commit_msg>Remove leftover debug prints<commit_after>require 'zmq'

module RZ
  module Context

  HALLO = ['HALLO'].freeze
  NOOP  = ['NOOP'].freeze
  DELIM = [''].freeze

  private

    # NOOP override this if you whant to log messages
    def log(level,&block)
    end

    def debug(&block)
      log :debug,&block 
    end

    def warn(&block)
      log :warn,&block 
    end

    def info(&block)
      log :info,&block 
    end

    def error(&block)
      log :error,&block 
    end

    def zmq_cleanup
      zmq_named_sockets.each_key do |name|
        zmq_named_socket_close name
      end
      zmq_sockets.each do |socket|
        zmq_socket_close socket
      end
      raise unless zmq_sockets.length.zero?
      zmq_context.close 
      @zmq_named_sockets = @zmq_context = @zmq_sockets = nil
    end

    def zmq_context
      @zmq_context ||= ZMQ::Context.new
    end

    def zmq_split(message)
      delim = nil
      message.each_with_index do |part,idx|
        if part.length == 0 
          delim = idx 
          break
        end
      end
      raise 'undelimited message' unless delim
      [message[0...delim],message[(delim+1)..-1]]
    end

    def zmq_recv(socket,flags=0)
      debug { "#{zmq_identity(socket)} recv message" }
      message = []
      loop do
        part = socket.recv(flags)
        return unless part
        message << part
        more = socket.getsockopt ZMQ::RCVMORE
        opts = 0
        opts |= ZMQ::SNDMORE if more
        break unless more
      end
      debug { "#{zmq_identity(socket)} recved message: #{message.inspect}" }
      message
    end

    def zmq_identity(socket)
      socket.getsockopt(ZMQ::IDENTITY) || 'UNKOWN'
    end

    def zmq_send(socket,message)
      debug { "#{zmq_identity(socket)} send message: #{message.inspect}" }
      message.each_with_index do |part,idx|
        last = message.length == idx+1
        flags = 0
        flags |= ZMQ::SNDMORE unless last
        unless socket.send part,flags
          raise
        end
      end
      debug { "#{zmq_identity(socket)} send finished" }
      self
    end

    def zmq_named_socket(name,type)
      zmq_named_sockets[name] ||= begin
                                    socket = zmq_socket(type)
                                    yield socket if block_given?
                                    socket
                                  end
    end

    def zmq_named_socket_close(name)
      socket = zmq_named_sockets.delete name
      unless socket
        warn { "requested to close inexistend named socket: #{name.inspect}" } 
      end
      zmq_socket_close socket if socket
    end

    def zmq_named_sockets
      @zmq_named_sockets ||= {}
    end

    def zmq_socket(type)
      socket = zmq_context.socket(type)
      zmq_sockets << socket
      socket
    end

    def zmq_socket_close(socket)
      socket.close
      zmq_sockets.delete socket
    end

    def zmq_sockets
      @zmq_sockets ||= []
    end
  end
end

{'fraction_non_alphanumeric': 0.08441230872708928, 'fraction_numerical': 0.0026904321506642003, 'mean_word_length': 1.854126679462572, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2710193', 'n_tokens_mistral': 1939, 'n_tokens_neox': 1842, 'n_words': 457}