Parallel take first adapter

Author: peterbanda

openai-core/src/main/scala/io/cequence/openaiscala/service/adapter/ChatCompletionServiceAdapter.scala

@@ -31,5 +31,4 @@ private class ChatCompletionServiceAdapter[S <: CloseableService](
     chatCompletionService.close()
     underlying.close()
   }
-
 }

openai-core/src/main/scala/io/cequence/openaiscala/service/adapter/MultiServiceAdapter.scala

@@ -36,4 +36,4 @@ private class RandomOrderAdapter[+S <: CloseableService](
   val underlyings: Seq[S]
 ) extends MultiServiceAdapter[S] {
   protected def calcIndex: Int = Random.nextInt(count)
-}
+}

openai-core/src/main/scala/io/cequence/openaiscala/service/adapter/OpenAIServiceAdapters.scala

@@ -1,6 +1,7 @@
 package io.cequence.openaiscala.service.adapter
 
 import akka.actor.Scheduler
+import akka.stream.Materializer
 import io.cequence.openaiscala.RetryHelpers.RetrySettings
 import io.cequence.openaiscala.domain.BaseMessage
 import io.cequence.openaiscala.domain.settings.CreateChatCompletionSettings
@@ -34,6 +35,12 @@ trait OpenAIServiceAdapters[S <: CloseableService] {
   ): S =
     wrapAndDelegate(new RandomOrderAdapter(underlyings))
 
+  def parallelTakeFirst(
+    underlyings: S*)(
+    implicit materializer: Materializer
+  ): S =
+    wrapAndDelegate(new ParallelTakeFirstAdapter(underlyings))
+
   def retry(
     underlying: S,
     log: Option[String => Unit] = None

openai-core/src/main/scala/io/cequence/openaiscala/service/adapter/ParallelTakeFirstAdapter.scala

@@ -0,0 +1,33 @@
+package io.cequence.openaiscala.service.adapter
+
+import akka.stream.Materializer
+import akka.stream.scaladsl.{Sink, Source}
+import io.cequence.wsclient.service.CloseableService
+import org.slf4j.LoggerFactory
+
+import scala.concurrent.Future
+
+private class ParallelTakeFirstAdapter[+S <: CloseableService](
+  underlyings: Seq[S]
+)(
+  implicit materializer: Materializer
+) extends ServiceWrapper[S]
+    with CloseableService {
+
+  private val logger = LoggerFactory.getLogger(getClass)
+
+  override protected[adapter] def wrap[T](
+    fun: S => Future[T]
+  ): Future[T] = {
+    logger.debug(s"Running parallel/redundant processing with ${underlyings.size} services.")
+
+    val sources = Source
+      .fromIterator(() => underlyings.toIterator)
+      .mapAsyncUnordered(underlyings.size)(fun)
+
+    sources.runWith(Sink.head)
+  }
+
+  override def close(): Unit =
+    underlyings.foreach(_.close())
+}