Rubyで簡単な言語サーバー

langserver.rb

require "json"
require "logger"

REQUEST_TABLE = {}
NOTIFICATION_TABLE = {}
LOGGER = Logger.new(File.join(__dir__, "debug.log"))

def log(msg) LOGGER.info(msg) end

def read_header
  lines = []
  loop do
    line = $stdin.gets
    break if line == "\r\n"
    lines << line
  end
  lines.join
end

def get_content_length(header)
  if m = header.match(/^Content-Length: (\d+)/)
    m[1].to_i
  else
    raise "content length not found"
  end
end

def read_request
  header = read_header()
  content_length = get_content_length(header)
  $stdin.read(content_length)
end

def write_response(resp)
  resp_json = JSON.pretty_generate(resp)
  log "[response_body]\n#{resp_json}\n[/response_body]"

  print "Content-Length: #{resp_json.bytesize}\r\n"
  print "\r\n"
  print resp_json
  $stdout.flush
end

def send_response_ok(id:, result:)
  write_response({ jsonrpc: "2.0", id:, result: })
end

def send_notification(method:, params:)
  write_response({ jsonrpc: "2.0", method:, params: })
end

# --------------------------------
# request

REQUEST_TABLE["initialize"] = ->(msg){
  msg => { id: }

  result = {
    capabilities: {
      textDocumentSync: 1
    }
  }
  send_response_ok(id:, result:)
}

REQUEST_TABLE["shutdown"] = ->(msg){
  log "bye"
  exit
}

# --------------------------------
# notification

def send_publish_diagnostics(uri, diagnostics)
  send_notification(
    method: "textDocument/publishDiagnostics",
    params: { uri:, diagnostics: }
  )
end

def diagnose(text)
  diagnostics = []

  text.lines.each_with_index do |line, li|
    if line.include?("test")
      ci_end = line.chomp.size # TODO マルチバイト文字がある場合の考慮
      diagnostics << {
        range: {
          start: { line: li, character: 0 },
          end: { line: li, character: ci_end },
        },
        message: "test message line#{li}"
      }
    end
  end

  diagnostics
end

NOTIFICATION_TABLE["initialized"] = ->(msg){
  # TODO
}

NOTIFICATION_TABLE["textDocument/didOpen"] = ->(msg){
  msg => {
    params: { textDocument: { uri:, text: } }
  }

  diagnostics = diagnose(text)
  send_publish_diagnostics(uri, diagnostics)
}

NOTIFICATION_TABLE["textDocument/didChange"] = ->(msg){
  msg => {
    params: {
      textDocument: { uri: },
      contentChanges: content_changes
    }
  }

  if content_changes.size >= 1
    # 先に diagnostics を全て除去する（暫定的な措置）
    send_publish_diagnostics(uri, [])

    text = content_changes.last[:text]
    diagnostics = diagnose(text)
    send_publish_diagnostics(uri, diagnostics)
  end
}

# --------------------------------

def dispatch(msg)
  case msg
  in { id:, method: }
    log "method (#{method})"
    REQUEST_TABLE.fetch(method).call(msg)
  in { method: }
    log "method (#{method})"
    NOTIFICATION_TABLE.fetch(method).call(msg)
  in { id: }
    raise "TODO response"
  else
    raise "invalid request type"
  end
end

loop do
  begin
    req_body = read_request()
    log "req_body.inspect (#{req_body.inspect})"

    msg = JSON.parse(req_body, symbolize_names: true)
    log "[request_body]\n#{JSON.pretty_generate(msg)}\n[/request_body]"

    dispatch(msg)
  rescue => e
    LOGGER.error e
  end
end

Rubyで簡単な言語サーバーを書いてみた
https://qiita.com/sonota88/items/86817527bcabdb15b5cd

Rubyで簡単な言語サーバーを書いてみた（VSCode版）
https://zenn.dev/sonota88/articles/8298ad68edbd28