<commit_before>package com.twitter.finagle.loadbalancer

import com.twitter.finagle._
import com.twitter.finagle.addr.WeightedAddress
import com.twitter.finagle.loadbalancer.distributor.AddrLifecycle._
import com.twitter.finagle.loadbalancer.distributor.BalancerEndpoints
import com.twitter.finagle.loadbalancer.distributor.CachedBalancer
import com.twitter.finagle.loadbalancer.distributor.WeightClass
import com.twitter.finagle.service.DelayedFactory
import com.twitter.finagle.service.FailingFactory
import com.twitter.finagle.service.ServiceFactoryRef
import com.twitter.finagle.stats.Counter
import com.twitter.finagle.stats.NullStatsReceiver
import com.twitter.finagle.stats.StatsReceiver
import com.twitter.finagle.stats.Verbosity
import com.twitter.finagle.util.Drv
import com.twitter.finagle.util.Rng
import com.twitter.logging.Logger
import com.twitter.util._
import scala.util.control.NonFatal

private[finagle] object TrafficDistributor {
  val log = Logger.get(classOf[TrafficDistributor[_, _]])

  /**
   * Creates a `newEndpoint` for each distinct [[Address]] in the `addrs`
   * stream. Calls to `newEndpoint` are cached based on the input address. The cache is
   * privy to [[Address]] with weight metadata and unwraps them. Weights
   * are extracted and coupled with the their respective result from `newEndpoint`. If
   * the [[Address]] does not have a weight, a default weight of 1.0 is used.
   *
   * For every distinct [[Address]] observed in `addrs`, it creates a `newEndpoint`.
   * Each resulting `newEndpoint` is paired with a weight extracted from the [[Address]].
   * Calls to `newEndpoint` are assumed to be expensive, so they are cached by input address.
   *
   * @param eagerEviction When set to false, an [[Address]] cache entry is only removed
   * when its associated [[ServiceFactory]] has a status that is not Status.Open. This allows
   * for stale cache entries across updates that are only evicted when a [[ServiceFactory]]
   * is no longer eligible to receive traffic (as indicated by its `status` field).
   */
  private[finagle] def weightEndpoints[Req, Rep](
    addrs: Activity[Set[Address]],
    newEndpoint: Address => ServiceFactory[Req, Rep],
    eagerEviction: Boolean,
  ): Event[Activity.State[Set[EndpointFactory[Req, Rep]]]] = {

    // The EndpointFactoryProxy enables us to replace an EndpointFactory's
    // Address without having to close the EndpointFactory and build a new one.
    final case class EndpointFactoryProxy(
      factory: EndpointFactory[Req, Rep],
      address: Address)
        extends EndpointFactory[Req, Rep] {

      override def status: Status = factory.status
      override def remake(): Unit = factory.remake()
      override def close(deadline: Time): Future[Unit] = factory.close(deadline)
      override def apply(conn: ClientConnection): Future[Service[Req, Rep]] = factory(conn)
    }

    val init = Map.empty[Address, EndpointFactoryProxy]
    safelyScanLeft(init, addrs.run.changes.dedup) {
      case (active, addrs) =>
        // Note, if an update contains multiple `Address` instances that are the same
        // with duplicate `weight` metadata, only one of the instances and its associated
        // factory is cached. Last write wins.<commit_msg>Finagle-core: Added toString() method to TrafficDistributor$EndpointFactoryProxy<commit_after>package com.twitter.finagle.loadbalancer

import com.twitter.finagle._
import com.twitter.finagle.addr.WeightedAddress
import com.twitter.finagle.loadbalancer.distributor.AddrLifecycle._
import com.twitter.finagle.loadbalancer.distributor.BalancerEndpoints
import com.twitter.finagle.loadbalancer.distributor.CachedBalancer
import com.twitter.finagle.loadbalancer.distributor.WeightClass
import com.twitter.finagle.service.DelayedFactory
import com.twitter.finagle.service.FailingFactory
import com.twitter.finagle.service.ServiceFactoryRef
import com.twitter.finagle.stats.Counter
import com.twitter.finagle.stats.NullStatsReceiver
import com.twitter.finagle.stats.StatsReceiver
import com.twitter.finagle.stats.Verbosity
import com.twitter.finagle.util.Drv
import com.twitter.finagle.util.Rng
import com.twitter.logging.Logger
import com.twitter.util._
import scala.util.control.NonFatal

private[finagle] object TrafficDistributor {
  val log = Logger.get(classOf[TrafficDistributor[_, _]])

  /**
   * Creates a `newEndpoint` for each distinct [[Address]] in the `addrs`
   * stream. Calls to `newEndpoint` are cached based on the input address. The cache is
   * privy to [[Address]] with weight metadata and unwraps them. Weights
   * are extracted and coupled with the their respective result from `newEndpoint`. If
   * the [[Address]] does not have a weight, a default weight of 1.0 is used.
   *
   * For every distinct [[Address]] observed in `addrs`, it creates a `newEndpoint`.
   * Each resulting `newEndpoint` is paired with a weight extracted from the [[Address]].
   * Calls to `newEndpoint` are assumed to be expensive, so they are cached by input address.
   *
   * @param eagerEviction When set to false, an [[Address]] cache entry is only removed
   * when its associated [[ServiceFactory]] has a status that is not Status.Open. This allows
   * for stale cache entries across updates that are only evicted when a [[ServiceFactory]]
   * is no longer eligible to receive traffic (as indicated by its `status` field).
   */
  private[finagle] def weightEndpoints[Req, Rep](
    addrs: Activity[Set[Address]],
    newEndpoint: Address => ServiceFactory[Req, Rep],
    eagerEviction: Boolean,
  ): Event[Activity.State[Set[EndpointFactory[Req, Rep]]]] = {

    // The EndpointFactoryProxy enables us to replace an EndpointFactory's
    // Address without having to close the EndpointFactory and build a new one.
    final case class EndpointFactoryProxy(
      factory: EndpointFactory[Req, Rep],
      address: Address)
        extends EndpointFactory[Req, Rep] {

      override def status: Status = factory.status
      override def remake(): Unit = factory.remake()
      override def close(deadline: Time): Future[Unit] = factory.close(deadline)
      override def apply(conn: ClientConnection): Future[Service[Req, Rep]] = factory(conn)
      override def toString: String = address.toString
    }

    val init = Map.empty[Address, EndpointFactoryProxy]
    safelyScanLeft(init, addrs.run.changes.dedup) {
      case (active, addrs) =>
        // Note, if an update contains multiple `Address` instances that are the same
        // with duplicate `weight` metadata, only one of the instances and its associated
        // factory is cached. Last write wins.
{'fraction_non_alphanumeric': 0.09047044632086852, 'fraction_numerical': 0.0006031363088057901, 'mean_word_length': 4.272655007949125, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, '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': '5715118', 'n_tokens_mistral': 1826, 'n_tokens_neox': 1800, 'n_words': 613}