Skip to content

Commit

Permalink
Use thread factory
Browse files Browse the repository at this point in the history
  • Loading branch information
retriku authored and jbwheatley committed Mar 7, 2022
1 parent eaca068 commit 2ae3814
Showing 1 changed file with 22 additions and 14 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -36,12 +36,12 @@ import dev.profunktor.fs2rabbit.model.RabbitConnection
import java.util.Collections
import java.util.concurrent.AbstractExecutorService
import java.util.concurrent.ExecutorService
import java.util.concurrent.Executors
import java.util.concurrent.ThreadFactory
import java.util.concurrent.TimeUnit
import javax.net.ssl.SSLContext
import scala.concurrent.ExecutionContext
import scala.jdk.CollectionConverters._
import java.util.concurrent.Executors

object ConnectionResource {
type ConnectionResource[F[_]] = Connection[Resource[F, *]]
Expand All @@ -63,20 +63,28 @@ object ConnectionResource {
}
}

val numOfThreads = Runtime.getRuntime().availableProcessors() * 2
val es = Executors.newFixedThreadPool(numOfThreads)
sys.addShutdownHook(es.shutdown())
val numOfThreads = Runtime.getRuntime().availableProcessors() * 2
val esF: F[ExecutorService] = threadFactory
.fold(Executors.newFixedThreadPool(numOfThreads).pure[F]) {
_.map(Executors.newFixedThreadPool(numOfThreads, _))
}
.map { es =>
sys.addShutdownHook(es.shutdown())
es
}

addThreadFactory.flatMap { fn =>
_make(
conf,
Some(ExecutionContext.fromExecutorService(es)),
sslCtx,
saslConf,
metricsCollector,
fn
)
}
for {
es <- esF
fn <- addThreadFactory
conn <- _make(
conf,
Some(ExecutionContext.fromExecutorService(es)),
sslCtx,
saslConf,
metricsCollector,
fn
)
} yield conn
}

def make[F[_]: Sync: Log](
Expand Down

0 comments on commit 2ae3814

Please sign in to comment.