3750646379

Build Type

push

github

Committed by Miranda Wilson

Commit Message

Merge branch 'release/0.6.1'

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

515 of 583 relevant lines covered (88.34%)

11.07 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.21

/src/Emitters/CurlEmitter.php

<?php
/*
    CurlEmitter.php

    Copyright (c) 2014-2022 Snowplow Analytics Ltd. All rights reserved.

    This program is licensed to you under the Apache License Version 2.0,
    and you may not use this file except in compliance with the Apache License
    Version 2.0. You may obtain a copy of the Apache License Version 2.0 at
    http://www.apache.org/licenses/LICENSE-2.0.

    Unless required by applicable law or agreed to in writing,
    software distributed under the Apache License Version 2.0 is distributed on
    an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
    express or implied. See the Apache License Version 2.0 for the specific
    language governing permissions and limitations there under.

    Authors: Joshua Beemster
    License: Apache License Version 2.0
*/

namespace Snowplow\Tracker\Emitters;
use Snowplow\Tracker\Emitter;

class CurlEmitter extends Emitter {

    // Emitter Parameters

    private $type;
    private $url;
    private $debug;
    private $requests_results;
    private $debug_payloads;

    // Curl Specific Parameters

    private $final_payload_buffer = array();
    private $curl_buffer = array();
    private $curl_limit;
    private $rolling_window;
    private $curl_timeout;

    /**
     * Constructs an async curl emitter.
     *
     * @param string $uri - Collector URI
     * @param string|null $protocol - What protocol we are using for the collector
     * @param string|null $type - The type of request we will be making to the collector
     * @param int|null $buffer_size - Emitter buffer size
     * @param bool $debug - Debug mode
     * @param int|null $curl_timeout - Maximum time the request is allowed to take, in seconds
     */
    public function __construct($uri, $protocol = NULL, $type = NULL, $buffer_size = NULL, $debug = false, $curl_timeout = NULL) {
        $this->type           = $this->getRequestType($type);
        $this->url            = $this->getCollectorUrl($this->type, $uri, $protocol);
        $this->curl_limit     = $this->type == "POST" ? self::CURL_AMOUNT_POST : self::CURL_AMOUNT_GET;
        $this->rolling_window = $this->type == "POST" ? self::CURL_WINDOW_POST : self::CURL_WINDOW_GET;
        $this->curl_timeout   = $curl_timeout;

        // If debug is on create a requests_results array
        if ($debug === true) {
            $this->debug = true;
            $this->debug_payloads = array();
            $this->requests_results = array();
        }
        else {
            $this->debug = false;
        }
        $buffer = $buffer_size == NULL ? self::CURL_BUFFER : $buffer_size;
        $this->setup("curl", $debug, $buffer);
    }

    /**
     * Push event buffers into curls and store them
     * - Wait until we have an allotted amount of curls before executing
     * - Or force the execution of the curl emitter
     *
     * @param $buffer - An array of events we are going to convert into curl resources
     * @param bool $curl_send - Whether or not we are going to send the buffered curl
     *                          objects before we reach the limit
     * @return bool|string - Either true or an error string
     */
    public function send($buffer, $curl_send = false) {
        $type = $this->type;
        $debug = $this->debug;

        // If the sent buffer contains events...
        if (count($buffer) > 0) {
            if ($type == "POST") {
                array_push($this->final_payload_buffer, $buffer);
            }
            else {
                foreach ($buffer as $event) {
                    array_push($this->final_payload_buffer, $event);
                }
            }
        }

        if (count($this->final_payload_buffer) >= $this->curl_limit) {
            return $this->mkCurlRequests($this->final_payload_buffer, $type, $debug);
        }
        else if ($curl_send === true) {
            if (count($this->final_payload_buffer) > 0) {
                return $this->mkCurlRequests($this->final_payload_buffer, $type, $debug);
            }
            else {
                return "Error: No curls to send.";
            }
        }
        return "Error: Still adding to the curl buffer; count ".count($this->final_payload_buffer)." - limit ".$this->curl_limit;
    }

