<commit_before>// deps
const Promise = require('bluebird');
const uuid = require('uuid');
const stringify = require('json-stringify-safe');
const EventEmitter = require('eventemitter3');
const os = require('os');
const is = require('is');
const assert = require('assert');
const opentracing = require('opentracing');
const {
  ConnectionError,
  NotPermittedError,
  ValidationError,
  InvalidOperationError,
  ArgumentError,
} = require('common-errors');

// lodash fp
const merge = require('lodash/merge');
const defaults = require('lodash/defaults');
const noop = require('lodash/noop');
const uniq = require('lodash/uniq');
const extend = require('lodash/extend');
const pick = require('lodash/pick');
const set = require('lodash/set');

// local deps
const Joi = require('joi');
const schema = require('./schema');
const pkg = require('../package.json');
const AMQP = require('./utils/transport');
const ReplyStorage = require('./utils/reply-storage');
const Backoff = require('./utils/recovery');
const Cache = require('./utils/cache');
const latency = require('./utils/latency');
const loggerUtils = require('./loggers');
const generateErrorMessage = require('./utils/error');
const helpers = require('./helpers');

// serialization functions
const { jsonSerializer, jsonDeserializer } = require('./utils/serialization');

// cache references
const { AmqpDLXError } = generateErrorMessage;
const { closeConsumer, wrapError, setQoS } = helpers;
const { Tags, FORMAT_TEXT_MAP } = opentracing;

// wrap promise
const wrapPromise = (span, promise) => (
  promise
    .catch((error) => {
      span.setTag(Tags.ERROR, true);
      span.log({ event: 'error', 'error.object': error, message: error.message, stack: error.stack });
      throw error;
    })
    .finally(() => {
      span.finish();
    })
);

/**
 * Routing function HOC with reply RPC enhancer
 * @param  {Function} messageHandler
 * @param  {AMQPTransport} transport
 * @returns {Function}
 */
const initRoutingFn = (messageHandler, transport) => {
  /**
   * Initiates consumer message handler.
   * @param  {Mixed} message - Data passed from the publisher.
   * @param  {Object} properties - AMQP Message properties.
   * @param  {Object} raw - Original AMQP message.
   * @param  {Function} [raw.ack] - Acknowledge if nack is `true`.
   * @param  {Function} [raw.reject] - Reject if nack is `true`.
   * @param  {Function} [raw.retry] - Retry msg if nack is `true`.
   * @returns {Void}
   */
  return function router(message, properties, raw) {
    // add instrumentation
    const appId = transport._parseInput(properties.appId);

    // opentracing instrumentation
    const childOf = this.tracer.extract(FORMAT_TEXT_MAP, properties.headers || {});
    const span = this.tracer.startSpan(`onConsume:${properties.routingKey}`, {
      childOf,
    });

    span.addTags({
      [Tags.SPAN_KIND]: Tags.SPAN_KIND_RPC_SERVER,
      [Tags.PEER_SERVICE]: appId.name,
      [Tags.PEER_HOSTNAME]: appId.host,
    });

    const next = !properties.replyTo || !properties.correlationId
      ? (error, data) => transport.noop(error, data, span)
      : (error, data) => transport.reply(properties, { error, data }, span);

    // define span in the original message
    // so that userland has access to it
    raw.span = span;

    return messageHandler(message, properties, raw, next);
  };<commit_msg>feat(initRoutingFn): evaluate replyTo/correlationId at runtime<commit_after>// deps
const Promise = require('bluebird');
const uuid = require('uuid');
const stringify = require('json-stringify-safe');
const EventEmitter = require('eventemitter3');
const os = require('os');
const is = require('is');
const assert = require('assert');
const opentracing = require('opentracing');
const {
  ConnectionError,
  NotPermittedError,
  ValidationError,
  InvalidOperationError,
  ArgumentError,
} = require('common-errors');

// lodash fp
const merge = require('lodash/merge');
const defaults = require('lodash/defaults');
const noop = require('lodash/noop');
const uniq = require('lodash/uniq');
const extend = require('lodash/extend');
const pick = require('lodash/pick');
const set = require('lodash/set');

// local deps
const Joi = require('joi');
const schema = require('./schema');
const pkg = require('../package.json');
const AMQP = require('./utils/transport');
const ReplyStorage = require('./utils/reply-storage');
const Backoff = require('./utils/recovery');
const Cache = require('./utils/cache');
const latency = require('./utils/latency');
const loggerUtils = require('./loggers');
const generateErrorMessage = require('./utils/error');
const helpers = require('./helpers');

// serialization functions
const { jsonSerializer, jsonDeserializer } = require('./utils/serialization');

// cache references
const { AmqpDLXError } = generateErrorMessage;
const { closeConsumer, wrapError, setQoS } = helpers;
const { Tags, FORMAT_TEXT_MAP } = opentracing;

// wrap promise
const wrapPromise = (span, promise) => (
  promise
    .catch((error) => {
      span.setTag(Tags.ERROR, true);
      span.log({ event: 'error', 'error.object': error, message: error.message, stack: error.stack });
      throw error;
    })
    .finally(() => {
      span.finish();
    })
);

/**
 * Routing function HOC with reply RPC enhancer
 * @param  {Function} messageHandler
 * @param  {AMQPTransport} transport
 * @returns {Function}
 */
const initRoutingFn = (messageHandler, transport) => {
  /**
   * Response Handler Function. Sends Reply or Noop log.
   * @param  {AMQPMessage} raw - Raw AMQP Message Structure
   * @param  {Error} error - Error if it happened.
   * @param  {Mixed} data - Response data.
   * @returns {Promise<*>}
   */
  function responseHandler(raw, error, data) {
    const { properties, span } = raw;
    return !properties.replyTo || !properties.correlationId
      ? transport.noop(error, data, span)
      : transport.reply(properties, { error, data }, span);
  }

  /**
   * Initiates consumer message handler.
   * @param  {Mixed} message - Data passed from the publisher.
   * @param  {Object} properties - AMQP Message properties.
   * @param  {Object} raw - Original AMQP message.
   * @param  {Function} [raw.ack] - Acknowledge if nack is `true`.
   * @param  {Function} [raw.reject] - Reject if nack is `true`.
   * @param  {Function} [raw.retry] - Retry msg if nack is `true`.
   * @returns {Void}
   */
  return function router(message, properties, raw) {
    // add instrumentation
    const appId = transport._parseInput(properties.appId);

    // opentracing instrumentation
    const childOf = this.tracer.extract(FORMAT_TEXT_MAP, properties.headers || {});
    const span = this.tracer.startSpan(`onConsume:${properties.routingKey}`, {
      childOf,
    });

    span.addTags({
      [Tags.SPAN_KIND]: Tags.SPAN_KIND_RPC_SERVER,
      [Tags.PEER_SERVICE]: appId.name,
      [Tags.PEER_HOSTNAME]: appId.host,
    });

    // define span in the original message
    // so that userland has access to it
    raw.span = span;

    return messageHandler(message, properties, raw, responseHandler.bind(undefined, raw));
  };
{'fraction_non_alphanumeric': 0.14924114671163574, 'fraction_numerical': 0.0002810567734682406, 'mean_word_length': 3.9150552486187844, 'pattern_counts': {'":': 0, '<': 4, '<?xml version=': 0, '>': 14, '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': '3890781', 'n_tokens_mistral': 2152, 'n_tokens_neox': 2114, 'n_words': 569}