completing the merge

#!/usr/bin/env ruby

$: << ::File.expand_path('../../lib', __FILE__)

require "fnordquery"

require "fnordquery/runner"

FnordQuery::Runner.new

 No newline at end of file

fnordquery

==========

fnordquery analyzes streams of usage/traffic/activity/etc-data. you feed in json-objects ("events") and it spits out pretty reports. the only schema-wise requirement is that each event contains a "_time" key.

fnordquery requires a fyrehose backend.

categorical_correlation (keyword correlation)

categorical_crosstab (chanels x filters)

categorical_uniqueness (zipfs law)

cluster_cardinality (number of searches per session)

cluster_repetition (search term repetition per session)

numeric_distribution (price/age dist, w. variance/dispersion and stuff)

numeric_correlation (numeric dependency)

retention_analysis

click through?

module FnordQuery; end

require "eventmachine"

require "json"

require "haml"

require "sinatra"

require "fnordquery/web/web"

require "fnordquery/web/app"

require "fnordquery/backends/redis_backend"

require "fnordquery/acceptor"

require "fnordquery/acceptors/tcp_acceptor"

require "fnordquery/acceptors/udp_acceptor"

require "fnordquery/query"

require "fnordquery/report"

require "fnordquery/report_builder"

require "fnordquery/report_manager"

require "fnordquery/reports/numeric_timeseries_report"

require "fnordquery/reports/categorical_topk_report"

 No newline at end of file

class FnordQuery::Acceptor

  def initialize(opts)

    @opts = opts

  end

  def execute(runner, backend)

    inbound_class = if @opts[:protocol] == :udp

      FnordQuery::UDPAcceptor

    else

      FnordQuery::TCPAcceptor

    end

    @opts[:listen] = [

      @opts[:host] || "0.0.0.0",

      @opts[:port] || 2323

    ]

    @opts.merge!(

      :runner => runner,

      :backend => backend

    )

    begin

      inbound_stream = inbound_class.start(@opts)

      puts "listening on #{@opts[:protocol]}://#{@opts[:listen][0..1].join(":")}"

    #rescue

    #  puts "error: cant start #{inbound_class.name}. port in use?"

    #  exit!(1)

    end

  end

end

 No newline at end of file

class FnordQuery::TCPAcceptor < EventMachine::Connection

  @@opts = nil

  def self.start(opts)

    @@opts = opts

    EM.start_server(*(opts[:listen] + [self]))

  end

  def self.options(opts)

    @@opts = opts

  end

  def receive_data(chunk)

    @buffer << chunk

    next_event

  end

  def next_event

    read_next_event

    push_next_event

  end

  def read_next_event

    while (event = @buffer.slice!(/^(.*)\n/))

      @events_buffered += 1

      @events << event

    end

  end

  def push_next_event

    return true if @events.empty?

    @events_buffered -= 1

    @backend.publish(@events.pop)

    close_connection?

    EM.next_tick(&method(:push_next_event))

  end

  def close_connection?

    @backend.hangup unless @streaming || (@events_buffered!=0)

  end

  def post_init

    @backend = @@opts[:backend][0].new(@@opts[:backend][1])

    @events_buffered = 0

    @streaming = true

    @buffer = ""

    @events = []

  end

  def unbind

    @streaming = false

    close_connection?

  end

end