    /**
     * Makes curl requests from payloads to be sent.
     *
     * @param array $payloads - Array of payloads to be sent
     * @param string $type - Type of requests to be made
     * @param bool $debug - If debug is on or not
     * @return bool|string - Returns true or a string of errors
     */
    private function mkCurlRequests($payloads, $type, $debug) {
        // Empty the global buffer.
        $this->final_payload_buffer = array();

        if ($type == 'POST') {
            foreach ($payloads as $buffer) {
                $payload = $this->getPostRequest($this->batchUpdateStm($buffer));
                $curl = $this->getCurlRequest($payload, $type);
                array_push($this->curl_buffer, $curl);

                // If debug is on; store the handle and the payload
                if ($debug) {
                    array_push($this->debug_payloads, array("handle" => $curl, "payload" => $payload));
                }
            }
        }
        else {
            foreach ($payloads as $event) {
                $payload = http_build_query($this->updateStm($event));
                $curl = $this->getCurlRequest($payload, $type);
                array_push($this->curl_buffer, $curl);
            }
        }
        return $this->rollingCurl($this->curl_buffer, $debug);
    }

    /**
     * Asynchronously sends curl requests.
     * - Prevents the queue from being held up by
     *   starting new requests as soon as any are done.
     *
     * @param array $curls - Array of curls to be sent
     * @param bool $debug - If Debug is on or not
     * @return bool|string - Returns true or a string of errors
     */
    private function rollingCurl($curls, $debug) {
        // Empty the global buffer.
        $this->curl_buffer = array();

        // Create a results string to log potential errors
        $res = true;
        $res_ = "";

        // Rolling Window == How many requests concurrently.
        $rolling_window = $this->rolling_window;
        $master = curl_multi_init();

        // Add cUrls to handler.
        $limit = ($rolling_window <= count($curls)) ? $rolling_window : count($curls);
        for ($i = 0; $i < $limit; $i++) {
            $ch = $curls[$i];
            curl_multi_add_handle($master, $ch);
        }

        // Execute the rolling curl
        $running = 0;
        do {
            do {
                $execrun = curl_multi_exec($master, $running);
            } while ($execrun == CURLM_CALL_MULTI_PERFORM);

            while ($done = curl_multi_info_read($master)) {
                if ($debug) {
                    $info = curl_getinfo($done['handle']);
                    if ($info['http_code'] != 200) {
                        $res_.= "Error: Curl Request Failed with response code - ".$info['http_code']."\n";
                    }

                    if ($this->type == "POST") {
                        $data = $this->getDebugData($done);
                    }
                    else {
                        $data = $info['url'];
                    }
                    $this->storeRequestResults($info['http_code'], $data);
                }

                // If there are still curls in the queue add them to the  multi curl handler
                if ($i < count($curls)) {
                    $ch = $curls[$i++];
                    curl_multi_add_handle($master, $ch);
                }
                curl_multi_remove_handle($master, $done['handle']);
                curl_close($done['handle']);
            }
        } while ($running);
        curl_multi_close($master);

        if ($res_ != "") {
            $res = $res_;
        }

        return $res;
    }

    /**
     * Generates and returns a curl resource
     *
     * @param string $payload - Data included in request
     * @param string $type - Type of request to be made
     * @return resource
     */
    private function getCurlRequest($payload, $type) {
        $ch = curl_init($this->url);
        if ($type == "POST") {
            $header = array(
                'Content-Type: '.self::POST_CONTENT_TYPE,
                'Accept: '.self::POST_ACCEPT,
                'Content-Length: '.strlen($payload));
            curl_setopt($ch, CURLOPT_CUSTOMREQUEST, "POST");
            curl_setopt($ch, CURLOPT_POSTFIELDS, $payload);
            curl_setopt($ch, CURLOPT_HTTPHEADER, $header);
        }
        else {
            curl_setopt($ch, CURLOPT_CUSTOMREQUEST, "GET");
            curl_setopt($ch, CURLOPT_URL, $this->url."?".$payload);
        }
        curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
        if ($this->curl_timeout != NULL) {
            curl_setopt($ch, CURLOPT_TIMEOUT, $this->curl_timeout);
        }
        return $ch;
    }

