wcld in Ruby

test.sh

#!/bin/sh

message=$0 || "this just happened #boom"
echo "this just happened #boom" | nc -c localhost 4321
echo '....'
sleep 1
psql wcld -c 'select * from events order by time DESC;'

wcld.rb

#!/usr/bin/env ruby -X19

# Porting Ryan Smith's excellent wcld to Ruby and 
# such. Tested with Rubinius-2.0.0pre. I mostly wrote this to wrap my
# head around cool.io and girl_friday.

require 'rubygems'
require 'girl_friday'
require 'connection_pool'
require 'cool.io'
require 'pg'

ADDR = '127.0.0.1'
PORT = 4321

# To create the necessary database:
#
#   psql> CREATE DATABASE wcld;
#   psql> CREATE TABLE events (tag varchar(255), log_line text, time timestamp);
#
POOL = ConnectionPool.new(:size => 4) { PGconn.new(:host => 'localhost', :dbname => 'wcld') }

# One downside to cool.io is that it doesn't have a myriad of protocols
# implemented, as EventMachine does. My lazy workaround is to shove
# blocking work into a separate thread. Mike Perham's `girl_friday` handles
# that nicely.
LOG_QUEUE = GirlFriday::WorkQueue.new(:log, :size => 4) do |msg|
  begin
    tag = msg[:tag]
    line = msg[:line]

    POOL.with_connection do |pg|
      pg.exec(<<-SQL) { puts "YEP!" }
        INSERT INTO events (tag, log_line, time)
        VALUES ('#{tag}', '#{line}', NOW())
      SQL
    end
    puts "LOG(#{tag}, #{Time.now.iso8601}): #{line}"
  rescue => e
    p e
  end
end

# This class handles the actual logic of the wcld protocol. It's easy like
# Sunday morning.
class Wcld

  def self.process(data)
    new.process(data)
  end

  def process(data)
    tag, line = parse(data)
    log(tag, line)
    "OK"
  end

  def parse(data)
    tag = if match = %r{\B#([A-Za-z0-9_]+)}.match(data)
            match.captures[0]
          end
    line = data
    [tag, line]
  end

  def log(tag, line)
    LOG_QUEUE.push(:tag => tag, :line => line)
  end

end

# Cool.io is an evented IO gizmo, like EventMachine, node.js, or Twisted. Its
# different from EM in that it's based on libevent, is more Ruby and less C,
# and it's threadsafe/re-entrant (IIRC).
#
# This is how you use it as an abstraction over TCP sockets. Implement
# callbacks, handle the logic, go about your business. Easy!
class WcldConnection < Cool.io::TCPSocket

  def on_connect
    puts "#{remote_addr}:#{remote_port} connected"
  end

  def on_close
    puts "#{remote_addr}:#{remote_port} disconnected"
  end

  def on_read(data)
    write Wcld.process(data)
  end

end

# You can write EM-style connection classes for cool.io, but if you just want
# to hack a simple daemon together, you can this use this quick DSL to
# implement the sockety parts.
cool.io.connection :wcld do
  on_connect do
    puts "#{remote_addr}:#{remote_port} connected"
  end

  on_close do
    puts "#{remote_addr}:#{remote_port} disconnected"
  end

  on_read do |data|
    write Wcld.process(data)
  end
end

# This is how you start a class-style daemon
puts "wcld listening on #{ADDR}:#{PORT}"
event_loop = Cool.io::Loop.default
Cool.io::TCPServer.new(ADDR, PORT, WcldConnection).attach(event_loop)
event_loop.run

# This is how you start a DSL-style daemon
# cool.io.server ADDR, PORT, :wcld
# cool.io.run