<commit_before>let debug = require('debug')('taskcluster-github');
let intree = require('./intree');
let taskcluster = require('taskcluster-client');
let slugid = require('slugid');
let yaml = require('js-yaml');
let assert = require('assert');
let _ = require('lodash');

let INSPECTOR_URL = 'https://tools.taskcluster.net/task-graph-inspector/#';
let STATUS_MAPPING = {
  running:  'pending',
  blocked:  'failure',
  finished: 'success',
};

/**
 * Create handlers
 *
 * options:
 * {
 *   credentials:        // Pulse credentials
 *   queueName:          // Queue name (optional)
 *   reference:          // A taskcluster exchange reference for this service
 *   monitor:            // base.monitor({...})
 *   jobQueueName:       // An optional queue name for the jobs handler
 *   statusQueueName:    // An optional queue name for the status handler
 *   context:            // Things we want to make available inside handlers
 * }
 */
class Handlers {
  constructor({credentials, monitor, reference, jobQueueName, statusQueueName, context}) {
    assert(credentials.username, 'credentials.username must be provided');
    assert(credentials.password, 'credentials.password must be provided');
    assert(monitor, 'monitor is required for statistics');
    assert(monitor, 'reference must be provided');
    this.credentials = credentials;
    this.monitor = monitor;
    this.reference = reference;
    this.connection = null;
    this.statusListener = null;
    this.jobListener = null;
    this.statusQueueName = statusQueueName;  // Optional
    this.jobQueueName = jobQueueName;  // Optional
    this.context = context;
  }

  async setup() {
    assert(this.connection === null, 'Cannot setup twice!');
    this.connection = new taskcluster.PulseConnection(this.credentials);
    this.statusListener = new taskcluster.PulseListener({
      queueName: this.statusQueueName,
      connection: this.connection,
    });
    this.jobListener = new taskcluster.PulseListener({
      queueName: this.jobQueueName,
      connection: this.connection,
    });

    // Listen for new jobs created via the api webhook endpoint
    let GithubEvents = taskcluster.createClient(this.reference);
    let githubEvents = new GithubEvents();
    await this.jobListener.bind(githubEvents.pullRequest({organization: '*', repository: '*', action: '*'}));
    await this.jobListener.bind(githubEvents.push({organization: '*', repository: '*'}));

    // Listen for state changes to the taskcluster taskgraph
    let schedulerEvents = new taskcluster.SchedulerEvents();
    let route = 'route.taskcluster-github.*.*.*';
    await this.statusListener.bind(schedulerEvents.taskGraphRunning(route));
    await this.statusListener.bind(schedulerEvents.taskGraphBlocked(route));
    await this.statusListener.bind(schedulerEvents.taskGraphFinished(route));

    this.jobListener.on('message', this.monitor.timedHandler('joblistener', jobHandler.bind(this)));
    this.statusListener.on('message', this.monitor.timedHandler('statuslistener', statusHandler.bind(this)));

    await this.jobListener.connect();
    await this.statusListener.connect();

    // If this is awaited, it should return [undefined, undefined]
    return Promise.all([this.jobListener.resume(), this.statusListener.resume()]);
  }

  async terminate() {
    await this.connection.close();
  }
}
module.exports = Handlers;

/**
 * Post updates to GitHub, when the status of a task changes.
 * TaskCluster States: http://docs.taskcluster.net/queue/exchanges/
 * GitHub Statuses: https://developer.github.com/v3/repos/statuses/
 **/
async function statusHandler(message) {
  debug('handling state change for message: ', message);
  let route = message.routes[0].split('.');
  try {
    await context.github.repos.createStatus({
      user: route[1],
      repo: route[2],
      sha: route[3],
      state: STATUS_MAPPING[message.payload.status.state],
      target_url: INSPECTOR_URL + message.payload.status.taskGraphId + '/',
      description: 'TaskGraph: ' + message.payload.status.state,
      context: 'Taskcluster',
    });
  } catch (e) {
    debug(`Failed to update status: ${route[1]}/${route[2]}@${route[3]}`);
    throw e;
  }
}<commit_msg>Fix missed context reference in handlers<commit_after>let debug = require('debug')('taskcluster-github');
let intree = require('./intree');
let taskcluster = require('taskcluster-client');
let slugid = require('slugid');
let yaml = require('js-yaml');
let assert = require('assert');
let _ = require('lodash');

