Data Transfer: Consumer

Consumers are used to send track requests to the Intelligence track server. These are:

• CURL: Saves track requests in memory and sends them to the server using the same process as the library.

• FORK_CURL (deprecated): Saves track requests in memory and sends them with the curl command line tool. It waits for curl like CURL, use CURL instead. connectionTimeout and readTimeout have no effect.

• FILE: Requests are saved in a separate file and sent via a cronjob to the track server at a specified interval.

• FILE_ROTATION: Best for high traffic websites. Requests are saved in separate files and sent via a cronjob to the track server at a specified interval.

• CUSTOM: Create your own consumer to send the tracking requests.

These constants are available in the class MappIntelligenceConsumerType.

Requests are collected in a queue and sent in batches with flush(), close() or at the end of the request (with requestInterval also while requests are added). Each batch has one of these results:

If the queue reaches maxQueueSize, it is sent first; if it is still full, the oldest requests are discarded with a warning.

$mappIntelligenceConfig = new MappIntelligenceConfig();
$mappIntelligenceConfig->setTrackDomain("analytics01.wt-eu02.net")
    ->setTrackId("111111111111111")
    ->setConsumerType(MappIntelligenceConsumerType::CURL)
    ->setMaxAttempt(1)
    ->setAttemptTimeout(100)
    ->setMaxBatchSize(1)
	->setMaxQueueSize(1000)
    ->setForceSSL(true);
$mappIntelligence = MappIntelligence::getInstance($mappIntelligenceConfig);

$mappIntelligenceConfig = new MappIntelligenceConfig();
$mappIntelligenceConfig->setTrackDomain("analytics01.wt-eu02.net")
    ->setTrackId("111111111111111")
    ->setConsumerType(MappIntelligenceConsumerType::FORK_CURL)
    ->setMaxAttempt(1)
    ->setAttemptTimeout(100)
    ->setMaxBatchSize(1)
	->setMaxQueueSize(1000)
    ->setForceSSL(true);
$mappIntelligence = MappIntelligence::getInstance($mappIntelligenceConfig);

$mappIntelligenceConfig = new MappIntelligenceConfig();
$mappIntelligenceConfig->setTrackDomain("analytics01.wt-eu02.net")
    ->setTrackId("111111111111111")
    ->setConsumerType(MappIntelligenceConsumerType::FILE)
    ->setFilename("path/to/request/file.log");
$mappIntelligence = MappIntelligence::getInstance($mappIntelligenceConfig);

$mappIntelligenceConfig = new MappIntelligenceConfig();
$mappIntelligenceConfig->setTrackDomain("analytics01.wt-eu02.net")
    ->setTrackId("111111111111111")
    ->setConsumerType(MappIntelligenceConsumerType::FILE_ROTATION)
	->setFilePath("path/to/file")
	->setFilePrefix("MappIntelligenceRequests")
	->setMaxAttempt(1)
    ->setAttemptTimeout(100)
	->setMaxFileLines(5000)
	->setMaxFileDuration(30 * 60 * 1000)
	->setMaxFileSize(12 * 1024 * 1024);
$mappIntelligence = MappIntelligence::getInstance($mappIntelligenceConfig);

class TestCustomConsumer implements MappIntelligenceConsumer
{
	public function sendBatch(array $batchContent) {
		// implement custom consumer logic
		return true;
	}
}
$mappIntelligenceConfig = new MappIntelligenceConfig();
$mappIntelligenceConfig->setTrackDomain("analytics01.wt-eu02.net")
    ->setTrackId("111111111111111")
	->setConsumerType(MappIntelligenceConsumerType::CUSTOM)
	->setConsumer(new TestCustomConsumer());
$mappIntelligence = MappIntelligence::getInstance($mappIntelligenceConfig);

Examples for custom consumers

The following examples show CUSTOM consumers for two typical use cases: buffering the requests in Kafka or Redis before they are sent to the track server, and checking your tracking in unit tests. The examples are not part of the library. Adapt them to your environment.

Kafka

Writes every request as one message into a Kafka topic (PHP extension rdkafka). The batch counts as SENT only when the broker has confirmed every message: flush() waits for the delivery reports, and the delivery report callback notes a failed message. An exception of produce() (for example a full producer queue) counts as RETRY.

class KafkaTrackingConsumer implements MappIntelligenceConsumer
{
	private $producer;
	private $topic;
	private $failed = false;

	/**
	 * @param \RdKafka\Conf $conf Producer configuration
	 * @param string $topicName
	 */
	public function __construct(\RdKafka\Conf $conf, $topicName) {
		$conf->setDrMsgCb(function ($kafka, $message) {
			if ($message->err !== RD_KAFKA_RESP_ERR_NO_ERROR) {
				$this->failed = true;
			}
		});
		$this->producer = new \RdKafka\Producer($conf);
		$this->topic = $this->producer->newTopic($topicName);
	}

