Streaming server-sent events (SSE) with Crystal

sse-example.cr

### streaming-server.cr

require "http/server"
require "json"
require "uuid"

def send_sse_frame(context, &)
  # SSE frame: "data: <json>\n\n"
  context.response.print "data: "
  yield context.response
  context.response.print "\n\n"
end

server = HTTP::Server.new do |context|
  current_client_id = UUID.random

  puts "Client #{current_client_id} connected."

  context.response.content_type = "text/event-stream"
  context.response.headers["Cache-Control"] = "no-cache"
  context.response.status_code = 200

  index = 0
  loop do
    index += 1

    send_sse_frame(context) do |response_io|
      { index: index, timestamp: Time.local }.to_json(response_io)
    end

    sleep 1.second
  rescue HTTP::Server::ClientError
    puts "Client #{current_client_id} disconnected."
    break
  rescue ex
    puts ex.message
    puts "Unexpected error, closing."
  end
end

puts "Streaming on http://localhost:3000"
server.bind_tcp 3000
server.listen

######## /streaming-server.cr

######## streaming-client.cr

require "http/client"
require "json"

# Example when parsing without a specific type
def parse_unstructured(data)
  begin
    obj = JSON.parse(data)

    puts "Parsed data to type: #{typeof(obj)}" # => JSON::Any
    puts "Event ##{obj["index"]}: #{obj["timestamp"]}"
  rescue e : JSON::ParseException
    STDERR.puts "bad JSON: #{data}"
  end
end

struct ServerEvent
  include JSON::Serializable
  property index : Int32
  property timestamp : Time

  def to_s
    "Event ##{@index}: #{@timestamp}"
  end
end

url = URI.parse("http://localhost:3000")
HTTP::Client.get(url) do |response|
  response_io = response.body_io.not_nil!

  loop do
    line = response_io.gets
    break unless line
    next if line.empty?

    if line.starts_with?("data:")
      data = line.lstrip("data:").strip
      # parse_unstructured(data)
      event = ServerEvent.from_json(data)
      puts event.to_s
    end
  end
end

######## /streaming-client.cr