    /**
     * Compiles events from buffer into a single string.
     *
     * @param array $buffer
     * @return string - Returns a json_encoded string with all of the events to be sent.
     */
    private function getPostRequest($buffer) {
        $data = json_encode(array("schema" => self::POST_REQ_SCHEMA, "data" => $buffer));
        return $data;
    }

    /**
     * Disables debug mode for the emitter
     * - If deleteLocal is true it will also empty
     *   the local cache of stored request codes and
     *   the associated payloads.
     * - Will then trigger a function in the base
     *   emitter class to clean out the physical
     *   debug records.
     *
     * @param bool $deleteLocal - Empty results array
     */
    public function turnOffDebug($deleteLocal) {
        $this->debug = false;
        if ($deleteLocal) {
            $this->requests_results = array();
        }

        // Switch Debug off in Base Emitter Class
        $this->debugSwitch($deleteLocal);
    }

    // Curl Return Functions

    /**
     * Returns the collector url
     *
     * @return string
     */
    public function returnUrl() {
        return $this->url;
    }

    /**
     * Returns the type of Request we will be making
     *
     * @return null|string
     */
    public function returnType() {
        return $this->type;
    }

    /**
     * Returns the current array of curls we have
     *
     * @return array
     */
    public function returnCurlBuffer() {
        return $this->curl_buffer;
    }

    /**
     * Returns the amount of curls we need before sending
     *
     * @return int
     */
    public function returnCurlAmount() {
        return $this->curl_limit;
    }

    /**
     * Returns the amount of simultaneous curls we send
     *
     * @return int
     */
    public function returnRollingWindow() {
        return $this->rolling_window;
    }

    // Debug Functions

    /**
     * Returns the array of stored results from a request
     *
     * @return array
     */
    public function returnRequestResults() {
        return $this->requests_results;
    }

    /**
     * Returns the payload associated with the finished curl
     *
     * @param array $done - The array made by finishing a curl
     * @return string - Payload
     */
    private function getDebugData($done) {
        $data = "";
        for ($i = 0; $i < count($this->debug_payloads); $i++) {
            $debug_payload = $this->debug_payloads[$i];
            if ($debug_payload["handle"] == $done['handle']) {
                $data = $debug_payload["payload"];

                // Delete the curl handle & payload from storage
                unset($this->debug_payloads[$i]);

                // Re-index the array after the deletion
                $this->debug_payloads = array_values($this->debug_payloads);
            }
        }
        return $data;
    }

    /**
     * Stores curl request response code
     *
     * @param $code
     * @param $data
     */
    private function storeRequestResults($code, $data) {
        $array["code"] = $code;
        if ($this->type == "GET") {
            $temp = substr($data,(strpos($data,"?")+1),-1);
            parse_str($temp, $output);
            $array["data"] = json_encode($output);
        }
        else {
            $array["data"] = $data;
        }
        array_push($this->requests_results, $array);
    }
}

