<commit_before>'use strict';

var P = require('bluebird');
var sUtil = require('../lib/util');
var kafka = P.promisifyAll(require('kafka-node'));
var HighLevelProducer = kafka.HighLevelProducer;
var KeyedMessage = kafka.KeyedMessage;

var options = {
    connectionString: undefined, //"kafka-event-bus.services.eqiad.wmflabs:2181/kafka/kafka-event-bus",
    clientId: 0
};

var client = new kafka.Client(options.connectionString,
        options.clientId, options);
var producer = new HighLevelProducer(client);

producer.on('ready', function() {
    console.log('kafka ready');
});

producer.on('error', console.log);


// shortcut
var HTTPError = sUtil.HTTPError;

/**
 * The main router object
 */
var router = sUtil.router();

/**
 * The main application object reported when this module is require()d
 */
var app;

var exampleSchema = {
	"title": "Example Schema",
	"type": "object",
	"properties": {
		"url": {
			"type": "string"
		},
		"type": {
			"type": "string"
		},
		"details": {
			"type": "string"
		}
	},
	"required": ["url", "type"]
};

var ajv = require('ajv')();
// add each validator
var validateMessage = ajv.compile(exampleSchema);


/**
 * PUT /topics/{name}
 * Create / update a topic. Accepts a JSON schema.
 */
router.put('/topics/:name', function(req, res) {
    producer.createTopics([req.params.name]);
    res.status(200).send('Topic created');
});
/*
 * Expected layout
 * {
 *    topic: 'topicName',
 *    messages: ['message body'],// multi messages should be a array, single message can be just a string or a KeyedMessage instance
 *    // These might be best controlled by the proxy
 *    partition: 0, //default 0
 *    attributes: 2, // default: 0
 * }
 */

function validateMessages(topic, messages) {
    messages = messages.map(function(msg) {
        if (!validateMessage(msg)) {
            throw new HTTPError({
                status: 400,
                body: {
                    type: 'invalid_message',
                    msg: msg
                }
            });
        }
        return JSON.stringify(msg);
    });
    return {
        topic: topic,
        messages: messages,
        attributes: 2, // snappy compression
    };
}


/**
 * POST /topics/{name}/
 * Enqueue one or more events. Each event needs to conform to the JSON schema
 * associated with this topic.
 */
router.post('/topics/:name', function(req, res) {
    return P.try(function() {
        var message = validateMessages(req.params.name, req.body);<commit_msg>Better validation error reporting<commit_after>'use strict';

var P = require('bluebird');
var sUtil = require('../lib/util');
var kafka = P.promisifyAll(require('kafka-node'));
var HighLevelProducer = kafka.HighLevelProducer;
var KeyedMessage = kafka.KeyedMessage;

var options = {
    connectionString: undefined, //"kafka-event-bus.services.eqiad.wmflabs:2181/kafka/kafka-event-bus",
    clientId: 0
};

var client = new kafka.Client(options.connectionString,
        options.clientId, options);
var producer = new HighLevelProducer(client);

producer.on('ready', function() {
    console.log('kafka ready');
});

producer.on('error', console.log);


// shortcut
var HTTPError = sUtil.HTTPError;

/**
 * The main router object
 */
var router = sUtil.router();

/**
 * The main application object reported when this module is require()d
 */
var app;

var exampleSchema = {
	"title": "Example Schema",
	"type": "object",
	"properties": {
		"url": {
			"type": "string"
		},
		"type": {
			"type": "string"
		},
		"details": {
			"type": "string"
		}
	},
	"required": ["url", "type"]
};

var ajv = require('ajv')();
// add each validator
var validateMessage = ajv.compile(exampleSchema);


/**
 * PUT /topics/{name}
 * Create / update a topic. Accepts a JSON schema.
 */
router.put('/topics/:name', function(req, res) {
    producer.createTopics([req.params.name]);
    res.status(200).send('Topic created');
});
/*
 * Expected layout
 * {
 *    topic: 'topicName',
 *    messages: ['message body'],// multi messages should be a array, single message can be just a string or a KeyedMessage instance
 *    // These might be best controlled by the proxy
 *    partition: 0, //default 0
 *    attributes: 2, // default: 0
 * }
 */

function validateMessages(topic, messages) {
    messages = messages.map(function(msg) {
        if (!validateMessage(msg)) {
            throw new HTTPError({
                status: 400,
                body: {
                    type: 'invalid_message',
                    original_message: msg,
                    validation_error: validateMessage.errors,
                }
            });
        }
        return JSON.stringify(msg);
    });
    return {
        topic: topic,
        messages: messages,
        attributes: 2, // snappy compression
    };
}


/**
 * POST /topics/{name}/
 * Enqueue one or more events. Each event needs to conform to the JSON schema
 * associated with this topic.
 */
router.post('/topics/:name', function(req, res) {
    return P.try(function() {
        var message = validateMessages(req.params.name, req.body);
{'fraction_non_alphanumeric': 0.14933018124507486, 'fraction_numerical': 0.006304176516942475, 'mean_word_length': 2.802996254681648, 'pattern_counts': {'":': 20, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2427875', 'n_tokens_mistral': 1604, 'n_tokens_neox': 1549, 'n_words': 414}