High performance Docker events api event handler prototype

docker-events.rb

#!/usr/bin/env ruby
require 'bundler/setup'

require 'http'
require 'celluloid'
require 'celluloid/io'
require 'json'


class Event
  include Celluloid
  def handle(event)
    Handlers.new.invoke(event)
  ensure
    terminate
  end

  class Handlers
    attr_reader :entries
    def initialize
      @entries = []
      add ::DieHandler
    end

    def add(klass, *args)
      new_entry = Entry.new(klass, *args)
      @entries <<  new_entry
    end

    def remove(entry)
      entries.delete_if {|e| e }
    end

    def invoke(event)
      chain = retrieve.dup
      chain.each {|entry| entry.call(event) }
    end
  private
    def retrieve
      entries.map(&:build)
    end

    class Entry
      attr_reader :klass

      def initialize(klass, *args)
        @klass = klass
        @args = args
      end

      def build
        @klass.new(*@args)
      end

      def ==(other)
        self.klass == other.klass
      end
    end
  end

end

class EventPoller
  include Celluloid::IO
  attr_reader :host

  def initialize(host)
    @host = host
  end

  def poll
    resp = HTTP.get("#{host}/events", socket_class: Celluloid::IO::TCPSocket).body
    loop do
      responses = ResponseParser.parse(resp.readpartial)
      responses.each { |event| Event.new.async.handle(event) }
      sleep 1
    end
  end

  module ResponseParser
    def self.parse(response)
      if response =~ /\}\{/
        parse_multi(response)
      else
        [JSON.parse(response)]
      end
    end

    def self.parse_multi(string)
      string.split('}{').collect do |s|
        value = case s
          when /^\{/    then "#{s}}"
          when /\}$/    then "{#{s}"
        end

        JSON.parse(value)
      end
    end
  end
end

class DieHandler
  def call(event)
    if event['status'] == 'die'
      puts "#{event['id']} DIED!!!"
    end
  end
end




host = ENV['DOCKER_HOST'] # Currenty just works over TCP/HTTP, URL must be http://...
EventPoller.new(host).poll