5761683854

pending completion

Build # 5761683854

Build Type

push

github

Committed by

matus-tomlein

Commit Message

Merge branch 'release/0.7.0'

Run Details

52 of 52 new or added lines in 3 files covered. (100.0%)

558 of 620 relevant lines covered (90.0%)

25.73 hits per line

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

88.71

/src/Emitters/SocketEmitter.php

<?php
/*
    SocketEmitter.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;
use Snowplow\Tracker\Emitters\RetryRequestManager;
use Exception;

class SocketEmitter extends Emitter {

    // Emitter Parameters

    private $uri;
    private $ssl;
    private $type;
    private $timeout;
    private $debug;
    private $requests_results;
    private $max_retry_attempts;
    private $retry_backoff_ms;

    // Socket Parameters

    private $socket_failed = false;
    private $socket;

    /**
     * Creates a Socket Emitter
     *
     * @param string $uri - Collector URI to be used for the request
     * @param bool|null $ssl - If the collector is using SSL
     * @param string|null $type - Type of request to be made (POST || GET)
     * @param float|int|null $timeout - Timeout for the socket connection
     * @param int|null $buffer_size - Number of events to buffer before making a POST request to collector
     * @param bool|null $debug - If debug is on
     * @param int|null $max_retry_attempts - The maximum number of times to retry a request. Defaults to 1.
     * @param int|null $retry_backoff_ms - The number of milliseconds to backoff before retrying a request. Defaults to 100ms.
     */
    public function __construct($uri, $ssl = NULL, $type = NULL, $timeout = NULL, $buffer_size = NULL, $debug = NULL, $max_retry_attempts = NULL, $retry_backoff_ms = NULL) {
        $this->type    = $this->getRequestType($type);
        $this->uri     = $uri;
        $this->ssl     = $ssl == NULL ? self::DEFAULT_SSL : (bool) $ssl;
        $this->timeout = $timeout == NULL ? self::SOCKET_TIMEOUT : $timeout;
        $this->max_retry_attempts = $max_retry_attempts;
        $this->retry_backoff_ms = $retry_backoff_ms;

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

    /**
     * Sends all the events in the buffer to the HTTP Socket.
     *
     * @param array $buffer
     * @return bool $res
     */
    public function send($buffer) {
        if (count($buffer) > 0) {
            $uri = $this->uri;
            $type = $this->type;
            $socket_made = $this->makeSocket();

            if (is_bool($socket_made) && $socket_made) {
                if ($type == "POST") {
                    $data = $this->getPostRequest($this->batchUpdateStm($buffer));
                    $body = $this->getRequestBody($uri, $data, $type);

                    // Send requests to the socket
                    $res = $this->makeRequest($body);
                    return $res;
                }
                else {
                    $res = "";
                    $res_ = "";
                    foreach ($buffer as $event) {
                        $data = http_build_query($this->updateStm($event));
                        $body = $this->getRequestBody($uri, $data, $type);

                        // Send request to the socket
                        $res = $this->makeRequest($body);
                        if (!is_bool($res)) {
                            $res_.= "Error: Socket write failed\n".$res;
                        }
                        $this->makeSocket();
                    }
                    fclose($this->socket);

                    // If we have had any errors return these
                    if ($res_ != "") {
                        $res = $res_;
                    }
                    return $res;
                }
            }
            else {
                return "Error: Socket could not be created\n".$socket_made;
            }
        }
        return "No events to write";
    }

    /**
     * Writes requests to the socket
     * - If Retry is set to True, will attempt a second Write.
     *
     * @param string $data - Data that we are sending to the socket.
     * @param bool $retry - If we want to allow the function to make a second attempt.
     * @return bool - Returns if write was successful.
     */
    private function makeRequest($data, $retry_request_manager = NULL) {
        $bytes_written = 0;
        $bytes_total = strlen($data);
        $closed = false;
        $res_ = "";

        if ($retry_request_manager == NULL) {
            $retry_request_manager = new RetryRequestManager($this->max_retry_attempts, $this->retry_backoff_ms);
        }

        // Try to send data while bytes still have to be written to the socket
        while (!$closed && $bytes_written < $bytes_total) {
            try {
                $written = fwrite($this->socket, substr($data, $bytes_written));
            }
            catch (Exception $e) {
                $res_.= "Error: fwrite exception - ".$e."\n";
                $closed = true;
            }
            if (!isset($written) || $written === false) {
                $closed = true;
            }
            else {
                $bytes_written += $written;
            }
        }

        $status_code = 0;

        if (!$closed) {
            $res = fread($this->socket, 2048);
            $status_code = $this->parseStatusCode($res);

            // If debug is on store the request result
            if ($this->debug) {
                $this->storeRequestResults($res, $data);
            }
        }

        fclose($this->socket);

        // If the socket could not be written attempt again
        if (($closed && $retry_request_manager->shouldRetryFailedRequest()) ||
            $retry_request_manager->shouldRetryForStatusCode($status_code)) {
            $retry_request_manager->incrementRetryCount();
            $retry_request_manager->backoff();

            $socket_made = $this->makeSocket();
            if (is_bool($socket_made) && $socket_made) {
                return $this->makeRequest($data, $retry_request_manager);
            }
            else {
                return "Error: Socket could not be created\n".$socket_made."\n".$res_;
            }
        } else if ($closed) {
            return "Error: socket cannot be written after retry\n".$res_;
        } else if (!$retry_request_manager->isGoodStatusCode($status_code)) {
            return "Error: collector responded with status code ".$status_code.", won't retry";
        }

        return true;
    }

    /**
     * Creates a new socket connection to the collector.
     * - If the socket has previously failed do not allow another attempt
     *
     * @return bool|resource - Returns a socket resource or false if it fails.
     */
    private function makeSocket() {
        if ($this->socket_failed) {
            return "Error: socket cannot be created";
        }

        $protocol = $this->ssl ? "ssl" : "tcp";
        $port     = $this->ssl ? 443 : 80;

        try {
            $socket = pfsockopen($protocol."://".$this->uri, $port, $errno, $errstr, $this->timeout);
            if ($errno != 0) {
                $this->socket_failed = true;
                return "Error: socket creation failed on error number - ".$errno;
            }
            $this->socket = $socket;
            return true;
        }
        catch (Exception $e) {
            $this->socket_failed = true;
            return "Error: socket exception - ".$e;
        }
    }

    /**
     * Builds a Request which will be sent to the socket.
     *
     * @param string $uri - Collector URI to be used for the request
     * @param string|array $data - Data to be included in the Request
     * @param string $type - Type of request to be made (POST || GET)
     * @return string - Returns the request body
     */
    private function getRequestBody($uri, $data, $type) {
        if ($type == "POST") {
            $req = "POST http://".$uri.self::POST_PATH." ";
            $req.= "HTTP/1.1\r\n";
            $req.= "Host: ".$uri."\r\n";
            $req.= "Content-Type: ".self::POST_CONTENT_TYPE."\r\n";
            $req.= "Content-length: ".strlen($data)."\r\n";
            $req.= "Accept: ".self::POST_ACCEPT."\r\n\r\n";
            $req.= $data."\r\n\r\n";
        }
        else {
            $req = "GET http://".$uri.self::GET_PATH."?".$data." ";
            $req.= "HTTP/1.1\r\n";
            $req.= "Host: ".$uri."\r\n";
            $req.= "Query: ".$data."\r\n";
            $req.= "\r\n";
        }
        return $req;
    }

    /**
     * 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);
    }

    // Socket Return Functions

    /**
     * Returns the collectors uri
     *
     * @return mixed
     */
    public function returnUri() {
        return $this->uri;
    }

    /**
     * Returns if the collector is using an ssl connection
     *
     * @return bool|null
     */
    public function returnSsl() {
        return $this->ssl;
    }

    /**
     * Returns the emitter timeout
     *
     * @return float|null
     */
    public function returnTimeout() {
        return $this->timeout;
    }

    /**
     * Returns the request type the emitter is using
     *
     * @return null|string
     */
    public function returnType() {
        return $this->type;
    }

    /**
     * Returns the socket resource
     *
     * @return resource|null
     */
    public function returnSocket() {
        return $this->socket;
    }

    /**
     * Returns if the socket has failed or not
     *
     * @return bool
     */
    public function returnSocketIsFailed() {
        return $this->socket_failed;
    }

    // Debug Functions

    /**
     * Returns the results array which contains all response codes and body payloads.
     *
     * @return array
     */
    public function returnRequestResults() {
        return $this->requests_results;
    }

    private function parseStatusCode($res) {
        $contents = explode("\n", $res);
        $status = explode(" ", $contents[0], 3);
        return $status[1];
    }

    /**
     * Sends the results string from the request and then gets the return code.
     *
     * @param string $res
     * @param string $data - Raw POST body we are writing to the socket
     */
    private function storeRequestResults($res, $data) {
        // Get the Response code...
        $contents = explode("\n", $res);
        $status = explode(" ", $contents[0], 3);
        $array["code"] = $status[1];

        // Store the Body Payload
        $contents = explode("\n", $data);
        if ($this->type == "POST") {
            $array["data"] = $contents[count($contents)-3];
        }
        else {
            $data = substr($contents[2], 7, -1);
            parse_str($data, $output);
            $array["data"] = json_encode($output);
        }

        // Push the response code and body payload into the results array
        array_push($this->requests_results, $array);
    }
}