let INSPECTOR_URL = 'https://tools.taskcluster.net/task-graph-inspector/#';
let STATUS_MAPPING = {
  running:  'pending',
  blocked:  'failure',
  finished: 'success',
};

/**
 * Create handlers
 *
 * options:
 * {
 *   credentials:        // Pulse credentials
 *   queueName:          // Queue name (optional)
 *   reference:          // A taskcluster exchange reference for this service
 *   monitor:            // base.monitor({...})
 *   jobQueueName:       // An optional queue name for the jobs handler
 *   statusQueueName:    // An optional queue name for the status handler
 *   context:            // Things we want to make available inside handlers
 * }
 */
class Handlers {
  constructor({credentials, monitor, reference, jobQueueName, statusQueueName, context}) {
    assert(credentials.username, 'credentials.username must be provided');
    assert(credentials.password, 'credentials.password must be provided');
    assert(monitor, 'monitor is required for statistics');
    assert(monitor, 'reference must be provided');
    this.credentials = credentials;
    this.monitor = monitor;
    this.reference = reference;
    this.connection = null;
    this.statusListener = null;
    this.jobListener = null;
    this.statusQueueName = statusQueueName;  // Optional
    this.jobQueueName = jobQueueName;  // Optional
    this.context = context;
  }

  async setup() {
    assert(this.connection === null, 'Cannot setup twice!');
    this.connection = new taskcluster.PulseConnection(this.credentials);
    this.statusListener = new taskcluster.PulseListener({
      queueName: this.statusQueueName,
      connection: this.connection,
    });
    this.jobListener = new taskcluster.PulseListener({
      queueName: this.jobQueueName,
      connection: this.connection,
    });

    // Listen for new jobs created via the api webhook endpoint
    let GithubEvents = taskcluster.createClient(this.reference);
    let githubEvents = new GithubEvents();
    await this.jobListener.bind(githubEvents.pullRequest({organization: '*', repository: '*', action: '*'}));
    await this.jobListener.bind(githubEvents.push({organization: '*', repository: '*'}));

    // Listen for state changes to the taskcluster taskgraph
    let schedulerEvents = new taskcluster.SchedulerEvents();
    let route = 'route.taskcluster-github.*.*.*';
    await this.statusListener.bind(schedulerEvents.taskGraphRunning(route));
    await this.statusListener.bind(schedulerEvents.taskGraphBlocked(route));
    await this.statusListener.bind(schedulerEvents.taskGraphFinished(route));

    this.jobListener.on('message', this.monitor.timedHandler('joblistener', jobHandler.bind(this)));
    this.statusListener.on('message', this.monitor.timedHandler('statuslistener', statusHandler.bind(this)));

    await this.jobListener.connect();
    await this.statusListener.connect();

    // If this is awaited, it should return [undefined, undefined]
    return Promise.all([this.jobListener.resume(), this.statusListener.resume()]);
  }

  async terminate() {
    await this.connection.close();
  }
}
module.exports = Handlers;

/**
 * Post updates to GitHub, when the status of a task changes.
 * TaskCluster States: http://docs.taskcluster.net/queue/exchanges/
 * GitHub Statuses: https://developer.github.com/v3/repos/statuses/
 **/
async function statusHandler(message) {
  debug('handling state change for message: ', message);
  let route = message.routes[0].split('.');
  try {
    await this.context.github.repos.createStatus({
      user: route[1],
      repo: route[2],
      sha: route[3],
      state: STATUS_MAPPING[message.payload.status.state],
      target_url: INSPECTOR_URL + message.payload.status.taskGraphId + '/',
      description: 'TaskGraph: ' + message.payload.status.state,
      context: 'Taskcluster',
    });
  } catch (e) {
    debug(`Failed to update status: ${route[1]}/${route[2]}@${route[3]}`);
    throw e;
  }
}
{'fraction_non_alphanumeric': 0.1342099028666193, 'fraction_numerical': 0.0018952854773750297, 'mean_word_length': 3.78356940509915, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 4, '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': '179862', 'n_tokens_mistral': 2399, 'n_tokens_neox': 2366, 'n_words': 569}