Class: Aptabase::Client

Inherits:
Object
  • Object
show all
Defined in:
lib/aptabase/client.rb

Overview

Aptabase analytics client.

‘track` is synchronous and cheap: it validates the event and pushes it onto an in-memory queue. A background worker thread flushes the queue every `flush_interval` seconds, or as soon as `max_batch_size` events are pending. Failed batches are re-queued and retried on the next flush.

Constant Summary collapse

HOSTS = 
{
  "EU" => "https://eu.aptabase.com",
  "US" => "https://us.aptabase.com"
}.freeze
MAX_BATCH_SIZE = 
25

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(app_key, app_version: "1.0.0", is_debug: false, max_batch_size: MAX_BATCH_SIZE, flush_interval: 10.0, timeout: 30.0, base_url: nil, logger: nil) ⇒ Client

Returns a new instance of Client.

Raises:



36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
 
# File 'lib/aptabase/client.rb', line 36

def initialize(app_key, app_version: "1.0.0", is_debug: false,
               max_batch_size: MAX_BATCH_SIZE, flush_interval: 10.0,
               timeout: 30.0, base_url: nil, logger: nil)
  validate_app_key!(app_key)
  raise ConfigurationError, "Maximum batch size is #{MAX_BATCH_SIZE} events" if max_batch_size > MAX_BATCH_SIZE
  raise ConfigurationError, "max_batch_size must be at least 1" if max_batch_size < 1

  @app_key = app_key
  @base_url = base_url || default_base_url(app_key)
  @system_props = SystemProperties.new(app_version: app_version, is_debug: is_debug)
  @max_batch_size = max_batch_size
  @flush_interval = flush_interval
  @logger = logger || Logger.new($stderr, progname: "aptabase", level: Logger::INFO)
  @transport = Transport.new(base_url: @base_url, app_key: app_key, timeout: timeout)
  @session = Session.new

  @queue = []
  @queue_mutex = Mutex.new
  @flush_mutex = Mutex.new
  @wake = ConditionVariable.new
  @worker = nil
  @stopping = false
  @pid = Process.pid
end

Instance Attribute Details

#app_key ⇒ Object (readonly)

Returns the value of attribute app_key.



20
21
22
 
# File 'lib/aptabase/client.rb', line 20

def app_key
  @app_key
end

#base_url ⇒ Object (readonly)

Returns the value of attribute base_url.



20
21
22
 
# File 'lib/aptabase/client.rb', line 20

def base_url
  @base_url
end

#logger ⇒ Object (readonly)

Returns the value of attribute logger.



20
21
22
 
# File 'lib/aptabase/client.rb', line 20

def logger
  @logger
end

Class Method Details

.start(app_key, **options) ⇒ Object

Block form: yields the client and guarantees stop/flush on exit.

Aptabase::Client.start("A-EU-123") { |client| client.track("event") }



25
26
27
28
29
30
31
32
33
34
 
# File 'lib/aptabase/client.rb', line 25

def self.start(app_key, **options)
  client = new(app_key, **options)
  return client unless block_given?

  begin
    yield client
  ensure
    client.stop
  end
end

Instance Method Details

#flush ⇒ Object

Synchronously send all queued events in batches of ‘max_batch_size`. On failure the unsent events stay queued for the next flush.



83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
 
# File 'lib/aptabase/client.rb', line 83

def flush
  @flush_mutex.synchronize do
    loop do
      batch = @queue_mutex.synchronize { @queue.shift(@max_batch_size) }
      break if batch.empty?

      begin
        @transport.post_events(batch.map { |event| event.to_h(@system_props) })
        logger.debug { "[aptabase] sent #{batch.size} event(s)" }
      rescue NetworkError => e
        logger.warn("[aptabase] failed to send #{batch.size} event(s), will retry: #{e.message}")
        @queue_mutex.synchronize { @queue.unshift(*batch) }
        break
      end
    end
  end
  nil
end

#pending_count ⇒ Object

Number of events queued and not yet delivered.



121
122
123
 
# File 'lib/aptabase/client.rb', line 121

def pending_count
  @queue_mutex.synchronize { @queue.size }
end

#stop ⇒ Object

Stop the background worker and flush any remaining events. The client can keep being used afterwards; tracking restarts the worker.



104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
 
# File 'lib/aptabase/client.rb', line 104

def stop
  worker = nil
  @queue_mutex.synchronize do
    @stopping = true
    worker = @worker
    @wake.broadcast
  end
  worker.join(@flush_interval + 1) if worker && worker != Thread.current
  flush
  @queue_mutex.synchronize do
    @worker = nil
    @stopping = false
  end
  nil
end

#track(event_name, props = nil) ⇒ Object

Track an analytics event.

client.track("app_started")
client.track("purchase", { "product_id" => "abc", "price" => 29.99 })

Raises:



65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
 
# File 'lib/aptabase/client.rb', line 65

def track(event_name, props = nil)
  unless event_name.is_a?(String) && !event_name.empty?
    raise ValidationError, "Event name is required and must be a non-empty string"
  end
  raise ValidationError, "Event properties must be a Hash" if !props.nil? && !props.is_a?(Hash)

  event = Event.new(name: event_name, session_id: @session.id, props: props)

  @queue_mutex.synchronize do
    ensure_worker
    @queue << event
    @wake.signal if @queue.size >= @max_batch_size
  end
  nil
end