Data Transfer: Consumer

Consumers normally send tracking requests to the Intelligence tracking server. The options to do this are:

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

• FORK_CURL: Saves tracking requests in memory and sends them in a separate, forked process to the server​.

• FILE: Best for high traffic websites. Requests are saved in a separate file and sent via a cronjob to the tracking server in a specified interval.

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

All consumers are fed by the request queue of the tracking instance. Adding a request does not send it: the queue is sent when flush() is called, every requestInterval seconds if an interval is configured, when the queue reaches maxQueueSize and when the tracking instance is closed (see Classes). A batch contains at most maxBatchSize requests and is closed at about 20 MB request body.

What happens with a batch is decided by the consumer:

Note

Requests that are still in the queue when the tracking instance is closed get one last flush with up to maxAttempt attempts, within the 30-second deadline of close(). Whatever is still undelivered after that is lost. When the queue reaches maxQueueSize, it is sent; if that fails, the oldest requests are discarded (logged at level WARN), at least maxBatchSize of them, so the next attempt is made only after that many new requests. 0 means unlimited.

const config = new MappIntelligenceConfig()
    .setTrackDomain("analytics01.wt-eu02.net")
    .setTrackId("111111111111111")
    .setConsumerType(MappIntelligenceConsumerType.HTTP_CLIENT)
    .setMaxAttempt(1)
    .setAttemptTimeout(100)
    .setMaxBatchSize(50)
    .setMaxQueueSize(1000)
    .setRequestInterval(0);

const config = new MappIntelligenceConfig()
    .setTrackDomain("analytics01.wt-eu02.net")
    .setTrackId("111111111111111")
    .setConsumerType(MappIntelligenceConsumerType.FORK_CURL)
    .setMaxAttempt(1)
    .setAttemptTimeout(100)
    .setMaxBatchSize(50)
    .setMaxQueueSize(1000);

const config = new MappIntelligenceConfig()
    .setTrackDomain("analytics01.wt-eu02.net")
    .setTrackId("111111111111111")
    .setConsumerType(MappIntelligenceConsumerType.FILE)
	.setFilePath("path/to/file")
	.setFilePrefix("MappIntelligenceRequests")
	.setMaxAttempt(1)
    .setAttemptTimeout(100)
	.setMaxFileLines(5000)
	.setMaxFileDuration(30 * 60 * 1000)
	.setMaxFileSize(12 * 1024 * 1024);

class TestCustomConsumer implements IMappIntelligenceConsumer {
    public async sendBatch(batchContent: Array<string>): Promise<boolean> {
        // implement custom consumer logic: true = SENT, false = RETRY
        return true;
    }
}
const mappIntelligenceConfig = new MappIntelligenceConfig()
    .setTrackDomain("analytics01.wt-eu02.net")
    .setTrackId("111111111111111")
    .setConsumerType(MappIntelligenceConsumerType.CUSTOM)
    .setConsumer(new TestCustomConsumer());

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 (client library kafkajs). producer.send() resolves only when the broker has confirmed every message of the batch. If it rejects, the library logs the error and treats the batch as RETRY.

import {Kafka, Producer} from 'kafkajs';

class KafkaTrackingConsumer implements IMappIntelligenceConsumer {
    public constructor(private readonly producer: Producer, private readonly topic: string) {}

    public async sendBatch(batchContent: Array<string>): Promise<boolean> {
        await this.producer.send({
            topic: this.topic,
            messages: batchContent.map((request) => ({value: request}))
        });

        return true;
    }
}

// one producer per process, disconnect it yourself when the process stops
const producer = new Kafka({clientId: 'shop', brokers: ['kafka:9092']}).producer();
await producer.connect();

const mappIntelligenceConfig = new MappIntelligenceConfig()
    .setTrackDomain("analytics01.wt-eu02.net")
    .setTrackId("111111111111111")
    .setConsumerType(MappIntelligenceConsumerType.CUSTOM)
    .setConsumer(new KafkaTrackingConsumer(producer, 'mapp-intelligence-requests'));

Redis

Appends the requests to a Redis list (client library ioredis). RPUSH with all requests of the batch is one command, so the batch is appended completely or not at all. If it rejects, the library logs the error and treats the batch as RETRY.

import {Redis} from 'ioredis';

class RedisTrackingConsumer implements IMappIntelligenceConsumer {
    public constructor(private readonly redis: Redis, private readonly key: string) {}

    public async sendBatch(batchContent: Array<string>): Promise<boolean> {
        await this.redis.rpush(this.key, ...batchContent);

        return true;
    }
}

// one client per process, quit it yourself when the process stops;
// maxRetriesPerRequest: 1 rejects quickly while Redis is not reachable, instead of keeping the flush waiting
const redis = new Redis({host: 'redis', port: 6379, maxRetriesPerRequest: 1});

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

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 HTTP_CLIENT 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 HTTP_CLIENT 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.

class InMemoryTestConsumer implements IMappIntelligenceConsumer {
    public readonly requests: Array<string> = [];
    public available: boolean = true;

    public async sendBatch(batchContent: Array<string>): Promise<boolean> {
        if (!this.available) {
            return false;
        }

        this.requests.push(...batchContent);
        return true;
    }
}

Usage in a Jest test. close() sends the remaining requests, so they are in the consumer once it has resolved:

test('tracks the page', async () => {
    const consumer = new InMemoryTestConsumer();
    const mappIntelligenceConfig = new MappIntelligenceConfig()
        .setTrackDomain("analytics01.wt-eu02.net")
        .setTrackId("111111111111111")
        .setConsumerType(MappIntelligenceConsumerType.CUSTOM)
        .setConsumer(consumer);

    const mappIntelligence = new MappIntelligenceTracking(mappIntelligenceConfig);
    await mappIntelligence.track(new MappIntelligenceDataMap().page(new MappIntelligencePage('en.home')));
    await mappIntelligence.close();

    expect(consumer.requests).toHaveLength(1);
    expect(consumer.requests[0]).toContain('en.home');
});