<commit_before>require 'protobuf/rpc/servers/zmq/util'
require 'protobuf/rpc/servers/zmq/worker'
require 'protobuf/rpc/servers/zmq/broker'
require 'protobuf/rpc/dynamic_discovery.pb'
require 'securerandom'

STDOUT.sync = true

module Protobuf
  module Rpc
    module Zmq
      class Server
        include ::Protobuf::Rpc::Zmq::Util

        attr_accessor :options, :workers, :zmq_context

        def initialize(options)
          @options = default_options.merge(options)
          @workers = []

          init_zmq_context
          init_beacon_socket if broadcast_beacons?
          init_shutdown_socket
        rescue
          teardown
          raise
        end

        def backend_ip
          frontend_ip
        end

        def backend_port
          options[:worker_port]
        end

        def backend_uri
          "tcp://#{backend_ip}:#{backend_port}"
        end

        def beacon_interval
          [options[:beacon_interval].to_i, 1].max
        end

        def beacon_ip
          "255.255.255.255"
        end

        def beacon_port
          options[:beacon_port].to_i
        end

        def broadcast_beacons?
          !brokerless? && options[:dynamic_discovery]
        end

        def broadcast_flatline
          flatline = ::Protobuf::DynamicDiscovery::Beacon.new(
            :beacon_type => ::Protobuf::DynamicDiscovery::BeaconType::FLATLINE,
            :uuid => uuid,
            :address => frontend_ip,
            :port => frontend_port.to_s
          )

          @beacon_socket.send flatline.to_s, 0
        end

        def broadcast_heartbeat
          heartbeat = ::Protobuf::DynamicDiscovery::Beacon.new(
            :beacon_type => ::Protobuf::DynamicDiscovery::BeaconType::HEARTBEAT,
            :uuid => uuid,
            :address => frontend_ip,
            :port => frontend_port.to_s,
            :ttl => beacon_interval * 3,
            :service_classes => ["not sure"] # TODO: Figure out what goes here
          )

          @beacon_socket.send heartbeat.to_s, 0
        end

        def brokerless?
          !!options[:workers_only]
        end

        def frontend_ip
          @frontend_ip ||= resolve_ip(options[:host])
        end

        def frontend_port
          options[:port]
        end

        def frontend_uri
          "tcp://#{frontend_ip}:#{frontend_port}"
        end

        def maintenance_interval
          [reaping_interval, beacon_interval].min<commit_msg>Add services to the heartbeat beacon<commit_after>require 'protobuf/rpc/servers/zmq/util'
require 'protobuf/rpc/servers/zmq/worker'
require 'protobuf/rpc/servers/zmq/broker'
require 'protobuf/rpc/dynamic_discovery.pb'
require 'securerandom'

STDOUT.sync = true

module Protobuf
  module Rpc
    module Zmq
      class Server
        include ::Protobuf::Rpc::Zmq::Util

        attr_accessor :options, :workers, :zmq_context

        def initialize(options)
          @options = default_options.merge(options)
          @workers = []

          init_zmq_context
          init_beacon_socket if broadcast_beacons?
          init_shutdown_socket
        rescue
          teardown
          raise
        end

        def backend_ip
          frontend_ip
        end

        def backend_port
          options[:worker_port]
        end

        def backend_uri
          "tcp://#{backend_ip}:#{backend_port}"
        end

        def beacon_interval
          [options[:beacon_interval].to_i, 1].max
        end

        def beacon_ip
          "255.255.255.255"
        end

        def beacon_port
          options[:beacon_port].to_i
        end

        def broadcast_beacons?
          !brokerless? && options[:dynamic_discovery]
        end

        def broadcast_flatline
          flatline = ::Protobuf::DynamicDiscovery::Beacon.new(
            :beacon_type => ::Protobuf::DynamicDiscovery::BeaconType::FLATLINE,
            :uuid => uuid,
            :address => frontend_ip,
            :port => frontend_port.to_s
          )

          @beacon_socket.send flatline.to_s, 0
        end

        def broadcast_heartbeat
          services = ::Protobuf::Rpc::Service.services
          heartbeat = ::Protobuf::DynamicDiscovery::Beacon.new(
            :beacon_type => ::Protobuf::DynamicDiscovery::BeaconType::HEARTBEAT,
            :uuid => uuid,
            :address => frontend_ip,
            :port => frontend_port.to_s,
            :ttl => beacon_interval * 3,
            :service_classes => services.map(&:name)
          )

          @beacon_socket.send heartbeat.to_s, 0
        end

        def brokerless?
          !!options[:workers_only]
        end

        def frontend_ip
          @frontend_ip ||= resolve_ip(options[:host])
        end

        def frontend_port
          options[:port]
        end

        def frontend_uri
          "tcp://#{frontend_ip}:#{frontend_port}"
        end

        def maintenance_interval
          [reaping_interval, beacon_interval].min
{'fraction_non_alphanumeric': 0.0904126213592233, 'fraction_numerical': 0.006472491909385114, 'mean_word_length': 1.8833819241982508, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 23, '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': '5380667', 'n_tokens_mistral': 1483, 'n_tokens_neox': 1416, 'n_words': 269}