<commit_before>/**
 * Copyright 2015, deepsense.io
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.deepsense.workflowexecutor.rabbitmq

import java.util

import akka.actor.{ActorRef, ActorSystem}
import com.rabbitmq.client.Channel
import com.thenewmotion.akka.rabbitmq.{ChannelActor, CreateChannel, RichConnectionActor}

import io.deepsense.workflowexecutor.communication.mq.MQCommunication
import io.deepsense.workflowexecutor.communication.mq.serialization.{MessageMQDeserializer, MessageMQSerializer}

case class MQCommunicationFactory(
  system: ActorSystem,
  connection: ActorRef,
  mqMessageSerializer: MessageMQSerializer,
  mqMessageDeserializer: MessageMQDeserializer) {

  val exchangeType = "topic"

  def registerSubscriber(topic: String, subscriber: ActorRef): Unit = {
    val subscriberName = MQCommunication.subscriberName(topic)
    connection ! CreateChannel(
      ChannelActor.props(setupSubscriber(topic, subscriber)),
      Some(subscriberName))
  }

  def createPublisher(topic: String, publisherActorName: String): ActorRef = {
    val publisher = createMQPublisher(topic)
    system.actorOf(PublisherActor.props(topic, publisher), publisherActorName)
  }

  private def setupSubscriber(
    topic: String,
    subscriber: ActorRef)(
    channel: Channel, self: ActorRef): Unit = {
    val queueName: String = MQCommunication.queueName(topic)
    val queue = channel.queueDeclare(
      queueName,
      false,
      false,
      true,
      new util.HashMap[String, AnyRef]()).getQueue
    channel.queueBind(queue, MQCommunication.Exchange.seahorse, topic)
    val consumer = MQSubscriber(subscriber, mqMessageDeserializer, channel)
    channel.basicConsume(queue, true, consumer)
  }

  private def createMQPublisher(topic: String): MQPublisher = {
    val publisherName = MQCommunication.publisherName(topic)
    val channelActor: ActorRef =
      connection.createChannel(ChannelActor.props(), Some(publisherName))
    MQPublisher(MQCommunication.Exchange.seahorse, mqMessageSerializer, channelActor)<commit_msg>DS-3004 Create subscribers synchronously<commit_after>/**
 * Copyright 2015, deepsense.io
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.deepsense.workflowexecutor.rabbitmq

import java.util

import akka.actor.{ActorRef, ActorSystem}
import com.rabbitmq.client.Channel
import com.thenewmotion.akka.rabbitmq.{ChannelActor, CreateChannel, RichConnectionActor}

import io.deepsense.workflowexecutor.communication.mq.MQCommunication
import io.deepsense.workflowexecutor.communication.mq.serialization.{MessageMQDeserializer, MessageMQSerializer}

case class MQCommunicationFactory(
  system: ActorSystem,
  connection: ActorRef,
  mqMessageSerializer: MessageMQSerializer,
  mqMessageDeserializer: MessageMQDeserializer) {

  val exchangeType = "topic"

  def registerSubscriber(topic: String, subscriber: ActorRef): Unit = {
    val subscriberName = MQCommunication.subscriberName(topic)
    connection.createChannel(ChannelActor.props(setupSubscriber(topic, subscriber)),
      Some(subscriberName))
  }

  def createPublisher(topic: String, publisherActorName: String): ActorRef = {
    val publisher = createMQPublisher(topic)
    system.actorOf(PublisherActor.props(topic, publisher), publisherActorName)
  }

  private def setupSubscriber(
    topic: String,
    subscriber: ActorRef)(
    channel: Channel, self: ActorRef): Unit = {
    val queueName: String = MQCommunication.queueName(topic)
    val queue = channel.queueDeclare(
      queueName,
      false,
      false,
      true,
      new util.HashMap[String, AnyRef]()).getQueue
    channel.queueBind(queue, MQCommunication.Exchange.seahorse, topic)
    val consumer = MQSubscriber(subscriber, mqMessageDeserializer, channel)
    channel.basicConsume(queue, true, consumer)
  }

  private def createMQPublisher(topic: String): MQPublisher = {
    val publisherName = MQCommunication.publisherName(topic)
    val channelActor: ActorRef =
      connection.createChannel(ChannelActor.props(), Some(publisherName))
    MQPublisher(MQCommunication.Exchange.seahorse, mqMessageSerializer, channelActor)
{'fraction_non_alphanumeric': 0.08443322981366459, 'fraction_numerical': 0.0038819875776397515, 'mean_word_length': 4.835787089467724, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '1121912', 'n_tokens_mistral': 1526, 'n_tokens_neox': 1404, 'n_words': 386}