1	<?php
2	/*
3	SocketEmitter.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	use Snowplow\Tracker\Emitters\RetryRequestManager;
25	use Exception;
26
27	class SocketEmitter extends Emitter {
28
29	// Emitter Parameters
30
31	private $uri;
32	private $ssl;
33	private $type;
34	private $timeout;
35	private $debug;
36	private $requests_results;
37	private $max_retry_attempts;
38	private $retry_backoff_ms;
39
40	// Socket Parameters
41
42	private $socket_failed = false;
43	private $socket;
44
45	/**
46	* Creates a Socket Emitter
47	*
48	* @param string $uri - Collector URI to be used for the request
49	* @param bool\|null $ssl - If the collector is using SSL
50	* @param string\|null $type - Type of request to be made (POST \|\| GET)
51	* @param float\|int\|null $timeout - Timeout for the socket connection
52	* @param int\|null $buffer_size - Number of events to buffer before making a POST request to collector
53	* @param bool\|null $debug - If debug is on
54	* @param int\|null $max_retry_attempts - The maximum number of times to retry a request. Defaults to 1.
55	* @param int\|null $retry_backoff_ms - The number of milliseconds to backoff before retrying a request. Defaults to 100ms.
56	*/
57	public function __construct($uri, $ssl = NULL, $type = NULL, $timeout = NULL, $buffer_size = NULL, $debug = NULL, $max_retry_attempts = NULL, $retry_backoff_ms = NULL) {
58	$this->type = $this->getRequestType($type);	36✔
59	$this->uri = $uri;	36✔
60	$this->ssl = $ssl == NULL ? self::DEFAULT_SSL : (bool) $ssl;	36✔
61	$this->timeout = $timeout == NULL ? self::SOCKET_TIMEOUT : $timeout;	36✔
62	$this->max_retry_attempts = $max_retry_attempts;	36✔
63	$this->retry_backoff_ms = $retry_backoff_ms;	36✔
64
65	// If debug is on create a requests_results array
66	if ($debug === true) {	36✔
67	$this->debug = true;	32✔
68	$this->requests_results = array();	32✔
69	}
70	else {
71	$this->debug = false;	4✔
72	}
73	$buffer = $buffer_size == NULL ? self::SOCKET_BUFFER : $buffer_size;	36✔
74	$this->setup("socket", $debug, $buffer);	36✔
75	}
76
77	/**
78	* Sends all the events in the buffer to the HTTP Socket.
79	*
80	* @param array $buffer
81	* @return bool $res
82	*/
83	public function send($buffer) {
84	if (count($buffer) > 0) {	32✔
85	$uri = $this->uri;	32✔
86	$type = $this->type;	32✔
87	$socket_made = $this->makeSocket();	32✔
88
89	if (is_bool($socket_made) && $socket_made) {	32✔
90	if ($type == "POST") {	32✔
91	$data = $this->getPostRequest($this->batchUpdateStm($buffer));	12✔
92	$body = $this->getRequestBody($uri, $data, $type);	12✔
93
94	// Send requests to the socket
95	$res = $this->makeRequest($body);	12✔
96	return $res;	12✔
97	}
98	else {
99	$res = "";	20✔
100	$res_ = "";	20✔
101	foreach ($buffer as $event) {	20✔
102	$data = http_build_query($this->updateStm($event));	20✔
103	$body = $this->getRequestBody($uri, $data, $type);	20✔
104
105	// Send request to the socket
106	$res = $this->makeRequest($body);	20✔
107	if (!is_bool($res)) {	20✔
108	$res_.= "Error: Socket write failed\n".$res;	6✔
109	}
110	$this->makeSocket();	20✔
111	}
112	fclose($this->socket);	20✔
113
114	// If we have had any errors return these
115	if ($res_ != "") {	20✔
116	$res = $res_;	6✔
117	}
118	return $res;	20✔
119	}
120	}
121	else {
122	return "Error: Socket could not be created\n".$socket_made;	×
123	}
124	}
125	return "No events to write";	×
126	}
127
128	/**
129	* Writes requests to the socket
130	* - If Retry is set to True, will attempt a second Write.
131	*
132	* @param string $data - Data that we are sending to the socket.
133	* @param bool $retry - If we want to allow the function to make a second attempt.
134	* @return bool - Returns if write was successful.
135	*/
136	private function makeRequest($data, $retry_request_manager = NULL) {
137	$bytes_written = 0;	32✔
138	$bytes_total = strlen($data);	32✔
139	$closed = false;	32✔
140	$res_ = "";	32✔
141
142	if ($retry_request_manager == NULL) {	32✔
143	$retry_request_manager = new RetryRequestManager($this->max_retry_attempts, $this->retry_backoff_ms);	32✔
144	}
145
146	// Try to send data while bytes still have to be written to the socket
147	while (!$closed && $bytes_written < $bytes_total) {	32✔
148	try {
149	$written = fwrite($this->socket, substr($data, $bytes_written));	32✔
150	}
151	catch (Exception $e) {	×
152	$res_.= "Error: fwrite exception - ".$e."\n";	×
153	$closed = true;	×
154	}
155	if (!isset($written) \|\| $written === false) {	32✔
156	$closed = true;	×
157	}
158	else {
159	$bytes_written += $written;	32✔
160	}
161	}
162
163	$status_code = 0;	32✔
164
165	if (!$closed) {	32✔
166	$res = fread($this->socket, 2048);	32✔
167	$status_code = $this->parseStatusCode($res);	32✔
168
169	// If debug is on store the request result
170	if ($this->debug) {	32✔
171	$this->storeRequestResults($res, $data);	32✔
172	}
173	}
174
175	fclose($this->socket);	32✔
176
177	// If the socket could not be written attempt again
178	if (($closed && $retry_request_manager->shouldRetryFailedRequest()) \|\|	32✔
179	$retry_request_manager->shouldRetryForStatusCode($status_code)) {	32✔
180	$retry_request_manager->incrementRetryCount();	4✔
181	$retry_request_manager->backoff();	4✔
182
183	$socket_made = $this->makeSocket();	4✔
184	if (is_bool($socket_made) && $socket_made) {	4✔
185	return $this->makeRequest($data, $retry_request_manager);	4✔
186	}
187	else {
188	return "Error: Socket could not be created\n".$socket_made."\n".$res_;	×
189	}
190	} else if ($closed) {	32✔
191	return "Error: socket cannot be written after retry\n".$res_;	×
192	} else if (!$retry_request_manager->isGoodStatusCode($status_code)) {	32✔
193	return "Error: collector responded with status code ".$status_code.", won't retry";	6✔
194	}
195
196	return true;	26✔
197	}
198
199	/**
200	* Creates a new socket connection to the collector.
201	* - If the socket has previously failed do not allow another attempt
202	*
203	* @return bool\|resource - Returns a socket resource or false if it fails.
204	*/
205	private function makeSocket() {
206	if ($this->socket_failed) {	32✔
207	return "Error: socket cannot be created";	×
208	}
209
210	$protocol = $this->ssl ? "ssl" : "tcp";	32✔
211	$port = $this->ssl ? 443 : 80;	32✔
212
213	try {
214	$socket = pfsockopen($protocol."://".$this->uri, $port, $errno, $errstr, $this->timeout);	32✔
215	if ($errno != 0) {	32✔
216	$this->socket_failed = true;	×
217	return "Error: socket creation failed on error number - ".$errno;	×
218	}
219	$this->socket = $socket;	32✔
220	return true;	32✔
221	}
222	catch (Exception $e) {	×
223	$this->socket_failed = true;	×
224	return "Error: socket exception - ".$e;	×
225	}
226	}
227
228	/**
229	* Builds a Request which will be sent to the socket.
230	*
231	* @param string $uri - Collector URI to be used for the request
232	* @param string\|array $data - Data to be included in the Request
233	* @param string $type - Type of request to be made (POST \|\| GET)
234	* @return string - Returns the request body
235	*/
236	private function getRequestBody($uri, $data, $type) {
237	if ($type == "POST") {	32✔
238	$req = "POST http://".$uri.self::POST_PATH." ";	12✔
239	$req.= "HTTP/1.1\r\n";	12✔
240	$req.= "Host: ".$uri."\r\n";	12✔
241	$req.= "Content-Type: ".self::POST_CONTENT_TYPE."\r\n";	12✔
242	$req.= "Content-length: ".strlen($data)."\r\n";	12✔
243	$req.= "Accept: ".self::POST_ACCEPT."\r\n\r\n";	12✔
244	$req.= $data."\r\n\r\n";	12✔
245	}
246	else {
247	$req = "GET http://".$uri.self::GET_PATH."?".$data." ";	20✔
248	$req.= "HTTP/1.1\r\n";	20✔
249	$req.= "Host: ".$uri."\r\n";	20✔
250	$req.= "Query: ".$data."\r\n";	20✔
251	$req.= "\r\n";	20✔
252	}
253	return $req;	32✔
254	}
255
256	/**
257	* Compiles events from buffer into a single string.
258	*
259	* @param array $buffer
260	* @return string - Returns a json_encoded string with all of the events to be sent.
261	*/
262	private function getPostRequest($buffer) {
263	$data = json_encode(array("schema" => self::POST_REQ_SCHEMA, "data" => $buffer));	12✔
264	return $data;	12✔
265	}
266
267	/**
268	* Disables debug mode for the emitter
269	* - If deleteLocal is true it will also empty
270	* the local cache of stored request codes and
271	* the associated payloads.
272	* - Will then trigger a function in the base
273	* emitter class to clean out the physical
274	* debug records.
275	*
276	* @param bool $deleteLocal - Empty results array
277	*/
278	public function turnOffDebug($deleteLocal) {
279	$this->debug = false;	32✔
280	if ($deleteLocal) {	32✔
281	$this->requests_results = array();	32✔
282	}
283
284	// Switch debug off in Base Emitter Class
285	$this->debugSwitch($deleteLocal);	32✔
286	}
287
288	// Socket Return Functions
289
290	/**
291	* Returns the collectors uri
292	*
293	* @return mixed
294	*/
295	public function returnUri() {
296	return $this->uri;	2✔
297	}
298
299	/**
300	* Returns if the collector is using an ssl connection
301	*
302	* @return bool\|null
303	*/
304	public function returnSsl() {
305	return $this->ssl;	2✔
306	}
307
308	/**
309	* Returns the emitter timeout
310	*
311	* @return float\|null
312	*/
313	public function returnTimeout() {
314	return $this->timeout;	2✔
315	}
316
317	/**
318	* Returns the request type the emitter is using
319	*
320	* @return null\|string
321	*/
322	public function returnType() {
323	return $this->type;	2✔
324	}
325
326	/**
327	* Returns the socket resource
328	*
329	* @return resource\|null
330	*/
331	public function returnSocket() {
332	return $this->socket;	2✔
333	}
334
335	/**
336	* Returns if the socket has failed or not
337	*
338	* @return bool
339	*/
340	public function returnSocketIsFailed() {
341	return $this->socket_failed;	2✔
342	}
343
344	// Debug Functions
345
346	/**
347	* Returns the results array which contains all response codes and body payloads.
348	*
349	* @return array
350	*/
351	public function returnRequestResults() {
352	return $this->requests_results;	32✔
353	}
354
355	private function parseStatusCode($res) {
356	$contents = explode("\n", $res);	32✔
357	$status = explode(" ", $contents[0], 3);	32✔
358	return $status[1];	32✔
359	}
360
361	/**
362	* Sends the results string from the request and then gets the return code.
363	*
364	* @param string $res
365	* @param string $data - Raw POST body we are writing to the socket
366	*/
367	private function storeRequestResults($res, $data) {
368	// Get the Response code...
369	$contents = explode("\n", $res);	32✔
370	$status = explode(" ", $contents[0], 3);	32✔
371	$array["code"] = $status[1];	32✔
372
373	// Store the Body Payload
374	$contents = explode("\n", $data);	32✔
375	if ($this->type == "POST") {	32✔
376	$array["data"] = $contents[count($contents)-3];	12✔
377	}
378	else {
379	$data = substr($contents[2], 7, -1);	20✔
380	parse_str($data, $output);	20✔
381	$array["data"] = json_encode($output);	20✔
382	}
383
384	// Push the response code and body payload into the results array
385	array_push($this->requests_results, $array);	32✔
386	}
387	}

snowplow / snowplow-php-tracker / 5761683854

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