Внедрение зависимостей с использованием монады Cats-effect Resource

Монада Cats-effect Resource предоставляет отличную монадическую абстракцию над паттерном try-with-resource. Например, она позволяет управлять жизненным циклом зависимостей, включая закрытие/финализацию ресурса, когда он больше не нужен (закрытие соединения с базой данных, освобождение кэша при завершении работы). В сочетании с компонуемостью монад это стало очень популярным подходом для управления зависимостями — до такой степени, что такие библиотеки Scala, как http4s, предоставляют свои зависимости обернутыми в монаду Resource.

Оффтоп: одним из самых главных преимуществ монады Resource является то, что она позволяет высвобождать ресурсы не в той манере, в которой они были инстанцированы, чего не делают даже такие монстры, как Google Guice.

Сейчас каждый разработчик приложений на Scala, создавая свое приложение, так или иначе сталкивается с управлением зависимостями. Существуют библиотеки, которые помогают управлять зависимостями, например macwire. Но обычно для выполнения внедрения зависимостей достаточно просто использовать монаду Resource, как, например, в следующем примере простенького http-сервиса:

import cats.effect.*
 
 // Репозиторий, два сервиса и api:
 class Repository(conn: ConnectionFactory) {}
 
 class ServiceA(repo: Repository) {}
 class ServiceB(repo: Repository) {}
 
 class HttpServerTask(serviceA: ServiceA, serviceB: ServiceB) {
   def run: IO[Unit] = ???
 }
 
 // Часть с внедрением зависимостей:
 object Dependencies {
   private val conn: Resource[IO, ConnectionFactory] = ???
   
   private val repo: Resource[IO, Repository] = for {
     conn <- this.conn
  } yield new Repository(conn)
    
   private val serviceA: Resource[IO, ServiceA] = for {
     repo <- this.repo
  } yield new ServiceA(repo)
   
   private val serviceB: Resource[IO, ServiceB] = for {
     repo <- this.repo
  } yield new ServiceB(repo)
   
   val server: Resource[IO, HttpServerTask] = for {
     serviceA <- this.serviceA
     serviceB <- this.serviceB
  } yield new HttpServerTask(serviceA, serviceB)
 }
 
 // Точка входа в приложение
 object Main extends IOApp.Simple {
   def run = Dependencies.server.use(_.run())
 }

Давайте разбираться, что мы здесь видим. Мы можем сказать, что наш объект Depencies — это контейнер внедрения зависимостей (DI-контейнер), а server — это выходная зависимость (exit dependency), то есть зависимость, которая будет использоваться вне контейнера. Все остальные зависимости являются внутренними зависимостями.

У этого кода есть одна серьезная проблема: из-за ленивой природы cats-effect, хотя и приятно думать, что conn — это val, на самом деле он будет выполняться столько раз, сколько на него ссылаются, в данном примере — дважды, чего мы не хотим, потому что тогда у нас будет два пула соединений с одной и той же базой данных в одном приложении. Стандартная схема внедрения зависимостей заключается в том, чтобы иметь только один экземпляр каждой зависимости, если не указано иное для конкретных случаев.

Поэтому обычно эта проблема решается в терминах функционального мышления, и первый подход, который приходит в голову, — это преобразовать всю эту разводку в функцию с большим for-comprehension, которая возвращает выходную зависимость, как здесь:

...
 
 // Здесь мы указываем наши выходные зависимости, оборачивая это все в case-класс для добавления новых выходных зависимостей в будущем
 case class Dependencies(server: HttpServerTask)
 
 object Dependencies {
   val conn: Resource[IO, ConnectionFactory] = ???
 
   def apply: Resource[IO, Dependencies] = for {
     conn <- this.conn
     repo <- new Repository(conn)
     serviceA <- new ServiceA(repo)
     serviceB <- new ServiceB(repo)
     server <- new HttpServerTask(serviceA, serviceB)
  } yield Dependencies(server)
   
 }

