class ObjectFramework::GRPC::Server

# File lib/object_framework/grpc/server.rb, line 81
def initialize(application:, services:, controllers:, host: "127.0.0.1", port: 0,
               name: :grpc, max_requests: 128, max_connections: 128,
               message_limit: 1024 * 1024, header_limit: 32 * 1024,
               execution_timeout: 30, idle_timeout: 30, shutdown_timeout: 5,
               tls_context: nil, status_presenter: nil)
  @application, @services, @controllers, @host, @port, @name = application, services.dup, controllers.dup, host, port, name
  @max_requests, @max_connections, @message_limit, @header_limit = max_requests, max_connections, message_limit, header_limit
  @execution_timeout, @idle_timeout, @shutdown_timeout, @tls_context = execution_timeout, idle_timeout, shutdown_timeout, tls_context
  @presenter = status_presenter || StatusPresenter.new(registry: application.errors)
  @mutex, @workers, @connections, @routes = Mutex.new, {}, {}.compare_by_identity, {}
end

# File lib/object_framework/grpc/server.rb, line 138
def active_count; @mutex.synchronize { @workers.size }; end

# File lib/object_framework/grpc/server.rb, line 95
def configure(application)
  application.errors.register(Unimplemented, category: :unimplemented, code: "rpc_unimplemented", title: "RPC method not implemented")
  application.errors.register(HTTP2::ProtocolError, category: :bad_request, code: "http2_protocol_error", title: "Invalid HTTP/2 request")
  application.errors.register(HTTP2::CompressionError, category: :bad_request, code: "hpack_error", title: "Invalid compressed headers")
end

# File lib/object_framework/grpc/server.rb, line 93
def requires; []; end

# File lib/object_framework/grpc/server.rb, line 129
def start
  @listener = TCPServer.new(@host, @port)
  @port = @listener.local_address.ip_port
  @wake_reader, @wake_writer = IO.pipe
  @running = true
  @acceptor = Thread.new { accept_connections }
  self
end

# File lib/object_framework/grpc/server.rb, line 140
def stop
  return unless @acceptor
  @running = false
  @wake_writer.write_nonblock(".", exception: false)
  @acceptor.join
  connections, workers = @mutex.synchronize do
    @workers.each_value { |stream| stream.context.cancellation.cancel }
    [@connections.to_a, @workers.keys]
  end
  connections.each { |connection, _| connection.close }
  deadline = HTTP::Timing.now + @shutdown_timeout
  (workers + connections.map(&:last)).each { |thread| thread.join([deadline - HTTP::Timing.now, 0].max) }
  @listener.close
  @wake_reader.close
  @wake_writer.close
  @acceptor = nil
end

# File lib/object_framework/grpc/server.rb, line 101
def validate!
  unless [@max_requests, @max_connections, @message_limit, @header_limit].all? { |value| value.is_a?(Integer) && value.positive? }
    raise ConfigurationError, "gRPC limits must be positive integers"
  end
  unless [@execution_timeout, @idle_timeout, @shutdown_timeout].all? { |value| value.is_a?(Numeric) && value.positive? }
    raise ConfigurationError, "gRPC timeouts must be positive"
  end
  @services.each do |service|
    service.methods.each_value do |method|
      controller = @controllers.fetch(method.controller) { raise ConfigurationError, "Missing RPC controller" }
      raise ConfigurationError, "RPC controllers must implement call" unless controller.respond_to?(:call)
      controller.validate! if controller.respond_to?(:validate!)
      path = "/#{service.name}/#{method.name}"
      raise ConfigurationError, "Duplicate RPC route" if @routes.key?(path)
      @routes[path] = method
    end
    service.finalize!
  end
  @routes.freeze
  @controllers.freeze
  @services.freeze
  if @tls_context
    raise ConfigurationError, "TLS requires a certificate and private key" unless @tls_context.cert && @tls_context.key
    @tls_context.alpn_select_cb = ->(protocols) { protocols.include?("h2") ? "h2" : raise(HTTP2::ProtocolError, "h2 required") }
    @tls_context.setup
  end
end