<?php
require_once(dirname(__FILE__) . '/AbstractConsumer.php');

/**
 * Consumes messages and sends them to a host/endpoint using cURL
 */
class ConsumerStrategies_CurlConsumer extends ConsumerStrategies_AbstractConsumer {

	protected $_host;

	protected $_endpoint;

	protected $_connect_timeout;

	protected $_timeout;

	protected $_protocol;

	protected $_fork = null;


	/**
	 * Creates a new CurlConsumer and assigns properties from the $options array
	 *
	 * @param array $options
	 * @throws Exception
	 */
	public function __construct( $options ) {
		parent::__construct($options);

		$this->_host = $options['host'];
		$this->_endpoint = $options['endpoint'];
		$this->_connect_timeout = array_key_exists('connect_timeout', $options) ? $options['connect_timeout'] : 5;
		$this->_timeout = array_key_exists('timeout', $options) ? $options['timeout'] : 30;
		$this->_protocol = array_key_exists('use_ssl', $options) && ( true == $options['use_ssl'] ) ? 'https' : 'http';
		$this->_fork = array_key_exists('fork', $options) ? ( true == $options['fork'] ) : false;

		// ensure the environment is workable for the given settings
		if (true == $this->_fork) {
			$exists = function_exists('exec');
			if (!$exists) {
				throw new Exception('The "exec" function must exist to use the cURL consumer in "fork" mode. Try setting fork = false or use another consumer.');
			}
			$disabled = explode(', ', ini_get('disable_functions'));
			$enabled = !in_array('exec', $disabled);
			if (!$enabled) {
				throw new Exception('The "exec" function must be enabled to use the cURL consumer in "fork" mode. Try setting fork = false or use another consumer.');
			}
		} else {
			if (!function_exists('curl_init')) {
				throw new Exception('The cURL PHP extension is required to use the cURL consumer with fork = false. Try setting fork = true or use another consumer.');
			}
		}
	}


	/**
	 * Write to the given host/endpoint using either a forked cURL process or using PHP's cURL extension
	 *
	 * @param array $batch
	 * @return bool
	 */
	public function persist( $batch ) {
		if (count($batch) > 0) {
			$data = 'data=' . $this->_encode($batch);
			$url = $this->_protocol . '://' . $this->_host . $this->_endpoint;
			if ($this->_fork) {
				return $this->_execute_forked($url, $data);
			} else {
				return $this->_execute($url, $data);
			}
		} else {
			return true;
		}
	}

	/**
	 * Write using the cURL php extension
	 *
	 * @param $url
	 * @param $data
	 * @return bool
	 */
	protected function _execute( $url, $data ) {
		if ($this->_debug()) {
			$this->_log("Making blocking cURL call to $url");
		}

		$ch = curl_init();
		curl_setopt($ch, CURLOPT_URL, $url);
		curl_setopt($ch, CURLOPT_HEADER, 0);
		curl_setopt($ch, CURLOPT_CONNECTTIMEOUT, $this->_connect_timeout);
		curl_setopt($ch, CURLOPT_TIMEOUT, $this->_timeout);
		curl_setopt($ch, CURLOPT_POST, 1);
		curl_setopt($ch, CURLOPT_RETURNTRANSFER, 1);
		curl_setopt($ch, CURLOPT_POSTFIELDS, $data);
		$response = curl_exec($ch);
		if (false === $response) {
			$curl_error = curl_error($ch);
			$curl_errno = curl_errno($ch);
			curl_close($ch);
			$this->_handleError($curl_errno, $curl_error);
			return false;
		} else {
			curl_close($ch);
			if (trim($response) == '1') {
				return true;
			} else {
				$this->_handleError(0, $response);
				return false;
			}
		}
	}


	/**
	 * Write using a forked cURL process
	 *
	 * @param $url
	 * @param $data
	 * @return bool
	 */
	protected function _execute_forked( $url, $data ) {

		if ($this->_debug()) {
			$this->_log("Making forked cURL call to $url");
		}

		$exec = 'curl -X POST -H "Content-Type: application/x-www-form-urlencoded" -d ' . $data . ' "' . $url . '"';

		if (!$this->_debug()) {
			$exec .= ' >/dev/null 2>&1 &';
		}

		exec($exec, $output, $return_var);

		if (0 != $return_var) {
			$this->_handleError($return_var, $output);
		}

		return  0 == $return_var;
	}

	public function getConnectTimeout() {
		return $this->_connect_timeout;
	}

	public function getEndpoint() {
		return $this->_endpoint;
	}

	public function getFork() {
		return $this->_fork;
	}

	public function getHost() {
		return $this->_host;
	}

	public function getOptions() {
		return $this->_options;
	}

	public function getProtocol() {
		return $this->_protocol;
	}

	public function getTimeout() {
		return $this->_timeout;
	}
}