Из-за своей простоты этот подход широко используется для небольших приложений (пример). Но у него есть и некоторые недостатки:

• Наличие одного большого for-comprehension — не расширяемое решение, которое начинает выглядеть очень плохо, когда он разрастается за счет добавления новых зависимостей;

• Добавление дополнительных выходных зависимостей требует инстанцирования всех этих зависимостей, что касается и графа зависимостей. Предположим, у нас есть 2 API и 2 точки входа (которые будут развернуты как 2 отдельных сервиса), при таком подходе нам нужно либо написать 2 отдельных for-comprehension, либо инстанцировать их все вместе и использовать только один, что, мягко говоря, не оптимально;

• Если мы хотим, чтобы каждая зависимость могла быть выходной зависимостью, например, чтобы можно было использовать каждую зависимость вне DI-контейнера, что легко позволяют делать java-библиотеки вроде Google Guice, то такой подход нам не подходит.

Другой подход заключается в том, чтобы вернуться на шаг назад к отдельным зависимостям и просто попытаться инициализировать их, но отслеживать shutdown функции инициализированных зависимостей, чтобы иметь возможность изящно завершать их при остановке приложения.

Для этого нам понадобится вспомогательный класс Allocator, который будет отвечать за инициализацию зависимостей, отслеживание инициализированных зависимостей и их финализацию:

import cats.effect.*
import cats.effect.unsafe.IORuntime

class Allocator(implicit runtime: IORuntime) {
  // Ссылка, которая будет отслеживать финализаторы
  private val shutdown: Ref[IO, IO[Unit]] = Ref.unsafe(IO.unit)

  // Метод аллокации зависимостей
  def allocate[A](resource: Resource[IO, A]): A =
    resource.allocated.flatMap { case (a, release) =>
      // Закрываем этот ресурс, а после его закрытия - все предшествующие
      shutdown.update(release *> _).map(_ => a)
  }.unsafeRunSync()

  // Завершение зависимостей
  def shutdownAll: IO[Unit] = {
    shutdown.getAndSet(IO.unit).flatten
  }

}

Этот класс позволяет нам выполнять неидиоматическое разворачивание ресурсов и следить за финализаторами, чтобы завершать их в правильном порядке. Неидиоматичность будет исправлена позже, а сейчас посмотрите, как выглядит код с применением этого простого класса:

class Dependencies(val allocator: Allocator) {
  lazy val conn: ConnectionFactory = allocator.allocate {
     ???
  }
   
  lazy val repo = new Repository(conn)
    
  lazy val serviceA = new ServiceA(repo)
   
  lazy val serviceB = new ServiceB(repo)
   
  lazy val server = HttpServerTask(serviceA, serviceB)
}

Как видите, только одна зависимость нуждалась в Allocator, поскольку ей было что финализировать (ConnectionFactory). Код значительно упростился и позволяет нам добиться желаемого:

• Инстанцирование зависимостей по-прежнему ленивое, но теперь ленивость достигается не за счет ленивости Resource, а за счет механизма lazy val Scala.

• Теперь все зависимости могут быть выставлены как выходные зависимости. Каждая зависимость может быть свободно использована снаружи.

• И снова код прост и расширяем. Мы можем группировать похожие зависимости в отдельных классах Dependencies, используя для них всего один метод инстанцирования.

Теперь давайте разберемся с проблемой, которая заключается в том, что мы нарушили идиоматичность паттерна Resource внутри DI-контейнера. Если вы посмотрите на класс Allocator, то он сам по себе имеет метод shutdownAll, который позволяет нам теперь обернуть весь наш DI-контейнер в одну монаду Resource:

