diff --git a/lib/stathat.rb b/lib/stathat.rb index 711fb32..21d29e7 100644 --- a/lib/stathat.rb +++ b/lib/stathat.rb @@ -10,30 +10,71 @@ class Common CLASSIC_VALUE_URL = "https://api.stathat.com/v" CLASSIC_COUNT_URL = "https://api.stathat.com/c" EZ_URL = "https://api.stathat.com/ez" + EZ_URI = URI(EZ_URL) class << self - def send_to_stathat(url, args) + def stathat_uri(url, args) uri = URI.parse(url) begin uri.query = URI.encode_www_form(args) - rescue NoMethodError => e + rescue NoMethodError # backwards compatability for pre 1.9.x uri.query = args.map { |arg, val| arg.to_s + "=" + CGI::escape(val.to_s) }.join('&') end - resp = Net::HTTP.get(uri) + return uri + end + + def send_to_stathat(url, args) + uri = stathat_uri(url, args) + resp = Net::HTTP.get_response(uri) return Response.new(resp) end + + def send_ez_batch_to_stathat(batch_args, ezkey) + resp = Net::HTTP.start(EZ_URI.host, EZ_URI.port, :use_ssl => true) do |http| + http.post EZ_URI.path, + { :ezkey => ezkey, :data => batch_args }.to_json, + "Content-Type" => "application/json" + end + return Response.new(resp) + end end end + class Export + class Error < StandardError + attr_reader :response + + def initialize(msg, response) + @response = response + super(msg) + end + end + + EXPORT_URL = "https://www.stathat.com/x" + + class << self + def get(access_token, paths, args) + uri = Common::stathat_uri(([EXPORT_URL, access_token] + paths).join('/'), args) + resp = Net::HTTP.get_response(uri) + + if resp.kind_of?(Net::HTTPSuccess) && resp.body + return JSON.parse(resp.body) + else + raise Error.new("export error #{resp.inspect} for #{uri.inspect}", resp) + end + end + end + end + class SyncAPI class << self def ez_post_value(stat_name, ezkey, value, timestamp=nil) args = { :stat => stat_name, :ezkey => ezkey, - :value => value } + :value => Float(value) } args[:t] = timestamp unless timestamp.nil? Common::send_to_stathat(Common::EZ_URL, args) end @@ -41,7 +82,7 @@ def ez_post_value(stat_name, ezkey, value, timestamp=nil) def ez_post_count(stat_name, ezkey, count, timestamp=nil) args = { :stat => stat_name, :ezkey => ezkey, - :count => count } + :count => Integer(count) } args[:t] = timestamp unless timestamp.nil? Common::send_to_stathat(Common::EZ_URL, args) end @@ -49,7 +90,7 @@ def ez_post_count(stat_name, ezkey, count, timestamp=nil) def post_count(stat_key, user_key, count, timestamp=nil) args = { :key => stat_key, :ukey => user_key, - :count => count } + :count => Integer(count) } args[:t] = timestamp unless timestamp.nil? Common::send_to_stathat(Common::CLASSIC_COUNT_URL, args) end @@ -57,7 +98,7 @@ def post_count(stat_key, user_key, count, timestamp=nil) def post_value(stat_key, user_key, value, timestamp=nil) args = { :key => stat_key, :ukey => user_key, - :value => value } + :value => Float(value) } args[:t] = timestamp unless timestamp.nil? Common::send_to_stathat(Common::CLASSIC_VALUE_URL, args) end @@ -66,20 +107,25 @@ def post_value(stat_key, user_key, value, timestamp=nil) class API class << self + attr_accessor :max_batch + attr_accessor :pool_size + attr_accessor :max_queue_size + attr_accessor :batch_sleep_seconds + def ez_post_value(stat_name, ezkey, value, timestamp=nil, &block) - Reporter.instance.ez_post_value(stat_name, ezkey, value, timestamp, block) + Reporter.instance.ez_post_value(stat_name, ezkey, Float(value), timestamp, block) end def ez_post_count(stat_name, ezkey, count, timestamp=nil, &block) - Reporter.instance.ez_post_count(stat_name, ezkey, count, timestamp, block) + Reporter.instance.ez_post_count(stat_name, ezkey, Integer(count), timestamp, block) end def post_count(stat_key, user_key, count, timestamp=nil, &block) - Reporter.instance.post_count(stat_key, user_key, count, timestamp, block) + Reporter.instance.post_count(stat_key, user_key, Integer(count), timestamp, block) end def post_value(stat_key, user_key, value, timestamp=nil, &block) - Reporter.instance.post_value(stat_key, user_key, value, timestamp, block) + Reporter.instance.post_value(stat_key, user_key, Float(value), timestamp, block) end end end @@ -134,22 +180,51 @@ def ez_post_count(stat_name, ezkey, count, timestamp, cb) def run_pool @runlock.synchronize { @running = true } @pool = [] - 5.times do |i| + (API.pool_size || 5).times do |i| @pool[i] = Thread.new do while true do - point = @que.pop - # XXX check for error? - begin - resp = Common::send_to_stathat(point[:url], point[:args]) - if point[:cb] - point[:cb].call(resp) + points = [@que.pop] + while points.size < (API.max_batch || 1) && @que.length > 0 + points << @que.pop(true) rescue ThreadError + end + + groups = points.group_by{|point| point[:args][:ezkey]} + groups.each do |ezkey, batch| + if ezkey.nil? + batch.each do |point| + # XXX check for error? + begin + resp = Common::send_to_stathat(point[:url], point[:args]) + if point[:cb] + point[:cb].call(resp) + end + rescue + puts "Exception in StatHat encoding or callback #{$!.inspect}" + end + end + + else + batch_args = batch.map{|point| point[:args]} + batch_args.each{|args| args.delete(:ezkey)} + + begin + resp = Common::send_ez_batch_to_stathat(batch_args, ezkey) + batch.each do |point| + if point[:cb] + point[:cb].call(resp) + end + end + rescue + puts "Exception in StatHat encoding or callback #{$!.inspect}" end - rescue - pp $! + end end + @runlock.synchronize { break unless @running } + + sleep API.batch_sleep_seconds unless API.batch_sleep_seconds.nil? end end end @@ -166,6 +241,10 @@ def stop_pool() def enqueue(url, args, cb=nil) return false unless @running + if API.max_queue_size && @que.length > API.max_queue_size + puts "Dropping StatHat queue" + @que.clear + end point = {:url => url, :args => args, :cb => cb} @que << point true @@ -173,29 +252,34 @@ def enqueue(url, args, cb=nil) end class Response - def initialize(body) - @body = body + def initialize(resp) + @resp = resp @parsed = nil end + def success? + return valid? && (@resp.body.nil? || msg == "ok") + end + def valid? - return status == 200 + return @resp.body.nil? ? @resp.kind_of?(Net::HTTPSuccess) : status == 200 end def status parse - return @parsed['status'] + return @resp.body.nil? ? @resp.status : @parsed['status'] end def msg parse - return @parsed['msg'] + return @resp.body.nil? ? nil : @parsed['msg'] end private def parse return unless @parsed.nil? - @parsed = JSON.parse(@body) + return if @resp.body.nil? + @parsed = JSON.parse(@resp.body) end end end diff --git a/test/test_stathat.rb b/test/test_stathat.rb index 76140d9..6445851 100644 --- a/test/test_stathat.rb +++ b/test/test_stathat.rb @@ -2,68 +2,123 @@ class TestStathat < MiniTest::Unit::TestCase + def setup + StatHat::API.pool_size = 1 + StatHat::API.max_batch = 2 + end def test_ez_value - StatHat::API.ez_post_value("test ez value stat", "test@stathat.com", 0.92) do |resp| - assert(resp.valid?, "response was invalid") - assert_equal(resp.msg, "ok", "message should be 'ok'") - assert_equal(resp.status, 200, "status should be 200") + r = wait_for_resp do |cb| + StatHat::API.ez_post_value("test ez value stat", "test@stathat.com", 0.92, &cb) end - sleep(1) + assert_success(r) end def test_ez_count - StatHat::API.ez_post_value("test ez count stat", "test@stathat.com", 12) do |r| - assert(r.valid?, "response was invalid") - assert_equal(r.msg, "ok", "message should be 'ok'") - assert_equal(r.status, 200, "status should be 200") + r = wait_for_resp do |cb| + StatHat::API.ez_post_count("test ez count stat", "test@stathat.com", 12, &cb) end - sleep(1) + assert_success(r) end def test_classic_count_bad_keys - StatHat::API.post_count("XXXXXXXX", "YYYYYYYY", 12) do |r| - assert_equal(r.valid?, false, "response was valid") - assert_equal(r.msg, "invalid keys", "incorrect error message") - assert_equal(r.status, 500, "incorrect status code") + r = wait_for_resp do |cb| + StatHat::API.post_count("XXXXXXXX", "YYYYYYYY", 12, &cb) end - sleep(1) + assert_failure(r) end def test_classic_value_bad_keys - StatHat::API.post_value("ZZZZZZZZ", "YYYYYYYYY", 0.92) do |r| - assert_equal(r.valid?, false, "response was valid") - assert_equal(r.msg, "invalid keys", "incorrect error message") - assert_equal(r.status, 500, "incorrect status code") + r = wait_for_resp do |cb| + StatHat::API.post_value("ZZZZZZZZ", "YYYYYYYYY", 0.92, &cb) end - sleep(1) + assert_failure(r) end def test_ez_value_sync - resp = StatHat::SyncAPI.ez_post_value("test ez value stat", "test@stathat.com", 0.92) - assert(resp.valid?, "response was invalid") - assert_equal(resp.msg, "ok", "message should be 'ok'") - assert_equal(resp.status, 200, "status should be 200") + r = StatHat::SyncAPI.ez_post_value("test ez value stat", "test@stathat.com", 0.92) + assert_success(r) end def test_ez_count_sync - resp = StatHat::SyncAPI.ez_post_value("test ez count stat", "test@stathat.com", 12) - assert(resp.valid?, "response was invalid") - assert_equal(resp.msg, "ok", "message should be 'ok'") - assert_equal(resp.status, 200, "status should be 200") + r = StatHat::SyncAPI.ez_post_count("test ez count stat", "test@stathat.com", 12) + assert_success(r) end def test_classic_count_bad_keys_sync r = StatHat::SyncAPI.post_count("XXXXXXXX", "YYYYYYYY", 12) - assert_equal(r.valid?, false, "response was valid") - assert_equal(r.msg, "invalid keys", "incorrect error message") - assert_equal(r.status, 500, "incorrect status code") + assert_failure(r) end def test_classic_value_bad_keys_sync r = StatHat::SyncAPI.post_value("ZZZZZZZZ", "YYYYYYYYY", 0.92) + assert_failure(r) + end + + def test_string_value + r = wait_for_resp do |cb| + StatHat::API.ez_post_value("ZZZZZZZZ", "YYYYYYYYY", "0.92", &cb) + end + assert_success(r) + + assert_raises(ArgumentError) do + StatHat::API.ez_post_count("XXXXXXXX", "YYYYYYYY", "1.0") + end + end + + def test_ez_batch + final = wait_for_resp do |cb| + StatHat::API.post_count("XXXXXXXX", "YYYYYYYY", 11) do |r| + assert_failure(r) + end + + StatHat::API.ez_post_value("test ez value stat", "test@stathat.com", 0.92) do |r| + assert_success(r) + end + + StatHat::API.ez_post_count("test ez count stat", "test@stathat.com", 12) do |r| + assert_success(r) + end + + StatHat::API.ez_post_count("test ez count stat2", "test@stathat.com", 13, &cb) + end + + assert_success(final) + end + + private + + def assert_success(r) + assert(r.valid?, "response was invalid") + assert(r.success?, "response should be successful #{r.inspect}") + end + + def assert_failure(r) assert_equal(r.valid?, false, "response was valid") assert_equal(r.msg, "invalid keys", "incorrect error message") assert_equal(r.status, 500, "incorrect status code") + refute(r.success?, "response should not be successful #{r.inspect}") + end + + def wait_for_resp + m = Mutex.new + cv = ConditionVariable.new + resp = nil + + cb = lambda do |r| + resp = r + m.synchronize do + cv.signal + end + end + + m.synchronize do + yield cb + start = Time.now + assert cv.wait(m, 60) + assert Time.now - start < 50 + end + + resp end end