<commit_before>package reactivemongo.core.nodeset

import java.net.InetSocketAddress
import java.util.concurrent.{Executor, Executors}

import akka.actor.Actor.Receive
import akka.io.Tcp.{Register, Connected, Connect}
import akka.io.{Tcp, IO}
import akka.pattern.ask
import akka.actor._
import akka.util.Timeout
import reactivemongo.core.actors.{AwaitingResponse, RequestMakerExpectingResponse}
import scala.collection.immutable.HashMap
import scala.concurrent.ExecutionContext.Implicits.global
import org.jboss.netty.buffer.HeapChannelBufferFactory
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory
import org.jboss.netty.channel.{Channel, ChannelPipeline, Channels}
import reactivemongo.api.{MongoConnectionOptions, ReadPreference}
import reactivemongo.bson._
import reactivemongo.core._
import reactivemongo.core.{SocketHandler, ConnectionManager}
import reactivemongo.core.protocol.{Request, _}
import reactivemongo.utils.LazyLogger

import scala.collection.generic.CanBuildFrom

package object utils {
  def updateFirst[A, M[T] <: Iterable[T]](coll: M[A])(ƒ: A => Option[A])(implicit cbf: CanBuildFrom[M[_], A, M[A]]): M[A] = {
    val builder = cbf.apply
    def run(iterator: Iterator[A]): Unit = {
      while (iterator.hasNext) {
        val e = iterator.next
        val updated = ƒ(e)
        if (updated.isDefined) {
          builder += updated.get
          builder ++= iterator
        } else builder += e
      }
    }
    builder.result
  }

  def update[A, M[T] <: Iterable[T]](coll: M[A])(f: PartialFunction[A, A])(implicit cbf: CanBuildFrom[M[_], A, M[A]]): (M[A], Boolean) = {
    val builder = cbf.apply
    val (head, tail) = coll.span(!f.isDefinedAt(_))
    builder ++= head
    if (!tail.isEmpty) {
      builder += f(tail.head)
      builder ++= tail.drop(1)
    }
    builder.result -> !tail.isEmpty
  }
}

class NodeSet(
     val name: Option[String],
     val authenticates: Set[Authenticate]) extends Actor with ActorLogging {
  import NodeSet._
  import RandomPick._

  var nodes: Vector[ActorRef] = Vector.empty[ActorRef]
  var version: Option[Long] = None
  var channelsMapping = Map.empty[Int, ActorRef]
  var mongosConnections = Map.empty[Int, ActorRef]
  var statusConnections = Map.empty[NodeStatus, List[(Int, ActorRef)]]

  def primary = statusConnections.get(NodeStatus.Primary).map(_.getRandom()._2)
  def secondary = statusConnections.get(NodeStatus.Secondary).map(_.getRandom()._2)


  //val mongos: Option[Node] = nodes.find(_.isMongos)
  /*
  val primary: Option[Node] = nodes.find(_.status == NodeStatus.Primary)
  val secondaries = new RoundRobiner(nodes.filter(_.status == NodeStatus.Secondary))
  val queryable = secondaries.subject ++ primary
  val nearestGroup = new RoundRobiner(queryable.sortWith { _.pingInfo.ping < _.pingInfo.ping })
  val nearest = nearestGroup.subject.headOption
  val protocolMetadata = primary.orElse(secondaries.subject.headOption).map(_.protocolMetadata).getOrElse(ProtocolMetadata.Default)
  */

  def primary(authenticated: Authenticated): Option[Node] =
    primary.filter(_.authenticated.exists(_ == authenticated))

  def isReachable = !primary.isEmpty || !secondaries.subject.isEmpty
//
//  def updateNodeByChannelId(id: Int)(ƒ: Node => Node) =
//    updateByChannelId(id)(identity)(ƒ)
//
//  def updateConnectionByChannelId(id: Int)(ƒ: Connection => Connection) =
//    updateByChannelId(id)(ƒ)(identity)

//  def updateByChannelId(id: Int)(ƒc: Connection => Connection)(ƒn: Node => Node) = {
//    copy(nodes = nodes.map { node =>
//      val (connections, updated) = utils.update(node.connections) {
//        case conn if (conn.chanel == id) => ƒc(conn)
//      }
//      if (updated) ƒn(node.copy(connections = connections))
//      else node
//    })
//  }

  def pickByChannelId(id: Int) = channelsMapping.get(id)

  def pickConnection(channel: Int) = channelsMapping.get(channel)

//  def pickForWrite: Option[(Node, Connection)] =
//    primary.view.map(node => node -> node.authenticatedConnections.subject.headOption).collectFirst {
//      case (node, Some(connection)) => node -> connection
//    }

  private val pickConnectionAndFlatten: Option[Node] => Option[(Node, Connection)] = _.map(node => node -> node.authenticatedConnections.pick).collect {
    case (node, Some(connection)) => (node, connection)
  }
//
//  private def pickFromGroupWithFilter(roundRobiner: RoundRobiner[Node, Vector], filter: Option[BSONDocument => Boolean], fallback: => Option[Node]) =
//    filter.fold(fallback)(f =>
//      roundRobiner.pickWithFilter(_.tags.fold(false)(f)))

  def pick(preference: ReadPreference): Option[ActorRef] = if(mongosConnections.isEmpty){
    preference match {
      case ReadPreference.Primary => primary
    }
  } else {
    mongosConnections.getRandom._2
  }

  // http://docs.mongodb.org/manual/reference/read-preference/
//  def pick(preference: ReadPreference): Option[(Node, Connection)] = {
//    if (mongos.isDefined) {
//      pickConnectionAndFlatten(mongos)
//    } else preference match {
//      case ReadPreference.Primary                    => pickConnectionAndFlatten(primary)
//      case ReadPreference.PrimaryPreferred(filter)   => pickConnectionAndFlatten(primary.orElse(pickFromGroupWithFilter(secondaries, filter, secondaries.pick)))
//      case ReadPreference.Secondary(filter)          => pickConnectionAndFlatten(pickFromGroupWithFilter(secondaries, filter, secondaries.pick))
//      case ReadPreference.SecondaryPreferred(filter) => pickConnectionAndFlatten(pickFromGroupWithFilter(secondaries, filter, secondaries.pick).orElse(primary))
//      case ReadPreference.Nearest(filter)            => pickConnectionAndFlatten(pickFromGroupWithFilter(nearestGroup, filter, nearest))
//    }
//  }
//
//  def createNeededChannels(receiver: => ActorRef, connectionManager: => ActorRef, upTo: Int): NodeSet =
//    copy(nodes = nodes.foldLeft(Vector.empty[Node]) { (nodes, node) =>
//      nodes :+ node.createNeededChannels(receiver, connectionManager,  upTo)
//    })

  override def receive: Receive = {
    case AddNode(address) => {
      log.info("Adding Node to NodeSet with address {}", address)
      val node = context.actorOf(Props(classOf[Node]))
        nodes = node +: nodes
    }

  }
}