1	<?php
2	/*
3	CurlEmitter.php
4
5	Copyright (c) 2014-2022 Snowplow Analytics Ltd. All rights reserved.
6
7	This program is licensed to you under the Apache License Version 2.0,
8	and you may not use this file except in compliance with the Apache License
9	Version 2.0. You may obtain a copy of the Apache License Version 2.0 at
10	http://www.apache.org/licenses/LICENSE-2.0.
11
12	Unless required by applicable law or agreed to in writing,
13	software distributed under the Apache License Version 2.0 is distributed on
14	an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
15	express or implied. See the Apache License Version 2.0 for the specific
16	language governing permissions and limitations there under.
17
18	Authors: Joshua Beemster
19	License: Apache License Version 2.0
20	*/
21
22	namespace Snowplow\Tracker\Emitters;
23	use Snowplow\Tracker\Emitter;
24
25	class CurlEmitter extends Emitter {
26
27	// Emitter Parameters
28
29	private $type;
30	private $url;
31	private $debug;
32	private $requests_results;
33	private $debug_payloads;
34
35	// Curl Specific Parameters
36
37	private $final_payload_buffer = array();
38	private $curl_buffer = array();
39	private $curl_limit;
40	private $rolling_window;
41	private $curl_timeout;
42
43	/**
44	* Constructs an async curl emitter.
45	*
46	* @param string $uri - Collector URI
47	* @param string\|null $protocol - What protocol we are using for the collector
48	* @param string\|null $type - The type of request we will be making to the collector
49	* @param int\|null $buffer_size - Emitter buffer size
50	* @param bool $debug - Debug mode
51	* @param int\|null $curl_timeout - Maximum time the request is allowed to take, in seconds
52	*/
53	public function __construct($uri, $protocol = NULL, $type = NULL, $buffer_size = NULL, $debug = false, $curl_timeout = NULL) {
54	$this->type = $this->getRequestType($type);	19✔
55	$this->url = $this->getCollectorUrl($this->type, $uri, $protocol);	19✔
56	$this->curl_limit = $this->type == "POST" ? self::CURL_AMOUNT_POST : self::CURL_AMOUNT_GET;	19✔
57	$this->rolling_window = $this->type == "POST" ? self::CURL_WINDOW_POST : self::CURL_WINDOW_GET;	19✔
58	$this->curl_timeout = $curl_timeout;	19✔
59
60	// If debug is on create a requests_results array
61	if ($debug === true) {	19✔
62	$this->debug = true;	16✔
63	$this->debug_payloads = array();	16✔
64	$this->requests_results = array();	16✔
65	}
66	else {
67	$this->debug = false;	3✔
68	}
69	$buffer = $buffer_size == NULL ? self::CURL_BUFFER : $buffer_size;	19✔
70	$this->setup("curl", $debug, $buffer);	19✔
71	}
72
73	/**
74	* Push event buffers into curls and store them
75	* - Wait until we have an allotted amount of curls before executing
76	* - Or force the execution of the curl emitter
77	*
78	* @param $buffer - An array of events we are going to convert into curl resources
79	* @param bool $curl_send - Whether or not we are going to send the buffered curl
80	* objects before we reach the limit
81	* @return bool\|string - Either true or an error string
82	*/
83	public function send($buffer, $curl_send = false) {
84	$type = $this->type;	16✔
85	$debug = $this->debug;	16✔
86
87	// If the sent buffer contains events...
88	if (count($buffer) > 0) {	16✔
89	if ($type == "POST") {	16✔
90	array_push($this->final_payload_buffer, $buffer);	8✔
91	}
92	else {
93	foreach ($buffer as $event) {	8✔
94	array_push($this->final_payload_buffer, $event);	8✔
95	}
96	}
97	}
98
99	if (count($this->final_payload_buffer) >= $this->curl_limit) {	16✔
100	return $this->mkCurlRequests($this->final_payload_buffer, $type, $debug);	1✔
101	}
102	else if ($curl_send === true) {	16✔
103	if (count($this->final_payload_buffer) > 0) {	15✔
104	return $this->mkCurlRequests($this->final_payload_buffer, $type, $debug);	15✔
105	}
106	else {
107	return "Error: No curls to send.";	×
108	}
109	}
110	return "Error: Still adding to the curl buffer; count ".count($this->final_payload_buffer)." - limit ".$this->curl_limit;	1✔
111	}
112
113	/**
114	* Makes curl requests from payloads to be sent.
115	*
116	* @param array $payloads - Array of payloads to be sent
117	* @param string $type - Type of requests to be made
118	* @param bool $debug - If debug is on or not
119	* @return bool\|string - Returns true or a string of errors
120	*/
121	private function mkCurlRequests($payloads, $type, $debug) {
122	// Empty the global buffer.
123	$this->final_payload_buffer = array();	16✔
124
125	if ($type == 'POST') {	16✔
126	foreach ($payloads as $buffer) {	8✔
127	$payload = $this->getPostRequest($this->batchUpdateStm($buffer));	8✔
128	$curl = $this->getCurlRequest($payload, $type);	8✔
129	array_push($this->curl_buffer, $curl);	8✔
130
131	// If debug is on; store the handle and the payload
132	if ($debug) {	8✔
133	array_push($this->debug_payloads, array("handle" => $curl, "payload" => $payload));	8✔
134	}
135	}
136	}
137	else {
138	foreach ($payloads as $event) {	8✔
139	$payload = http_build_query($this->updateStm($event));	8✔
140	$curl = $this->getCurlRequest($payload, $type);	8✔
141	array_push($this->curl_buffer, $curl);	8✔
142	}
143	}
144	return $this->rollingCurl($this->curl_buffer, $debug);	16✔
145	}
146
147	/**
148	* Asynchronously sends curl requests.
149	* - Prevents the queue from being held up by
150	* starting new requests as soon as any are done.
151	*
152	* @param array $curls - Array of curls to be sent
153	* @param bool $debug - If Debug is on or not
154	* @return bool\|string - Returns true or a string of errors
155	*/
156	private function rollingCurl($curls, $debug) {
157	// Empty the global buffer.
158	$this->curl_buffer = array();	16✔
159
160	// Create a results string to log potential errors
161	$res = true;	16✔
162	$res_ = "";	16✔
163
164	// Rolling Window == How many requests concurrently.
165	$rolling_window = $this->rolling_window;	16✔
166	$master = curl_multi_init();	16✔
167
168	// Add cUrls to handler.
169	$limit = ($rolling_window <= count($curls)) ? $rolling_window : count($curls);	16✔
170	for ($i = 0; $i < $limit; $i++) {	16✔
171	$ch = $curls[$i];	16✔
172	curl_multi_add_handle($master, $ch);	16✔
173	}
174
175	// Execute the rolling curl
176	$running = 0;	16✔
177	do {
178	do {
179	$execrun = curl_multi_exec($master, $running);	16✔
180	} while ($execrun == CURLM_CALL_MULTI_PERFORM);	16✔
181
182	while ($done = curl_multi_info_read($master)) {	16✔
183	if ($debug) {	16✔
184	$info = curl_getinfo($done['handle']);	16✔
185	if ($info['http_code'] != 200) {	16✔
186	$res_.= "Error: Curl Request Failed with response code - ".$info['http_code']."\n";	2✔
187	}
188
189	if ($this->type == "POST") {	16✔
190	$data = $this->getDebugData($done);	8✔
191	}
192	else {
193	$data = $info['url'];	8✔
194	}
195	$this->storeRequestResults($info['http_code'], $data);	16✔
196	}
197
198	// If there are still curls in the queue add them to the multi curl handler
199	if ($i < count($curls)) {	16✔
200	$ch = $curls[$i++];	1✔
201	curl_multi_add_handle($master, $ch);	1✔
202	}
203	curl_multi_remove_handle($master, $done['handle']);	16✔
204	curl_close($done['handle']);	16✔
205	}
206	} while ($running);	16✔
207	curl_multi_close($master);	16✔
208
209	if ($res_ != "") {	16✔
210	$res = $res_;	2✔
211	}
212
213	return $res;	16✔
214	}
215
216	/**
217	* Generates and returns a curl resource
218	*
219	* @param string $payload - Data included in request
220	* @param string $type - Type of request to be made
221	* @return resource
222	*/
223	private function getCurlRequest($payload, $type) {
224	$ch = curl_init($this->url);	16✔
225	if ($type == "POST") {	16✔
226	$header = array(	8✔
227	'Content-Type: '.self::POST_CONTENT_TYPE,	8✔
228	'Accept: '.self::POST_ACCEPT,	8✔
229	'Content-Length: '.strlen($payload));	8✔
230	curl_setopt($ch, CURLOPT_CUSTOMREQUEST, "POST");	8✔
231	curl_setopt($ch, CURLOPT_POSTFIELDS, $payload);	8✔
232	curl_setopt($ch, CURLOPT_HTTPHEADER, $header);	8✔
233	}
234	else {
235	curl_setopt($ch, CURLOPT_CUSTOMREQUEST, "GET");	8✔
236	curl_setopt($ch, CURLOPT_URL, $this->url."?".$payload);	8✔
237	}
238	curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);	16✔
239	if ($this->curl_timeout != NULL) {	16✔
240	curl_setopt($ch, CURLOPT_TIMEOUT, $this->curl_timeout);	×
241	}
242	return $ch;	16✔
243	}
244
245	/**
246	* Compiles events from buffer into a single string.
247	*
248	* @param array $buffer
249	* @return string - Returns a json_encoded string with all of the events to be sent.
250	*/
251	private function getPostRequest($buffer) {
252	$data = json_encode(array("schema" => self::POST_REQ_SCHEMA, "data" => $buffer));	8✔
253	return $data;	8✔
254	}
255
256	/**
257	* Disables debug mode for the emitter
258	* - If deleteLocal is true it will also empty
259	* the local cache of stored request codes and
260	* the associated payloads.
261	* - Will then trigger a function in the base
262	* emitter class to clean out the physical
263	* debug records.
264	*
265	* @param bool $deleteLocal - Empty results array
266	*/
267	public function turnOffDebug($deleteLocal) {
268	$this->debug = false;	16✔
269	if ($deleteLocal) {	16✔
270	$this->requests_results = array();	16✔
271	}
272
273	// Switch Debug off in Base Emitter Class
274	$this->debugSwitch($deleteLocal);	16✔
275	}
276
277	// Curl Return Functions
278
279	/**
280	* Returns the collector url
281	*
282	* @return string
283	*/
284	public function returnUrl() {
285	return $this->url;	2✔
286	}
287
288	/**
289	* Returns the type of Request we will be making
290	*
291	* @return null\|string
292	*/
293	public function returnType() {
294	return $this->type;	1✔
295	}
296
297	/**
298	* Returns the current array of curls we have
299	*
300	* @return array
301	*/
302	public function returnCurlBuffer() {
303	return $this->curl_buffer;	1✔
304	}
305
306	/**
307	* Returns the amount of curls we need before sending
308	*
309	* @return int
310	*/
311	public function returnCurlAmount() {
312	return $this->curl_limit;	1✔
313	}
314
315	/**
316	* Returns the amount of simultaneous curls we send
317	*
318	* @return int
319	*/
320	public function returnRollingWindow() {
321	return $this->rolling_window;	1✔
322	}
323
324	// Debug Functions
325
326	/**
327	* Returns the array of stored results from a request
328	*
329	* @return array
330	*/
331	public function returnRequestResults() {
332	return $this->requests_results;	16✔
333	}
334
335	/**
336	* Returns the payload associated with the finished curl
337	*
338	* @param array $done - The array made by finishing a curl
339	* @return string - Payload
340	*/
341	private function getDebugData($done) {
342	$data = "";	8✔
343	for ($i = 0; $i < count($this->debug_payloads); $i++) {	8✔
344	$debug_payload = $this->debug_payloads[$i];	8✔
345	if ($debug_payload["handle"] == $done['handle']) {	8✔
346	$data = $debug_payload["payload"];	8✔
347
348	// Delete the curl handle & payload from storage
349	unset($this->debug_payloads[$i]);	8✔
350
351	// Re-index the array after the deletion
352	$this->debug_payloads = array_values($this->debug_payloads);	8✔
353	}
354	}
355	return $data;	8✔
356	}
357
358	/**
359	* Stores curl request response code
360	*
361	* @param $code
362	* @param $data
363	*/
364	private function storeRequestResults($code, $data) {
365	$array["code"] = $code;	16✔
366	if ($this->type == "GET") {	16✔
367	$temp = substr($data,(strpos($data,"?")+1),-1);	8✔
368	parse_str($temp, $output);	8✔
369	$array["data"] = json_encode($output);	8✔
370	}
371	else {
372	$array["data"] = $data;	8✔
373	}
374	array_push($this->requests_results, $array);	16✔
375	}
376	}

snowplow / snowplow-php-tracker / 3750646379

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous