Simple implementation of a concurrent server handling requests on STDIN

concurrent_rpc.cr

# How to use / how it works:
#
# NOTE: The server reads on STDIN, and writes on STDOUT
#
# Start the program. Now you can write something then press enter.
# The server will do something 'useful' with the input, and gives you the result.
#
# The requests/responses are handled asynchronously, and can handle multiple requests at the same time.
#
# So you can spam the input (write a lot of lines) then close the input (Ctrl-d on a terminal), and
# see the responses being written as you type or after you closed the input.
#

record Request,
  id : Int32,
  data : String

record Result,
  req : Request,
  data : String

def get_requests(reader, input_channel)
  next_request_id = 0
  while data = reader.gets
    req = Request.new next_request_id, data
    next_request_id += 1
    input_channel.send req
  end
  input_channel.send nil # send a notification to the server: no more input!!
end

def handle_request(req, result_channel)
  sleep rand(3) # fake work
  new_data = req.data.upcase # do something useful

  result = Result.new req, new_data
  result_channel.send result
end

def run_server(reader, writer)
  # start the input fiber
  input_channel = Channel(Request | Nil).new
  spawn get_requests(reader, input_channel)

  channels = [] of Channel(Request | Nil) | Channel(Result)
  channels << input_channel

  # start server main loop
  until channels.empty?
    channel_index, data = Channel.select(channels.map &.receive_select_action)

    if channels[channel_index] == input_channel
      # this is the input channel!

      if data.nil?
        puts "Input closed, no more data to come..."
        channels.delete_at(channel_index) # remove the input channel from the channel list
        # we can't break out of the loop, because there may be some fibers that
        # are still processing a request.
        next
      end

      # we have a request!
      req = data.as(Request)

      # channel for the result
      result_channel = Channel(Result).new

      # handle the request in a new fiber
      spawn handle_request(req, result_channel)

      # add to the list of channels we monitor
      channels << result_channel
    else
      # one of the result channel received something!
      result = data.as(Result)

      # write a formatted output to the `writer` IO
      writer.puts "Got result for request #{result.req.id}: #{result.data}"

      channels.delete_at(channel_index) # we don't need this channel anymore
    end
  end
end

run_server(STDIN, STDOUT)