elementor / vendor_prefixed / mixpanel / lib / Producers / MixpanelBaseProducer.php

MixpanelBaseProducer.php in Elementor Website Builder – more than just a page builder 4.3.1, at vendor_prefixed/mixpanel/lib/Producers/MixpanelBaseProducer.php

189 lines 6.2 KB

Raw Download Zip

1	<?php
2
3	namespace ElementorDeps;
4
5	require_once \dirname(__FILE__) . "/../Base/MixpanelBase.php";
6	require_once \dirname(__FILE__) . "/../ConsumerStrategies/FileConsumer.php";
7	require_once \dirname(__FILE__) . "/../ConsumerStrategies/CurlConsumer.php";
8	require_once \dirname(__FILE__) . "/../ConsumerStrategies/SocketConsumer.php";
9	if (!\function_exists('json_encode')) {
10	throw new \Exception('The JSON PHP extension is required.');
11	}
12	/**
13	* Provides some base methods for use by a message Producer
14	*/
15	abstract class Producers_MixpanelBaseProducer extends Base_MixpanelBase
16	{
17	/**
18	* @var string a token associated to a Mixpanel project
19	*/
20	protected $_token;
21	/**
22	* @var array a queue to hold messages in memory before flushing in batches
23	*/
24	private $_queue = array();
25	/**
26	* @var ConsumerStrategies_AbstractConsumer the consumer to use when flushing messages
27	*/
28	private $_consumer = null;
29	/**
30	* @var array The list of available consumers
31	*/
32	private $_consumers = array("file" => "ConsumerStrategies_FileConsumer", "curl" => "ConsumerStrategies_CurlConsumer", "socket" => "ConsumerStrategies_SocketConsumer");
33	/**
34	* If the queue reaches this size we'll auto-flush to prevent out of memory errors
35	* @var int
36	*/
37	protected $_max_queue_size = 1000;
38	/**
39	* Creates a new MixpanelBaseProducer, assings Mixpanel project token, registers custom Consumers, and instantiates
40	* the desired consumer
41	* @param $token
42	* @param array $options
43	*/
44	public function __construct($token, $options = array())
45	{
46	parent::__construct($options);
47	// register any customer consumers
48	if (isset($options["consumers"])) {
49	$this->_consumers = \array_merge($this->_consumers, $options['consumers']);
50	}
51	// set max queue size
52	if (isset($options["max_queue_size"])) {
53	$this->_max_queue_size = $options['max_queue_size'];
54	}
55	// associate token
56	$this->_token = $token;
57	if ($this->_debug()) {
58	$this->_log("Using token: " . $this->_token);
59	}
60	// instantiate the chosen consumer
61	$this->_consumer = $this->_getConsumer();
62	}
63	/**
64	* Flush the queue when we destruct the client with retries
65	*/
66	public function __destruct()
67	{
68	$attempts = 0;
69	$max_attempts = 10;
70	$success = \false;
71	while (!$success && $attempts < $max_attempts) {
72	if ($this->_debug()) {
73	$this->_log("destruct flush attempt #" . ($attempts + 1));
74	}
75	$success = $this->flush();
76	$attempts++;
77	}
78	}
79	/**
80	* Iterate the queue and write in batches using the instantiated Consumer Strategy
81	* @param int $desired_batch_size
82	* @return bool whether or not the flush was successful
83	*/
84	public function flush($desired_batch_size = 50)
85	{
86	$queue_size = \count($this->_queue);
87	$succeeded = \true;
88	$num_threads = $this->_consumer->getNumThreads();
89	if ($this->_debug()) {
90	$this->_log("Flush called - queue size: " . $queue_size);
91	}
92	while ($queue_size > 0 && $succeeded) {
93	$batch_size = \min(array($queue_size, $desired_batch_size * $num_threads, $this->_options['max_batch_size'] * $num_threads));
94	$batch = \array_splice($this->_queue, 0, $batch_size);
95	$succeeded = $this->_persist($batch);
96	if (!$succeeded) {
97	if ($this->_debug()) {
98	$this->_log("Batch consumption failed!");
99	}
100	$this->_queue = \array_merge($batch, $this->_queue);
101	if ($this->_debug()) {
102	$this->_log("added batch back to queue, queue size is now {$queue_size}");
103	}
104	}
105	$queue_size = \count($this->_queue);
106	if ($this->_debug()) {
107	$this->_log("Batch of {$batch_size} consumed, queue size is now {$queue_size}");
108	}
109	}
110	return $succeeded;
111	}
112	/**
113	* Empties the queue without persisting any of the messages
114	*/
115	public function reset()
116	{
117	$this->_queue = array();
118	}
119	/**
120	* Returns the in-memory queue
121	* @return array
122	*/
123	public function getQueue()
124	{
125	return $this->_queue;
126	}
127	/**
128	* Returns the current Mixpanel project token
129	* @return string
130	*/
131	public function getToken()
132	{
133	return $this->_token;
134	}
135	/**
136	* Given a strategy type, return a new PersistenceStrategy object
137	* @return ConsumerStrategies_AbstractConsumer
138	*/
139	protected function _getConsumer()
140	{
141	$key = $this->_options['consumer'];
142	$Strategy = $this->_consumers[$key];
143	if ($this->_debug()) {
144	$this->_log("Using consumer: " . $key . " -> " . $Strategy);
145	}
146	$this->_options['endpoint'] = $this->_getEndpoint();
147	return new $Strategy($this->_options);
148	}
149	/**
150	* Add an array representing a message to be sent to Mixpanel to a queue.
151	* @param array $message
152	*/
153	public function enqueue($message = array())
154	{
155	\array_push($this->_queue, $message);
156	// force a flush if we've reached our threshold
157	if (\count($this->_queue) > $this->_max_queue_size) {
158	$this->flush();
159	}
160	if ($this->_debug()) {
161	$this->_log("Queued message: " . \json_encode($message));
162	}
163	}
164	/**
165	* Add an array representing a list of messages to be sent to Mixpanel to a queue.
166	* @param array $messages
167	*/
168	public function enqueueAll($messages = array())
169	{
170	foreach ($messages as $message) {
171	$this->enqueue($message);
172	}
173	}
174	/**
175	* Given an array of messages, persist it with the instantiated Persistence Strategy
176	* @param $message
177	* @return mixed
178	*/
179	protected function _persist($message)
180	{
181	return $this->_consumer->persist($message);
182	}
183	/**
184	* Return the endpoint that should be used by a consumer that consumes messages produced by this producer.
185	* @return string
186	*/
187	abstract function _getEndpoint();
188	}
189