<commit_before>package org.bitcoins.zmq

import java.net.{ InetSocketAddress, Socket }

import org.bitcoins.core.crypto.{ DoubleSha256Digest, HashDigest }
import org.bitcoins.core.protocol.blockchain.Block
import org.bitcoins.core.protocol.transaction.Transaction
import org.bitcoins.core.util.BitcoinSLogger
import org.zeromq.{ ZMQ, ZMsg }

import scala.concurrent.{ ExecutionContext, Future }

/**
 * This class is designed to consume a zmq stream from a cryptocurrency's daemon.
 * An example of this is  bitcoind. For information on how to setup your coin's  conf
 * file to be able to consume a zmq stream please see
 * [[https://github.com/bitcoin/bitcoin/blob/master/doc/zmq.md#usage]]
 * [[http://zguide.zeromq.org/java:psenvsub]]
 * @param socket
 * @param hashTxListener
 * @param hashBlockListener
 * @param rawTxListener
 * @param rawBlockListener
 */
class ZMQSubscriber(
  socket: InetSocketAddress,
  hashTxListener: Option[Seq[Byte] => Future[Unit]],
  hashBlockListener: Option[Seq[Byte] => Future[Unit]],
  rawTxListener: Option[Seq[Byte] => Future[Unit]],
  rawBlockListener: Option[Seq[Byte] => Future[Unit]]) {
  private val logger = BitcoinSLogger.logger

  private var run = false
  private val context = ZMQ.context(1)

  private val subscriber = context.socket(ZMQ.SUB)
  private val uri = socket.getHostString + ":" + socket.getPort

  def start()(implicit ec: ExecutionContext): Future[Unit] = Future {
    logger.info("starting zmq")
    subscriber.connect(uri)
    logger.info("Connection to zmq client successful")
    //subscribe to the appropriate feed
    hashTxListener.map { _ =>
      subscriber.subscribe(HashTx.topic.getBytes(ZMQ.CHARSET))
      logger.debug("subscribed to the transaction hashes from zmq")
    }

    rawTxListener.map { _ =>
      subscriber.subscribe(RawTx.topic.getBytes(ZMQ.CHARSET))
      logger.debug("subscribed to raw transactions from zmq")
    }

    hashBlockListener.map { _ =>
      subscriber.subscribe(HashBlock.topic.getBytes(ZMQ.CHARSET))
      logger.debug("subscribed to the hashblock stream from zmq")
    }

    rawBlockListener.map { _ =>
      subscriber.subscribe(RawBlock.topic.getBytes(ZMQ.CHARSET))
      logger.debug("subscribed to raw block stream from zmq")
    }

    run = true
    while (run) {
      val zmsg = ZMsg.recvMsg(subscriber)
      val notificationTypeStr = zmsg.pop().getString(ZMQ.CHARSET)
      val body = zmsg.pop().getData
      val processedMsg = processMsg(notificationTypeStr, body)
      processedMsg.onFailure {
        case err =>
          logger.error(err.getMessage)
      }
    }
  }

  /**
   * Stops running the zmq subscriber and cleans up after zmq
   * http://zguide.zeromq.org/java:psenvsub
   */
  def stop: Unit = {
    //i think this could technically not work, because currently we are blocking
    //on Zmsg.recvMsg in our while loop. If we don't get another message we won't
    //be able toe evaluate the while loop again. Moving forward with this for now.
    run = false
    subscriber.close()
    context.term()
  }

  /** Processes a message that we received the from the cryptocurrency daemon and then
    * applies the appropriate listener to that message.
    */
  private def processMsg(topic: String, body: Seq[Byte])(implicit ec: ExecutionContext): Future[Unit] = {
    val notification = ZMQNotification.fromString(topic)
    val res: Option[Future[Unit]] = notification.flatMap {
      case HashTx =>
        hashTxListener.map { f =>
          f(body)
        }
      case RawTx =>
        rawTxListener.map { f =>
          f(body)
        }
      case HashBlock =>
        hashBlockListener.map { f =>
          f(body)
        }
      case RawBlock =>
        rawBlockListener.map { f =>
          f(body)
        }
    }

    res match {
      case Some(f) => f
      case None => Future.successful(Unit)
    }
  }
}<commit_msg>Adding more logging around zmq connection<commit_after>package org.bitcoins.zmq

import java.net.{ InetSocketAddress, Socket }