object NodeSet {
  case class AddNode(address: InetSocketAddress)
  case class IsMaster(id: Int)
  case class OnDisconnect(chanel: Int)
  object PrimaryUnavaliable
}


case class ProtocolMetadata(
  minWireVersion: MongoWireVersion,
  maxWireVersion: MongoWireVersion,
  maxMessageSizeBytes: Int,<commit_msg>router<commit_after>package reactivemongo.core.nodeset

import java.net.InetSocketAddress
import java.util.concurrent.{Executor, Executors}

import akka.actor.Actor.Receive
import akka.io.Tcp.{Register, Connected, Connect}
import akka.io.{Tcp, IO}
import akka.pattern.ask
import akka.actor._
import akka.routing.{RoundRobinRoutingLogic, Router}
import akka.util.Timeout
import reactivemongo.core.actors.{AwaitingResponse, RequestMakerExpectingResponse}
import scala.collection.immutable.HashMap
import scala.concurrent.ExecutionContext.Implicits.global
import org.jboss.netty.buffer.HeapChannelBufferFactory
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory
import org.jboss.netty.channel.{Channel, ChannelPipeline, Channels}
import reactivemongo.api.{MongoConnectionOptions, ReadPreference}
import reactivemongo.bson._
import reactivemongo.core._
import reactivemongo.core.{SocketHandler, ConnectionManager}
import reactivemongo.core.protocol.{Request, _}
import reactivemongo.utils.LazyLogger

import scala.collection.generic.CanBuildFrom

package object utils {
  def updateFirst[A, M[T] <: Iterable[T]](coll: M[A])(ƒ: A => Option[A])(implicit cbf: CanBuildFrom[M[_], A, M[A]]): M[A] = {
    val builder = cbf.apply
    def run(iterator: Iterator[A]): Unit = {
      while (iterator.hasNext) {
        val e = iterator.next
        val updated = ƒ(e)
        if (updated.isDefined) {
          builder += updated.get
          builder ++= iterator
        } else builder += e
      }
    }
    builder.result
  }

  def update[A, M[T] <: Iterable[T]](coll: M[A])(f: PartialFunction[A, A])(implicit cbf: CanBuildFrom[M[_], A, M[A]]): (M[A], Boolean) = {
    val builder = cbf.apply
    val (head, tail) = coll.span(!f.isDefinedAt(_))
    builder ++= head
    if (!tail.isEmpty) {
      builder += f(tail.head)
      builder ++= tail.drop(1)
    }
    builder.result -> !tail.isEmpty
  }
}

class NodeSet(
     val name: Option[String],
     val authenticates: Set[Authenticate]) {
  import NodeSet._
  import RandomPick._

  var nodes: Vector[ActorRef] = Vector.empty[ActorRef]
  var version: Option[Long] = None

  var primaries : Router = null
  var secondaries : Router = null
  var mongos : Router = null
  var primaryPrefered : Router = null


  //val mongos: Option[Node] = nodes.find(_.isMongos)
  /*
  val primary: Option[Node] = nodes.find(_.status == NodeStatus.Primary)
  val secondaries = new RoundRobiner(nodes.filter(_.status == NodeStatus.Secondary))
  val queryable = secondaries.subject ++ primary
  val nearestGroup = new RoundRobiner(queryable.sortWith { _.pingInfo.ping < _.pingInfo.ping })
  val nearest = nearestGroup.subject.headOption
  val protocolMetadata = primary.orElse(secondaries.subject.headOption).map(_.protocolMetadata).getOrElse(ProtocolMetadata.Default)
  */
//
//  def primary(authenticated: Authenticated): Option[Node] =
//    primary.filter(_.authenticated.exists(_ == authenticated))
//
//  def isReachable = !primary.isEmpty || !secondaries.subject.isEmpty
//
//  def updateNodeByChannelId(id: Int)(ƒ: Node => Node) =
//    updateByChannelId(id)(identity)(ƒ)
//
//  def updateConnectionByChannelId(id: Int)(ƒ: Connection => Connection) =
//    updateByChannelId(id)(ƒ)(identity)

//  def updateByChannelId(id: Int)(ƒc: Connection => Connection)(ƒn: Node => Node) = {
//    copy(nodes = nodes.map { node =>
//      val (connections, updated) = utils.update(node.connections) {
//        case conn if (conn.chanel == id) => ƒc(conn)
//      }
//      if (updated) ƒn(node.copy(connections = connections))
//      else node
//    })
//  }

  def pickByChannelId(id: Int) = channelsMapping.get(id)

  def pickConnection(channel: Int) = channelsMapping.get(channel)

//  def pickForWrite: Option[(Node, Connection)] =
//    primary.view.map(node => node -> node.authenticatedConnections.subject.headOption).collectFirst {
//      case (node, Some(connection)) => node -> connection
//    }

  private val pickConnectionAndFlatten: Option[Node] => Option[(Node, Connection)] = _.map(node => node -> node.authenticatedConnections.pick).collect {
    case (node, Some(connection)) => (node, connection)
  }
//
//  private def pickFromGroupWithFilter(roundRobiner: RoundRobiner[Node, Vector], filter: Option[BSONDocument => Boolean], fallback: => Option[Node]) =
//    filter.fold(fallback)(f =>
//      roundRobiner.pickWithFilter(_.tags.fold(false)(f)))

  def pick(preference: ReadPreference): Option[ActorRef] = if(mongosConnections.isEmpty){
    preference match {
      case ReadPreference.Primary => primary
    }
  } else {
    mongosConnections.getRandom._2
  }

  // http://docs.mongodb.org/manual/reference/read-preference/
//  def pick(preference: ReadPreference): Option[(Node, Connection)] = {
//    if (mongos.isDefined) {
//      pickConnectionAndFlatten(mongos)
//    } else preference match {
//      case ReadPreference.Primary                    => pickConnectionAndFlatten(primary)
//      case ReadPreference.PrimaryPreferred(filter)   => pickConnectionAndFlatten(primary.orElse(pickFromGroupWithFilter(secondaries, filter, secondaries.pick)))
//      case ReadPreference.Secondary(filter)          => pickConnectionAndFlatten(pickFromGroupWithFilter(secondaries, filter, secondaries.pick))
//      case ReadPreference.SecondaryPreferred(filter) => pickConnectionAndFlatten(pickFromGroupWithFilter(secondaries, filter, secondaries.pick).orElse(primary))
//      case ReadPreference.Nearest(filter)            => pickConnectionAndFlatten(pickFromGroupWithFilter(nearestGroup, filter, nearest))
//    }
//  }
//
//  def createNeededChannels(receiver: => ActorRef, connectionManager: => ActorRef, upTo: Int): NodeSet =
//    copy(nodes = nodes.foldLeft(Vector.empty[Node]) { (nodes, node) =>
//      nodes :+ node.createNeededChannels(receiver, connectionManager,  upTo)
//    })

//  override def receive: Receive = {
//    case AddNode(address) => {
//      log.info("Adding Node to NodeSet with address {}", address)
//      val node = context.actorOf(Props(classOf[Node]))
//        nodes = node +: nodes
//    }
//
//  }

  def createChannel(seed: )
}

object NodeSet {
  case class AddNode(address: InetSocketAddress)
  case class IsMaster(id: Int)
  case class OnDisconnect(chanel: Int)
  object PrimaryUnavaliable
}


case class ProtocolMetadata(
  minWireVersion: MongoWireVersion,
  maxWireVersion: MongoWireVersion,
  maxMessageSizeBytes: Int,
{'fraction_non_alphanumeric': 0.13343593689880723, 'fraction_numerical': 0.000461716044632551, 'mean_word_length': 4.046990291262136, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 61, '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': '4925776', 'n_tokens_mistral': 4102, 'n_tokens_neox': 4019, 'n_words': 791}