<?php

namespace WPMUDEV_Analytics_Vendor;

/**
 * Portions of this class were borrowed from
 * https://github.com/segmentio/analytics-php/blob/master/lib/Analytics/Consumer/Socket.php.
 * Thanks for the work!
 *
 * WWWWWW||WWWWWW
 * W W W||W W W
 * ||
 * ( OO )__________
 * /  |           \
 * /o o|    MIT     \
 * \___/||_||__||_|| *
 * || ||  || ||
 * _||_|| _||_||
 * (__|__|(__|__|
 * (The MIT License)
 *
 * Copyright (c) 2013 Segment.io Inc. [email protected]
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated
 * documentation files (the 'Software'), to deal in the Software without restriction, including without limitation the
 * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to
 * permit persons to whom the Software is furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all copies or substantial portions of the
 * Software.
 *
 * THE SOFTWARE IS PROVIDED 'AS IS', WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE
 * WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS
 * OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
 * OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 */
require_once \dirname(__FILE__) . "/AbstractConsumer.php";
/**
 * Consumes messages and writes them to host/endpoint using a persistent socket
 */
class ConsumerStrategies_SocketConsumer extends ConsumerStrategies_AbstractConsumer
{
    /**
     * @var string the host to connect to (e.g. api.mixpanel.com)
     */
    private $_host;
    /**
     * @var string the host-relative endpoint to write to (e.g. /engage)
     */
    private $_endpoint;
    /**
     * @var int connect_timeout the socket connection timeout in seconds
     */
    private $_connect_timeout;
    /**
     * @var string the protocol to use for the socket connection
     */
    private $_protocol;
    /**
     * @var resource holds the socket resource
     */
    private $_socket;
    /**
     * @var bool whether or not to wait for a response
     */
    private $_async;
    /**
     * @var int the port to use for the socket connection
     */
    private $_port;
    /**
     * Creates a new SocketConsumer and assigns properties from the $options array
     * @param array $options
     */
    public function __construct($options = array())
    {
        parent::__construct($options);
        $this->_host = $options['host'];
        $this->_endpoint = $options['endpoint'];
        $this->_connect_timeout = isset($options['connect_timeout']) ? $options['connect_timeout'] : 5;
        $this->_async = isset($options['async']) && $options['async'] === \false ? \false : \true;
        if (\array_key_exists('use_ssl', $options) && $options['use_ssl'] == \true) {
            $this->_protocol = "ssl";
            $this->_port = 443;
        } else {
            $this->_protocol = "tcp";
            $this->_port = 80;
        }
    }
    /**
     * Write using a persistent socket connection.
     * @param array $batch
     * @return bool
     */
    public function persist($batch)
    {
        $socket = $this->_getSocket();
        if (!\is_resource($socket)) {
            return \false;
        }
        $data = "data=" . $this->_encode($batch);
        $body = "";
        $body .= "POST " . $this->_endpoint . " HTTP/1.1\r\n";
        $body .= "Host: " . $this->_host . "\r\n";
        $body .= "Content-Type: application/x-www-form-urlencoded\r\n";
        $body .= "Accept: application/json\r\n";
        $body .= "Content-length: " . \strlen($data) . "\r\n";
        $body .= "\r\n";
        $body .= $data;
        return $this->_write($socket, $body);
    }
    /**
     * Return cached socket if open or create a new persistent socket
     * @return bool|resource
     */
    private function _getSocket()
    {
        if (\is_resource($this->_socket)) {
            if ($this->_debug()) {
                $this->_log("Using existing socket");
            }
            return $this->_socket;
        } else {
            if ($this->_debug()) {
                $this->_log("Creating new socket at " . \time());
            }
            return $this->_createSocket();
        }
    }
    /**
     * Attempt to open a new socket connection, cache it, and return the resource
     * @param bool $retry
     * @return bool|resource
     */
    private function _createSocket($retry = \true)
    {
        try {
            $socket = \pfsockopen($this->_protocol . "://" . $this->_host, $this->_port, $err_no, $err_msg, $this->_connect_timeout);
            if ($this->_debug()) {
                $this->_log("Opening socket connection to " . $this->_protocol . "://" . $this->_host . ":" . $this->_port);
            }
            if ($err_no != 0) {
                $this->_handleError($err_no, $err_msg);
                return $retry == \true ? $this->_createSocket(\false) : \false;
            } else {
                // cache the socket
                $this->_socket = $socket;
                return $socket;
            }
        } catch (\Exception $e) {
            $this->_handleError($e->getCode(), $e->getMessage());
            return $retry == \true ? $this->_createSocket(\false) : \false;
        }
    }
    /**
     * Attempt to close and dereference a socket resource
     */
    private function _destroySocket()
    {
        $socket = $this->_socket;
        $this->_socket = null;
        \fclose($socket);
    }
    /**
     * Write $data through the given $socket
     * @param $socket
     * @param $data
     * @param bool $retry
     * @return bool
     */
    private function _write($socket, $data, $retry = \true)
    {
        $bytes_sent = 0;
        $bytes_total = \strlen($data);
        $socket_closed = \false;
        $success = \true;
        $max_bytes_per_write = 8192;
        // if we have no data to write just return true
        if ($bytes_total == 0) {
            return \true;
        }
        // try to write the data
        while (!$socket_closed && $bytes_sent < $bytes_total) {
            try {
                $bytes = \fwrite($socket, $data, $max_bytes_per_write);
                if ($this->_debug()) {
                    $this->_log("Socket wrote " . $bytes . " bytes");
                }
                // if we actually wrote data, then remove the written portion from $data left to write
                if ($bytes > 0) {
                    $data = \substr($data, $max_bytes_per_write);
                }
            } catch (\Exception $e) {
                $this->_handleError($e->getCode(), $e->getMessage());
                $socket_closed = \true;
            }
            if (isset($bytes) && $bytes) {
                $bytes_sent += $bytes;
            } else {
                $socket_closed = \true;
            }
        }
        // create a new socket if the current one is closed and retry the message
        if ($socket_closed) {
            $this->_destroySocket();
            if ($retry) {
                if ($this->_debug()) {
                    $this->_log("Retrying socket write...");
                }
                $socket = $this->_getSocket();
                if ($socket) {
                    return $this->_write($socket, $data, \false);
                }
            }
            return \false;
        }
        // only wait for the response in debug mode or if we explicitly want to be synchronous
        if ($this->_debug() || !$this->_async) {
            $res = $this->handleResponse(\fread($socket, 2048));
            if ($res["status"] != "200") {
                $this->_handleError($res["status"], $res["body"]);
                $success = \false;
            }
        }
        return $success;
    }
    /**
     * Parse the response from a socket write (only used for debugging)
     * @param $response
     * @return array
     */
    private function handleResponse($response)
    {
        $lines = \explode("\n", $response);
        // extract headers
        $headers = array();
        foreach ($lines as $line) {
            $kvsplit = \explode(":", $line);
            if (\count($kvsplit) == 2) {
                $header = $kvsplit[0];
                $value = $kvsplit[1];
                $headers[$header] = \trim($value);
            }
        }
        // extract status
        $line_one_exploded = \explode(" ", $lines[0]);
        $status = $line_one_exploded[1];
        // extract body
        $body = $lines[\count($lines) - 1];
        // if the connection has been closed lets kill the socket
        if (isset($headers["Connection"]) and $headers['Connection'] == "close") {
            $this->_destroySocket();
            if ($this->_debug()) {
                $this->_log("Server told us connection closed so lets destroy the socket so it'll reconnect on next call");
            }
        }
        $ret = array("status" => $status, "body" => $body);
        return $ret;
    }
}