	public function sendBatch(array $batchContent) {
		$this->failed = false;
		foreach ($batchContent as $request) {
			$this->topic->produce(RD_KAFKA_PARTITION_UA, 0, $request);
		}

		// waits for the delivery reports, at most 10 seconds
		return $this->producer->flush(10000) === RD_KAFKA_RESP_ERR_NO_ERROR && !$this->failed;
	}
}

$conf = new \RdKafka\Conf();
$conf->set('bootstrap.servers', 'kafka:9092');
$conf->set('acks', 'all');
// a message that is not delivered within 5 seconds fails before flush() gives up
$conf->set('message.timeout.ms', '5000');

$mappIntelligenceConfig = new MappIntelligenceConfig();
$mappIntelligenceConfig->setTrackDomain("analytics01.wt-eu02.net")
    ->setTrackId("111111111111111")
	->setConsumerType(MappIntelligenceConsumerType::CUSTOM)
	->setConsumer(new KafkaTrackingConsumer($conf, 'mapp-intelligence-requests'));
$mappIntelligence = MappIntelligence::getInstance($mappIntelligenceConfig);

Redis

Appends the requests to a Redis list (PHP extension redis). rPush() with all requests of the batch is one command, so the batch is appended completely or not at all. A RedisException (Redis not reachable) counts as RETRY.

class RedisTrackingConsumer implements MappIntelligenceConsumer
{
	private $redis;
	private $key;

	/**
	 * @param \Redis $redis Connected Redis client
	 * @param string $key Name of the list
	 */
	public function __construct(\Redis $redis, $key) {
		$this->redis = $redis;
		$this->key = $key;
	}

	public function sendBatch(array $batchContent) {
		return $this->redis->rPush($this->key, ...$batchContent) !== false;
	}
}

$redis = new \Redis();
// persistent connection, reused by the following requests of this PHP process; timeout in seconds
$redis->pconnect('redis', 6379, 1.0);

$mappIntelligenceConfig = new MappIntelligenceConfig();
$mappIntelligenceConfig->setTrackDomain("analytics01.wt-eu02.net")
    ->setTrackId("111111111111111")
	->setConsumerType(MappIntelligenceConsumerType::CUSTOM)
	->setConsumer(new RedisTrackingConsumer($redis, 'mapp-intelligence-requests'));
$mappIntelligence = MappIntelligence::getInstance($mappIntelligenceConfig);

Forwarding the requests from Kafka or Redis

Kafka and Redis only buffer the requests. A separate process of your own (a worker) has to read them and send them to the track server, the same way the CURL consumer does:

• POST to https://<trackDomain>/<trackId>/batch with the content type text/plain, one request per line in the body.

• At most 10,000 requests and less than 24 MB per batch.

• Handle the response like the CURL consumer (see the table of batch results above): a status below 400 means delivered; 5xx, 408, 429 and network errors are retried; any other 4xx can never be delivered.

• Commit the Kafka offset, or remove the requests from the Redis list, only after the batch was delivered. With Redis and one worker per list, for example read with LRANGE <key> 0 9999 and after delivery remove the batch with LTRIM <key> <number of requests> -1.

Note

If a batch is retried after part of it already reached Kafka or Redis (for example a timeout after the broker has stored the message), these requests are buffered twice. Kafka and Redis do not prevent this, and neither does the worker. If duplicates matter, deduplicate in the worker.

Test consumer

Keeps the requests in memory instead of sending them, so you can check in your own unit tests what the library would track, without network or files. $available = false simulates an unreachable track server: every batch is then RETRY and stays in the queue, and flush() returns false.

use MappIntelligence\MappIntelligenceConsumer;

class InMemoryTestConsumer implements MappIntelligenceConsumer
{
	public $requests = array();
	public $available = true;

	public function sendBatch(array $batchContent) {
		if (!$this->available) {
			return false;
		}

		array_push($this->requests, ...$batchContent);
		return true;
	}
}

Usage in a PHPUnit test. Use one instance per test instead of the shared getInstance(), so no request of a previous test is left over. flush() sends synchronously, so the requests are in the consumer once it has returned:

use MappIntelligence\MappIntelligence;
use MappIntelligence\Config\MappIntelligenceConfig;
use MappIntelligence\Consumer\MappIntelligenceConsumerType;
use MappIntelligence\Data\MappIntelligenceDataMap;
use MappIntelligence\Data\MappIntelligencePage;

// in your test class
public function testTrackPage()
{
	$consumer = new InMemoryTestConsumer();
	$mappIntelligenceConfig = new MappIntelligenceConfig();
	$mappIntelligenceConfig->setTrackDomain("analytics01.wt-eu02.net")
		->setTrackId("111111111111111")
		->setConsumerType(MappIntelligenceConsumerType::CUSTOM)
		->setConsumer($consumer);

	$mappIntelligence = new MappIntelligence($mappIntelligenceConfig);
	$mappIntelligence->track((new MappIntelligenceDataMap())->page(new MappIntelligencePage('en.home')));
	$mappIntelligence->flush();

	$this->assertCount(1, $consumer->requests);
	$this->assertStringContainsString('en.home', $consumer->requests[0]);
}

Please check Configuration to see all configuration options and explanations of the Mapp Intelligence PHP library, including default settings.