<commit_before>'use strict'

const amqplib = require('amqplib');

class PubSub {

  constructor(){
    this._rx = null;
    this._tx = null;
  }

  connect(url){
    return this.connectForRx(url).connectForTx(this._rx)
  }

  connectForRx(url){
    this._rx = (typeof url === 'string') ? amqplib.connect(url) : url
    return this
  }

  connectForTx(url){
    this._tx = (typeof url === 'string') ? amqplib.connect(url) : url
    return this
  }

  /**
   * @method publish
   * @description Publish a message to a specific queue
   * @author <NAME>
   */
  publish(queue, msg, opts){
    if(this._tx === null) throw new Error("No connection available for publishing")
    return this._tx.then(conn => conn.createChannel())
    .then(channel => [channel, channel.assertQueue(queue)])
    .then(results => results[0].sendToQueue(queue, new Buffer(JSON.stringify({data: msg})), opts)) // results[0] is the channel
  }

  subscribe(queue, opts, noAck){
    if(this._rx === null) throw new Error("No connection available for consuming")
    return this._rx.then(conn => conn.createChannel())
    .then(channel => Promise.all([channel, channel.assertQueue(queue)]))
    .then(results => {
      const channel = results[0]
      return new Promise((resolve, reject) => {
        channel.consume(queue, msg => {
          if(!noAck) channel.ack(msg)
          msg.content = JSON.parse(msg.content.toString()).data
          return (msg) ? resolve(msg) : reject(new Error("Message is null"))
        }, opts)
      })
    })
  }
}

module.exports = PubSub;
<commit_msg>PubSub Documentation added<commit_after>'use strict'

const amqplib = require('amqplib');

class PubSub {

  constructor(){
    this._rx = null;
    this._tx = null;
  }

  /**
   * @method connect
   * @description Connects to a AMQP server specified by a URL for both transmition and reception
   * @author <NAME>
   */
  connect(url){
    return this.connectForRx(url).connectForTx(this._rx)
  }

  /**
   * @method connectForRx
   * @description Connects to a AMQP server specified by a URL just for reception (listening messages)
   * @author <NAME>
   */
  connectForRx(url){
    this._rx = (typeof url === 'string') ? amqplib.connect(url) : url
    return this
  }

  /**
   * @method connectForTx
   * @description Connects to a AMQP server specified by a URL just for transmition (publish messages)
   * @author <NAME>
   */
  connectForTx(url){
    this._tx = (typeof url === 'string') ? amqplib.connect(url) : url
    return this
  }

  /**
   * @method publish
   * @description Publish a message to a specific queue
   * @author <NAME>
   */
  publish(queue, msg, opts){
    if(this._tx === null) throw new Error("No connection available for publishing")
    return this._tx.then(conn => conn.createChannel())
    .then(channel => [channel, channel.assertQueue(queue)])
    .then(results => results[0].sendToQueue(queue, new Buffer(JSON.stringify({data: msg})), opts)) // results[0] is the channel
  }

  /**
   * @method subscribe
   * @description Subscribe to a specific queue and attends the incoming messages in the Promise returned
   * @author <NAME>
   */
  subscribe(queue, opts, noAck){
    if(this._rx === null) throw new Error("No connection available for consuming")
    return this._rx.then(conn => conn.createChannel())
    .then(channel => Promise.all([channel, channel.assertQueue(queue)]))
    .then(results => {
      const channel = results[0]
      return new Promise((resolve, reject) => {
        channel.consume(queue, msg => {
          if(!noAck) channel.ack(msg)
          msg.content = JSON.parse(msg.content.toString()).data
          return (msg) ? resolve(msg) : reject(new Error("Message is null"))
        }, opts)
      })
    })
  }
}

module.exports = PubSub;

{'fraction_non_alphanumeric': 0.1426677273932644, 'fraction_numerical': 0.0015910898965791568, 'mean_word_length': 3.140504939626784, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 25, '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': '6028091', 'n_tokens_mistral': 1178, 'n_tokens_neox': 1161, 'n_words': 308}