Class: GetFluxly::Client

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

Overview

Synchronous GetFluxly client.

Batches events with retry, jitter, and X-Idempotency-Key. A batch goes out when the queue reaches flush_at, when the background flusher thread fires (every flush_interval seconds once the first event is enqueued; <= 0 disables it), on an explicit flush, and at_exit.

Constant Summary collapse

DEFAULT_API_HOST = 
"https://api.getfluxly.com"
EVENTS_BATCH_PATH = 
"/v1/events/batch"
ALIAS_PATH = 
"/v1/identify/alias"

Instance Method Summary collapse

Constructor Details

#initialize(token:, api_host: DEFAULT_API_HOST, flush_at: 20, flush_interval: 5.0, max_retries: 2, timeout: 5.0, max_queue_size: 1000, register_atexit: true) ⇒ Client

Returns a new instance of Client.

Raises:



16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
 
# File 'lib/getfluxly/client.rb', line 16

def initialize(
  token:,
  api_host: DEFAULT_API_HOST,
  flush_at: 20,
  flush_interval: 5.0,
  max_retries: 2,
  timeout: 5.0,
  max_queue_size: 1000,
  register_atexit: true
)
  raise GetFluxly::Error.new("token is required", code: "validation_error") if token.nil? || token.empty?

  @token = token
  @api_host = api_host
  @flush_at = flush_at
  @flush_interval = flush_interval

  @batch = Batch.new(max_size: max_queue_size)
  @http = Http.new(token: token, api_host: api_host, timeout: timeout, max_retries: max_retries)
  @shutdown = false
  @shutdown_mutex = Mutex.new
  @flusher = nil
  @flusher_pid = nil
  @wake_mutex = Mutex.new
  @wake = ConditionVariable.new
  @woken = false

  return unless register_atexit

  at_exit do
    shutdown
  rescue StandardError
    # at_exit: swallow so interpreter shutdown is clean
  end
end

Instance Method Details

#alias(user_id:, anonymous_id: nil, previous_id: nil, request_id: nil) ⇒ Object 



93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
 
# File 'lib/getfluxly/client.rb', line 93

def alias(user_id:, anonymous_id: nil, previous_id: nil, request_id: nil)
  Identity.require_alias!(user_id: user_id, anonymous_id: anonymous_id, previous_id: previous_id)

  body = { "user_id" => user_id }
  body["anonymous_id"] = anonymous_id if anonymous_id
  body["previous_id"] = previous_id if previous_id

  response = @http.post(
    ALIAS_PATH,
    body,
    idempotency_key: Http.generate_idempotency_key,
    request_id: request_id
  )

  alias_obj = response["alias"]
  if alias_obj.nil? || alias_obj == {}
    raise GetFluxly::Error.new(
      "alias response did not include alias data",
      code: "invalid_response",
      retryable: true,
      details: response.is_a?(Hash) ? response : {}
    )
  end

  alias_obj
end

#flush ⇒ Object 



120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
 
# File 'lib/getfluxly/client.rb', line 120

def flush
  result = Batch.empty_flush_result
  loop do
    batch = @batch.drain(@flush_at)
    break if batch.empty?

    begin
      response = @http.post(
        EVENTS_BATCH_PATH,
        { "events" => batch },
        idempotency_key: Http.generate_idempotency_key
      )
    rescue GetFluxly::Error => e
      @batch.requeue_front(batch) if e.retryable
      raise
    end

    result += Batch::FlushResult.new(
      (response["accepted"] || batch.size).to_i,
      (response["rejected"] || 0).to_i,
      1,
      Array(response["errors"])
    )
  end
  result
end

#identify(anonymous_id: nil, external_id: nil, user_id: nil, traits: nil) ⇒ Object 



71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
 
# File 'lib/getfluxly/client.rb', line 71

def identify(anonymous_id: nil, external_id: nil, user_id: nil, traits: nil)
  Identity.require_one_id!(
    anonymous_id: anonymous_id,
    external_id: external_id,
    user_id: user_id
  )

  # FIXME: identify is not wired end to end. The backend treats only
  # event == "identify" as an identify, but this sends "$identify", so
  # identity stitching never runs. A plain rename will not fix it: the
  # backend identify path requires BOTH anonymous_id and external_id,
  # which a server identify usually lacks. Needs a server side identify
  # contract first; until then use track() (see the SDK docs callout).
  payload = { "event" => "$identify", "message_id" => Http.generate_message_id }
  payload["anonymous_id"] = anonymous_id if anonymous_id
  payload["external_id"] = external_id if external_id
  payload["user_id"] = user_id if user_id
  payload["traits"] = traits if traits

  enqueue(payload)
end

#shutdown ⇒ Object Also known as: close 



147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
 
# File 'lib/getfluxly/client.rb', line 147

def shutdown
  flusher = nil
  @shutdown_mutex.synchronize do
    return Batch.empty_flush_result if @shutdown

    @shutdown = true
    flusher = @flusher
    @flusher = nil
  end
  # Wake the flusher so it exits promptly, and wait for it so a
  # scheduled flush can't race the final flush below. The join is
  # bounded by one in-flight HTTP round-trip (timeout x retries).
  # @woken makes the wake sticky: a broadcast that lands before the
  # flusher reaches wait would otherwise be lost.
  @wake_mutex.synchronize do
    @woken = true
    @wake.broadcast
  end
  flusher&.join(30) unless flusher == Thread.current
  flush
end

#track(event, anonymous_id: nil, external_id: nil, user_id: nil, properties: nil, timestamp: nil, context: nil) ⇒ Object 



52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
 
# File 'lib/getfluxly/client.rb', line 52

def track(event, anonymous_id: nil, external_id: nil, user_id: nil,
          properties: nil, timestamp: nil, context: nil)
  Identity.require_one_id!(
    anonymous_id: anonymous_id,
    external_id: external_id,
    user_id: user_id
  )

  payload = { "event" => event, "message_id" => Http.generate_message_id }
  payload["anonymous_id"] = anonymous_id if anonymous_id
  payload["external_id"] = external_id if external_id
  payload["user_id"] = user_id if user_id
  payload["properties"] = properties if properties
  payload["timestamp"] = timestamp if timestamp
  payload["context"] = context if context

  enqueue(payload)
end