Class: Mixpanel::Consumer

Inherits:
Object
  • Object
show all
Defined in:
lib/mixpanel-ruby/consumer.rb

Overview

A Consumer receives messages from a Mixpanel::Tracker, and sends them elsewhere- probably to Mixpanel's analytics services, but can also enqueue them for later processing, log them to a file, or do whatever else you might find useful.

You can provide your own consumer to your Mixpanel::Trackers, either by passing in an argument with a #send! method when you construct the tracker, or just passing a block to Mixpanel::Tracker.new

tracker = Mixpanel::Tracker.new(YOUR_MIXPANEL_TOKEN) do |type, message|
   # type will be one of :event, :profile_update or :import
   @kestrel.set(ANALYTICS_QUEUE, [type, message].to_json)
end

IMPORTANT SECURITY NOTE: Always pass credentials to the Consumer or Tracker constructor (credentials: ...). Credentials are stored as instance variables and used only for HTTP Basic Auth headers - they never appear in message JSON, preventing accidental credential leakage in logs or queue storage.

You can also instantiate the library consumers yourself, and use them wherever you would like. For example, the working that consumes the above queue might work like this:

mixpanel = Mixpanel::Consumer
while true
    message_json = @kestrel.get(ANALYTICS_QUEUE)
    mixpanel.send!(*JSON.load(message_json))
end

Mixpanel::Consumer is the default consumer. It sends each message, as the message is recieved, directly to Mixpanel.

Instance Method Summary collapse

Constructor Details

#initialize(events_endpoint = nil, update_endpoint = nil, groups_endpoint = nil, import_endpoint = nil, credentials: nil) ⇒ Consumer

Create a Mixpanel::Consumer. If you provide endpoint arguments, they will be used instead of the default Mixpanel endpoints. This can be useful for proxying, debugging, or if you prefer not to use SSL for your events.

Parameters:

  • credentials (ServiceAccountCredentials, nil) (defaults to: nil) —

    Service account credentials for authentication. Credentials are only used for the /import endpoint and feature flags.



69
70
71
72
73
74
75
76
77
78
79
 
# File 'lib/mixpanel-ruby/consumer.rb', line 69

def initialize(events_endpoint=nil,
               update_endpoint=nil,
               groups_endpoint=nil,
               import_endpoint=nil,
               credentials: nil)
  @events_endpoint = events_endpoint || 'https://api.mixpanel.com/track'
  @update_endpoint = update_endpoint || 'https://api.mixpanel.com/engage'
  @groups_endpoint = groups_endpoint || 'https://api.mixpanel.com/groups'
  @import_endpoint = import_endpoint || 'https://api.mixpanel.com/import'
  @credentials = credentials
end

Instance Method Details

#request(endpoint, form_data, credentials: nil, type: nil) ⇒ Object

Request takes an endpoint HTTP or HTTPS url, and a Hash of data to post to that url. It should return a pair of

[response code, response body]

as the result of the response. Response code should be nil if the request never receives a response for some reason.

For service account authentication, pass credentials (ServiceAccountCredentials object) and type (:import) as keyword arguments. The positional parameters are preserved for backward compatibility with custom Consumer subclasses.



153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
 
# File 'lib/mixpanel-ruby/consumer.rb', line 153

def request(endpoint, form_data, credentials: nil, type: nil)
  uri = URI(endpoint)

  # Add project_id as query parameter for import endpoint with service account credentials
  if credentials && type == :import
    unless credentials.is_a?(ServiceAccountCredentials)
      raise ArgumentError, "credentials must be ServiceAccountCredentials, got #{credentials.class}"
    end

    query_params = URI.decode_www_form(uri.query || '').to_h
    query_params['project_id'] = credentials.project_id
    uri.query = URI.encode_www_form(query_params)
  end

  request = Net::HTTP::Post.new(uri.request_uri)
  request.set_form_data(form_data)

  # Use Basic Auth with service account credentials for import endpoint
  if credentials && type == :import
    request.basic_auth(credentials.username, credentials.secret)
  end

  client = Net::HTTP.new(uri.host, uri.port)
  client.use_ssl = true
  client.open_timeout = 10
  client.continue_timeout = 10
  client.read_timeout = 10
  client.ssl_timeout = 10

  Mixpanel.with_http(client)

  response = client.request(request)
  [response.code, response.body]
end

#send(type, message) ⇒ Object

This method was deprecated in release 2.0.0, please use send! instead



137
138
139
140
 
# File 'lib/mixpanel-ruby/consumer.rb', line 137

def send(type, message)
    warn '[DEPRECATION] send has been deprecated as of release 2.0.0, please use send! instead'
    send!(type, message)
end

#send!(type, message) ⇒ Object

Send the given string message to Mixpanel. Type should be one of :event, :profile_update or :import, which will determine the endpoint.

Mixpanel::Consumer#send! sends messages to Mixpanel immediately on each call. To reduce the overall bandwidth you use when communicating with Mixpanel, you can also use Mixpanel::BufferedConsumer



87
88
89
90
91
92
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
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
 
# File 'lib/mixpanel-ruby/consumer.rb', line 87

def send!(type, message)
  type = type.to_sym
  endpoint = {
    :event => @events_endpoint,
    :profile_update => @update_endpoint,
    :group_update => @groups_endpoint,
    :import => @import_endpoint,
  }[type]

  decoded_message = JSON.load(message)
  api_key = decoded_message["api_key"]

  data = Base64.encode64(decoded_message["data"].to_json).gsub("\n", '')

  form_data = {"data" => data, "verbose" => 1}

  # Only add api_key to form data if using legacy API key (not service account credentials)
  # Service account credentials use HTTP Basic Auth instead
  if api_key && !@credentials
    form_data.merge!("api_key" => api_key)
  end

  begin
    # Use keyword arguments for credentials to maintain backward compatibility
    # with custom Consumer subclasses that override request(endpoint, form_data)
    response_code, response_body =
      if @credentials && type == :import
        request(endpoint, form_data, credentials: @credentials, type: type)
      else
        request(endpoint, form_data)
      end
  rescue => e
    raise ConnectionError.new("Could not connect to Mixpanel, with error \"#{e.message}\".")
  end

  result = {}
  if response_code.to_i == 200
    begin
      result = JSON.parse(response_body.to_s)
    rescue JSON::JSONError
      raise ServerError.new("Could not interpret Mixpanel server response: '#{response_body}'")
    end
  end

  if result['status'] != 1
    raise ServerError.new("Could not write to Mixpanel, server responded with #{response_code} returning: '#{response_body}'")
  end
end