<commit_before>'use strict'

const Hp = require('hemera-plugin')
const SafeStringify = require('fast-safe-stringify')
const SafeParse = require('json-parse-safe')
const Nats = require('node-nats-streaming')

exports.plugin = Hp(function hemeraNatsStreaming (options, next) {
  const hemera = this
  const topic = 'nats-streaming'
  const Joi = hemera.exposition['hemera-joi'].joi
  const DuplicateSubscriberError = hemera.createError('DuplicateSubscriber')
  const ParsingError = hemera.createError('ParsingError')
  const NotAvailableError = hemera.createError('NotAvailable')
  const stan = Nats.connect(options.clusterId, options.clientId, options.opts)
  const subList = {}

  hemera.ext('onClose', (next) => {
    hemera.log.debug('Stan closing ...')
    stan.close()
    next()
  })

  stan.on('error', (err) => {
    hemera.log.error(err)
    hemera.fatal()
  })

  stan.on('connect', function () {
    /**
     * Publish a message over NATS-Streaming server
     */
    hemera.add({
      topic,
      cmd: 'publish',
      subject: Joi.string().required(),
      data: Joi.alternatives().try(Joi.object(), Joi.array())
    }, function (req, reply) {
      function handler (err, guid) {
        if (err) {
          reply(err)
        } else {
          reply(null, guid)
        }
      }

      stan.publish(req.subject, SafeStringify(req.data), handler)
    })

    /**
     * Create a subscription on the NATS-Streaming server
     * All options are available expect manual acknowledgement because we use it as a proxy
     */
    hemera.add({
      topic,
      cmd: 'subscribe',
      subject: Joi.string().required(),
      queue: Joi.string().optional(),
      options: Joi.object().keys({
        setStartWithLastReceived: Joi.boolean(), // Subscribe starting with the most recently published value
        setDeliverAllAvailable: Joi.boolean(), // Receive all stored values in order
        setStartAtSequence: Joi.number().integer(), // Receive all messages starting at a specific sequence number
        setStartTime: Joi.date().iso(), // Subscribe starting at a specific time
        setStartAtTimeDelta: Joi.number().integer(),
        setDurableName: Joi.string(), // Create a durable subscription
        setMaxInFlight: Joi.number().integer(), // the maximum number of outstanding acknowledgements
        setManualAckMode: Joi.boolean().default(true),
        setAckWait: Joi.number().integer() // if an acknowledgement is not received within the configured timeout interval, NATS Streaming will attempt redelivery of the message (default 30 seconds)
      }).default()
    }, function (req, reply) {
      // avoid multiple subscribers for the same subject
      if (subList[req.subject]) {
        reply(new DuplicateSubscriberError(`Subscription for subject "${req.subject}" is already active`))
        return
      }

      const opts = stan.subscriptionOptions()

      for (var option in req.options) {
        if (req.options[option] !== undefined) {
          opts[option](req.options[option])
        }
      }

      this.log.debug(opts, 'Subscription options')

      const sub = stan.subscribe(req.subject, req.queue, opts)
      subList[req.subject] = sub

      // signal that subscription was created
      reply(null, { created: true, subject: req.subject, opts })

      sub.on('message', (msg) => {
        const result = SafeParse(msg.getData())
        const inboxChannel = topic + '.' + req.subject
        if (result.error) {
          const error = new ParsingError(`Message could not be parsed as JSON. Subscription "${inboxChannel}"`)<commit_msg>use subject<commit_after>'use strict'

const Hp = require('hemera-plugin')
const SafeStringify = require('fast-safe-stringify')
const SafeParse = require('json-parse-safe')
const Nats = require('node-nats-streaming')

exports.plugin = Hp(function hemeraNatsStreaming (options, next) {
  const hemera = this
  const topic = 'nats-streaming'
  const Joi = hemera.exposition['hemera-joi'].joi
  const DuplicateSubscriberError = hemera.createError('DuplicateSubscriber')
  const ParsingError = hemera.createError('ParsingError')
  const NotAvailableError = hemera.createError('NotAvailable')
  const stan = Nats.connect(options.clusterId, options.clientId, options.opts)
  const subList = {}

  hemera.ext('onClose', (next) => {
    hemera.log.debug('Stan closing ...')
    stan.close()
    next()
  })

  stan.on('error', (err) => {
    hemera.log.error(err)
    hemera.fatal()
  })

  stan.on('connect', function () {
    /**
     * Publish a message over NATS-Streaming server
     */
    hemera.add({
      topic,
      cmd: 'publish',
      subject: Joi.string().required(),
      data: Joi.alternatives().try(Joi.object(), Joi.array())
    }, function (req, reply) {
      function handler (err, guid) {
        if (err) {
          reply(err)
        } else {
          reply(null, guid)
        }
      }

      stan.publish(req.subject, SafeStringify(req.data), handler)
    })

    /**
     * Create a subscription on the NATS-Streaming server
     * All options are available expect manual acknowledgement because we use it as a proxy
     */
    hemera.add({
      topic,
      cmd: 'subscribe',
      subject: Joi.string().required(),
      queue: Joi.string().optional(),
      options: Joi.object().keys({
        setStartWithLastReceived: Joi.boolean(), // Subscribe starting with the most recently published value
        setDeliverAllAvailable: Joi.boolean(), // Receive all stored values in order
        setStartAtSequence: Joi.number().integer(), // Receive all messages starting at a specific sequence number
        setStartTime: Joi.date().iso(), // Subscribe starting at a specific time
        setStartAtTimeDelta: Joi.number().integer(),
        setDurableName: Joi.string(), // Create a durable subscription
        setMaxInFlight: Joi.number().integer(), // the maximum number of outstanding acknowledgements
        setManualAckMode: Joi.boolean().default(true),
        setAckWait: Joi.number().integer() // if an acknowledgement is not received within the configured timeout interval, NATS Streaming will attempt redelivery of the message (default 30 seconds)
      }).default()
    }, function (req, reply) {
      // avoid multiple subscribers for the same subject
      if (subList[req.subject]) {
        reply(new DuplicateSubscriberError(`Subscription for subject "${req.subject}" is already active`))
        return
      }

      const opts = stan.subscriptionOptions()

      for (var option in req.options) {
        if (req.options[option] !== undefined) {
          opts[option](req.options[option])
        }
      }

      this.log.debug(opts, 'Subscription options')

      const sub = stan.subscribe(req.subject, req.queue, opts)
      subList[req.subject] = sub

      // signal that subscription was created
      reply(null, { created: true, subject: req.subject, opts })

      sub.on('message', (msg) => {
        const result = SafeParse(msg.getData())
        const inboxChannel = topic + '.' + req.subject
        if (result.error) {
          const error = new ParsingError(`Message could not be parsed as JSON. Subject "${req.subject}"`)
{'fraction_non_alphanumeric': 0.12531293463143253, 'fraction_numerical': 0.0005563282336578582, 'mean_word_length': 2.9576224545954872, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 9, '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': '3418918', 'n_tokens_mistral': 2081, 'n_tokens_neox': 2038, 'n_words': 550}