<commit_before>var debug = require('debug')('micromono:rpc:socketmq')
var msgpack = require('msgpack-lite')
var socketmq = require('socketmq')

var REQ_NAME = 'micromono/api/rpc'

var SocketMQAdapter = module.exports = function() {}

SocketMQAdapter.prototype.type = 'socketmq'

SocketMQAdapter.prototype.getSocketMQ = function() {
  var smq = this.smq
  if (!smq) {
    smq = socketmq()
    smq.setMsgEncoder(msgpack, Buffer('m'))
    this.smq = smq
  }
  return smq
}

SocketMQAdapter.prototype.send = function(data) {
  var fn
  var args = data.args

  if ('function' === typeof args[args.length - 1]) {
    fn = args.pop()
    data.cid = true
  }

  this.adapter.smq.req(REQ_NAME, data, fn || function() {})
}

SocketMQAdapter.prototype.connect = function(provider) {
  var self = this
  var adapter = this.adapter
  var name = provider.name
  var endpoint = 'tcp://' + provider.host + ':' + provider.api.port
  debug('Connecting to service [%s] endpoint "%s"', name, endpoint)

  var smq = adapter.smq
  if (!smq) {
    // Create socketmq instance if not exists.
    smq = this.adapter.getSocketMQ()
    smq.on('disconnect', function(socket) {
      self.onProviderDisconnect(socket.provider)
      debug('Service [%s] provider "%s" closed', socket.provider.name, socket.provider.host)
    })
    smq.on('error', function(err, socket) {
      self.onProviderDisconnect(socket.provider)
      debug('Service [%s] provider "%s" error', socket.provider.name, socket.provider.host)
    })
  }

  var socket = smq.connect(endpoint)
  socket.provider = provider
}

SocketMQAdapter.prototype.startServer = function(port, host, callback) {<commit_msg>Call `onProviderDisconnect` only when there is a stream in error.<commit_after>var debug = require('debug')('micromono:rpc:socketmq')
var msgpack = require('msgpack-lite')
var socketmq = require('socketmq')

var REQ_NAME = 'micromono/api/rpc'

var SocketMQAdapter = module.exports = function() {}

SocketMQAdapter.prototype.type = 'socketmq'

SocketMQAdapter.prototype.getSocketMQ = function() {
  var smq = this.smq
  if (!smq) {
    smq = socketmq()
    smq.setMsgEncoder(msgpack, Buffer('m'))
    this.smq = smq
  }
  return smq
}

SocketMQAdapter.prototype.send = function(data) {
  var fn
  var args = data.args

  if ('function' === typeof args[args.length - 1]) {
    fn = args.pop()
    data.cid = true
  }

  this.adapter.smq.req(REQ_NAME, data, fn || function() {})
}

SocketMQAdapter.prototype.connect = function(provider) {
  var self = this
  var adapter = this.adapter
  var name = provider.name
  var endpoint = 'tcp://' + provider.host + ':' + provider.api.port
  debug('Connecting to service [%s] endpoint "%s"', name, endpoint)

  var smq = adapter.smq
  if (!smq) {
    // Create socketmq instance if not exists.
    smq = this.adapter.getSocketMQ()
    smq.on('disconnect', function(socket) {
      self.onProviderDisconnect(socket.provider)
      debug('Service [%s] provider "%s" disconnected', name, socket.provider.host)
    })
    smq.on('error', function(err) {
      var provider
      if (err.stream && err.stream.provider) {
        provider = err.stream.provider
        self.onProviderDisconnect(provider)
      }
      debug('Service [%s] provider "%s" error', name, provider ? provider.host : 'unknown host')
    })
  }

  var socket = smq.connect(endpoint)
  socket.provider = provider
}

SocketMQAdapter.prototype.startServer = function(port, host, callback) {
{'fraction_non_alphanumeric': 0.14231774615161197, 'fraction_numerical': 0.0005808887598024978, 'mean_word_length': 3.672998643147897, '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': '3784005', 'n_tokens_mistral': 1143, 'n_tokens_neox': 1113, 'n_words': 258}