import org.bitcoins.core.crypto.{ DoubleSha256Digest, HashDigest }
import org.bitcoins.core.protocol.blockchain.Block
import org.bitcoins.core.protocol.transaction.Transaction
import org.bitcoins.core.util.BitcoinSLogger
import org.zeromq.{ ZMQ, ZMsg }

import scala.concurrent.{ ExecutionContext, Future }

/**
 * This class is designed to consume a zmq stream from a cryptocurrency's daemon.
 * An example of this is  bitcoind. For information on how to setup your coin's  conf
 * file to be able to consume a zmq stream please see
 * [[https://github.com/bitcoin/bitcoin/blob/master/doc/zmq.md#usage]]
 * [[http://zguide.zeromq.org/java:psenvsub]]
 * @param socket
 * @param hashTxListener
 * @param hashBlockListener
 * @param rawTxListener
 * @param rawBlockListener
 */
class ZMQSubscriber(
  socket: InetSocketAddress,
  hashTxListener: Option[Seq[Byte] => Future[Unit]],
  hashBlockListener: Option[Seq[Byte] => Future[Unit]],
  rawTxListener: Option[Seq[Byte] => Future[Unit]],
  rawBlockListener: Option[Seq[Byte] => Future[Unit]]) {
  private val logger = BitcoinSLogger.logger

  private var run = false
  private val context = ZMQ.context(1)

  private val subscriber = context.socket(ZMQ.SUB)
  private val uri = socket.getHostString + ":" + socket.getPort

  def start()(implicit ec: ExecutionContext): Future[Unit] = Future {
    logger.info("starting zmq")
    val connected = subscriber.connect(uri)
    if (connected) {
      logger.info(s"Connection to zmq client successful to $uri")
      //subscribe to the appropriate feed
      hashTxListener.map { _ =>
        subscriber.subscribe(HashTx.topic.getBytes(ZMQ.CHARSET))
        logger.debug("subscribed to the transaction hashes from zmq")
      }

      rawTxListener.map { _ =>
        subscriber.subscribe(RawTx.topic.getBytes(ZMQ.CHARSET))
        logger.debug("subscribed to raw transactions from zmq")
      }

      hashBlockListener.map { _ =>
        subscriber.subscribe(HashBlock.topic.getBytes(ZMQ.CHARSET))
        logger.debug("subscribed to the hashblock stream from zmq")
      }

      rawBlockListener.map { _ =>
        subscriber.subscribe(RawBlock.topic.getBytes(ZMQ.CHARSET))
        logger.debug("subscribed to raw block stream from zmq")
      }

      run = true
      while (run) {
        val zmsg = ZMsg.recvMsg(subscriber)
        val notificationTypeStr = zmsg.pop().getString(ZMQ.CHARSET)
        val body = zmsg.pop().getData
        val processedMsg = processMsg(notificationTypeStr, body)
        processedMsg.onFailure {
          case err =>
            logger.error(err.getMessage)
        }
      }
    } else {
      logger.info(s"Failed to connect zmq client to $uri")
      Future.failed(throw new IllegalArgumentException(s"Failed to connect zmq client to $uri"))
    }

  }

  /**
   * Stops running the zmq subscriber and cleans up after zmq
   * http://zguide.zeromq.org/java:psenvsub
   */
  def stop: Unit = {
    //i think this could technically not work, because currently we are blocking
    //on Zmsg.recvMsg in our while loop. If we don't get another message we won't
    //be able toe evaluate the while loop again. Moving forward with this for now.
    run = false
    subscriber.close()
    context.term()
  }

  /**
   * Processes a message that we received the from the cryptocurrency daemon and then
   * applies the appropriate listener to that message.
   */
  private def processMsg(topic: String, body: Seq[Byte])(implicit ec: ExecutionContext): Future[Unit] = {
    val notification = ZMQNotification.fromString(topic)
    val res: Option[Future[Unit]] = notification.flatMap {
      case HashTx =>
        hashTxListener.map { f =>
          f(body)
        }
      case RawTx =>
        rawTxListener.map { f =>
          f(body)
        }
      case HashBlock =>
        hashBlockListener.map { f =>
          f(body)
        }
      case RawBlock =>
        rawBlockListener.map { f =>
          f(body)
        }
    }

    res match {
      case Some(f) => f
      case None => Future.successful(Unit)
    }
  }
}
{'fraction_non_alphanumeric': 0.10487774605932729, 'fraction_numerical': 0.0009929254064788382, 'mean_word_length': 2.9910846953937593, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 41, 'https://': 2, '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': '5932917', 'n_tokens_mistral': 2460, 'n_tokens_neox': 2379, 'n_words': 646}