<commit_before>var setup = require('../../setup');
var assert = require('assert');

describe('routes/{clusterListenForTasks,clusterStartTask}', function () {

	it('works', function (done) {

		this.slow(1000);
		this.timeout(2000);

		var instance = setup();
		instance.start();

		var actual = { tasks: [], responses: [] };
		var expect = { tasks: [], responses: [] };
		var close = { tasks: [], responses: [] };

		var buckets = ['b1','b2','b3'];
		var waiting = 1 + (buckets.length * 2);
		buckets.forEach(function (bucket) {
			expect.tasks.push({ b: bucket });
			expect.responses.push({ r: bucket });
		});

		// Once all messages have been sent,
		// verify that the data expected was sent/received.
		function complete() {
			actual.responses = actual.responses.sort(function (a, b) {
				return a.r.localeCompare(b.r);
			});
			actual.tasks = actual.tasks.sort(function (a, b) {
				return a.b.localeCompare(b.b);
			});
			assert.deepEqual(expect, actual);
			instance.stop();
			done();
		}

		// Ensure no buckets results in 400.
		instance.post('/listeners/', { bad: true }, {
			response: function (res) {
				assert(res.statusCode === 400);
				assert(res.headers['x-postable-cluster-id']);
				--waiting || complete();
			}
		});

		// Hook up a listener to all buckets.
		instance.post('/listeners/', { buckets: buckets }, {
			listener: function (item) {
				if (item.ignoreHeartbeat) return;
				// Verify the messages coming to the listener are correct.
				assert(item);
				assert(item.id);
				assert(item.listenerId);
				assert(item.data);
				assert(item.data.task);
				assert('number' === typeof item.data.index);
				actual.tasks.push(item.data.task);
				// Respond to the messages with the expected responses.
				instance.post('/tasks/' + item.id + '/results/' + item.listenerId, expect.responses[item.data.index]);
				--waiting || complete();
			},
			done: function (reason) {
				close.responses.push(reason);
			}
		});
		// Put a task on each bucket.
		setTimeout(function () {
			expect.tasks.forEach(function (t, i) {
				instance.post('/buckets/' + t.b + '/tasks/', { task: t, index: i }, {
					listener: function (item) {
						if (item.ignoreHeartbeat) return;
						// Verify that responses coming from the listeners look correct.
						assert(item);
						if (item.meta) {
							assert(item.meta.listenersPending && item.meta.listenersPending.length === 1);
							return;
						}
						assert(item.data);
						assert(item.listener);
						assert(item.listener.id);
						assert.deepEqual(buckets, item.listener.buckets);
						assert(item.timeout === false);
						actual.responses.push(item.data);
						--waiting || complete();
					},
					done: function (reason) {
						close.tasks.push(reason);
					}
				})
			});
		}, 5);

	});


	it('times out listeners', function (done) {

		this.slow(3000);
		this.timeout(2000);

		var instance = setup();
		instance.start();

		// Hook up a listener to all buckets.
		instance.post('/listeners/', { buckets: ['bt1'] }, {
			listener: function (item) {
				if (item.ignoreHeartbeat) return;
				instance.post('/tasks/' + item.id + '/results/' + item.listenerId, { foo: 'bar' });
			}
		});<commit_msg>Extend timeouts during tests for race conditions.<commit_after>var setup = require('../../setup');
var assert = require('assert');

describe('routes/{clusterListenForTasks,clusterStartTask}', function () {

	it('works', function (done) {

		this.slow(1000);
		this.timeout(2000);

		var instance = setup();
		instance.start();

		var actual = { tasks: [], responses: [] };
		var expect = { tasks: [], responses: [] };
		var close = { tasks: [], responses: [] };

		var buckets = ['b1','b2','b3'];
		var waiting = 1 + (buckets.length * 2);
		buckets.forEach(function (bucket) {
			expect.tasks.push({ b: bucket });
			expect.responses.push({ r: bucket });
		});

		// Once all messages have been sent,
		// verify that the data expected was sent/received.
		function complete() {
			actual.responses = actual.responses.sort(function (a, b) {
				return a.r.localeCompare(b.r);
			});
			actual.tasks = actual.tasks.sort(function (a, b) {
				return a.b.localeCompare(b.b);
			});
			assert.deepEqual(expect, actual);
			instance.stop();
			done();
		}

		// Ensure no buckets results in 400.
		instance.post('/listeners/', { bad: true }, {
			response: function (res) {
				assert(res.statusCode === 400);
				assert(res.headers['x-postable-cluster-id']);
				--waiting || complete();
			}
		});

		// Hook up a listener to all buckets.
		instance.post('/listeners/', { buckets: buckets }, {
			listener: function (item) {
				if (item.ignoreHeartbeat) return;
				// Verify the messages coming to the listener are correct.
				assert(item);
				assert(item.id);
				assert(item.listenerId);
				assert(item.data);
				assert(item.data.task);
				assert('number' === typeof item.data.index);
				actual.tasks.push(item.data.task);
				// Respond to the messages with the expected responses.
				instance.post('/tasks/' + item.id + '/results/' + item.listenerId, expect.responses[item.data.index]);
				--waiting || complete();
			},
			done: function (reason) {
				close.responses.push(reason);
			}
		});
		// Put a task on each bucket.
		setTimeout(function () {
			expect.tasks.forEach(function (t, i) {
				instance.post('/buckets/' + t.b + '/tasks/', { task: t, index: i }, {
					listener: function (item) {
						if (item.ignoreHeartbeat) return;
						// Verify that responses coming from the listeners look correct.
						assert(item);
						if (item.meta) {
							assert(item.meta.listenersPending && item.meta.listenersPending.length === 1);
							return;
						}
						assert(item.data);
						assert(item.listener);
						assert(item.listener.id);
						assert.deepEqual(buckets, item.listener.buckets);
						assert(item.timeout === false);
						actual.responses.push(item.data);
						--waiting || complete();
					},
					done: function (reason) {
						close.tasks.push(reason);
					}
				})
			});
		}, 5);

	});


	it('times out listeners', function (done) {

		this.slow(5000);
		this.timeout(5000);

		var instance = setup();
		instance.start();

		// Hook up a listener to all buckets.
		instance.post('/listeners/', { buckets: ['bt1'] }, {
			listener: function (item) {
				if (item.ignoreHeartbeat) return;
				instance.post('/tasks/' + item.id + '/results/' + item.listenerId, { foo: 'bar' });
			}
		});
{'fraction_non_alphanumeric': 0.1736629353233831, 'fraction_numerical': 0.009328358208955223, 'mean_word_length': 2.9466257668711657, 'pattern_counts': {'":': 0, '<': 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': '134050', 'n_tokens_mistral': 2555, 'n_tokens_neox': 2000, 'n_words': 472}