object Dependencies {
  // Безопасный метод для создания зависимостей:
  def apply(runtime: IORuntime): Resource[IO, Dependencies] =
    Resource.make {
      IO(unsafeCreate(runtime))
   } {
      _.allocator.shutdownAll
   }
 ​
  // Небезопасный метод, используйте его с осторожностью, так как здесь не выполняется завершение:
  def unsafeCreate(runtime: IORuntime): Dependencies =
    new Dependencies(new Allocator()(runtime))
}
​
class Dependencies(val allocator: Allocator) {
  ...
}
​
// Запускающий класс
object Main extends IOApp.Simple {
  // Теперь мы можем использовать любую зависимость, а не только server, так как все они могут быть выходными:
  private val dependencies = Dependencies(IORuntime.global)
  
  override def run = dependencies.use(_.server.run)
}

Вот и все! Этот метод является моим излюбленным способом реализации внедрения зависимостей с помощью Cats-Effect в Scala на сегодняшний день. Хоть и нарушая идиоматическую границу Resource внутри контейнера, он позволяет добиться повторного использования зависимостей простым и расширяемым способом, и мы по-прежнему возвращаемся к паттерну Resource для внешних пользователей, которые будут вызывать зависимости DI-контейнера.

Этот паттерн также позволяет легко разбивать зависимости на несколько контейнеров, например, так:

class AwsDependencies(val allocator: Allocator, config: Config) {
  val s3: AwsS3Client[IO] = allocator.allocate {
    Resource.fromAutoCloseable(IO.blocking {
      S3AsyncClient.builder()
        .region(config.region)
        .build()
    }).map(new AwsS3Client(_))
  }
}
 
class MainDependencies(val allocator: Allocator) {
  
  lazy val config = ???
  
  lazy val aws: AwsDependencies = new AwsDependencies(allocator, config.as[Config]("aws"))
  
  lazy val httpRoutes: Routes = new Routes(aws.s3)
  
}

Вот полный пример окончательного кода:

import cats.effect.*
import cats.effect.unsafe.IORuntime
import org.postgresql.core.ConnectionFactory

// repository, two services, and api:
class Repository(conn: ConnectionFactory) {}

class ServiceA(repo: Repository) {}

class ServiceB(repo: Repository) {}

class HttpServerTask(serviceA: ServiceA, serviceB: ServiceB) {
  def run: IO[Unit] = ???
}

class Allocator(implicit runtime: IORuntime) {
  // Ref that will keep track of finalizers
  private val shutdown: Ref[IO, IO[Unit]] = Ref.unsafe(IO.unit)

  // Method to allocate dependencies
  def allocate[A](resource: Resource[IO, A]): A =
    resource.allocated.flatMap { case (a, release) =>
      // Shutdown this resource, and after shutdown all previous
      shutdown.update(release *> _).map(_ => a)
    }.unsafeRunSync()

  // Shutdown dependencies
  def shutdownAll: IO[Unit] = {
    shutdown.getAndSet(IO.unit).flatten
  }

}

// Dependency Injection part:
object Dependencies {
  // Safe method to create dependencies:
  def apply(runtime: IORuntime): Resource[IO, Dependencies] =
    Resource.make {
      IO(unsafeCreate(runtime))
    } {
      _.allocator.shutdownAll
    }

  // Unsafe method, use it carefully as no shutdown is executed:
  def unsafeCreate(runtime: IORuntime): Dependencies =
    new Dependencies(new Allocator()(runtime))
}

class Dependencies(val allocator: Allocator) {

  lazy val conn: ConnectionFactory = allocator.allocate {
    ???
  }

  lazy val repo = new Repository(conn)

  lazy val serviceA = new ServiceA(repo)

  lazy val serviceB = new ServiceB(repo)

  lazy val server = HttpServerTask(serviceA, serviceB)
}

// Runner class
object Main extends IOApp.Simple {
  // Now we can use any dependency and not just `server` as all of them are exposed:
  private val dependencies = Dependencies(IORuntime.global)

  override def run = dependencies.use(_.server.run)
}

В заключение приглашаем всех разработчиков на Scala на открытые уроки в OTUS:

• Type classes в Scala, 14 декабря

• Пишем акторы Akka на Scala, 20 декабря