Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Support node filters in cluster config #657

Merged
merged 1 commit into from
Jan 29, 2022
Merged

Support node filters in cluster config #657

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 9 additions & 1 deletion modules/core/src/main/scala/dev/profunktor/redis4cats/config.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,24 +18,32 @@ package dev.profunktor.redis4cats

import scala.concurrent.duration._

import io.lettuce.core.cluster.models.partitions.RedisClusterNode
import io.lettuce.core.cluster.ClusterClientOptions

object config {

// Builder-style abstract class instead of case class to allow for bincompat-friendly extension in future versions.
sealed abstract class Redis4CatsConfig {
val shutdown: ShutdownConfig
val topologyViewRefreshStrategy: TopologyViewRefreshStrategy
val nodeFilter: RedisClusterNode => Boolean
def withShutdown(shutdown: ShutdownConfig): Redis4CatsConfig
def withTopologyViewRefreshStrategy(strategy: TopologyViewRefreshStrategy): Redis4CatsConfig
def withNodeFilter(nodeFilter: RedisClusterNode => Boolean): Redis4CatsConfig
}

object Redis4CatsConfig {
private case class Redis4CatsConfigImpl(
shutdown: ShutdownConfig,
topologyViewRefreshStrategy: TopologyViewRefreshStrategy = NoRefresh
topologyViewRefreshStrategy: TopologyViewRefreshStrategy = NoRefresh,
nodeFilter: RedisClusterNode => Boolean = ClusterClientOptions.DEFAULT_NODE_FILTER.test
) extends Redis4CatsConfig {
override def withShutdown(_shutdown: ShutdownConfig): Redis4CatsConfig = copy(shutdown = _shutdown)
override def withTopologyViewRefreshStrategy(strategy: TopologyViewRefreshStrategy): Redis4CatsConfig =
copy(topologyViewRefreshStrategy = strategy)
override def withNodeFilter(_nodeFilter: RedisClusterNode => Boolean): Redis4CatsConfig =
copy(nodeFilter = _nodeFilter)
}
def apply(): Redis4CatsConfig = Redis4CatsConfigImpl(ShutdownConfig())
}
Expand Down
15 changes: 12 additions & 3 deletions modules/core/src/main/scala/dev/profunktor/redis4cats/connection/RedisClusterClient.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ import dev.profunktor.redis4cats.JavaConversions._
import dev.profunktor.redis4cats.config._
import dev.profunktor.redis4cats.data.NodeId
import dev.profunktor.redis4cats.effect._
import io.lettuce.core.cluster.models.partitions.{ Partitions => JPartitions }
import io.lettuce.core.cluster.models.partitions.{ RedisClusterNode, Partitions => JPartitions }
import io.lettuce.core.cluster.{
ClusterClientOptions,
ClusterTopologyRefreshOptions,
Expand All @@ -47,7 +47,7 @@ object RedisClusterClient {
Log[F].info(s"Acquire Redis Cluster client") *>
RedisExecutor[F]
.lift(JClusterClient.create(uri.map(_.underlying).asJava))
.flatTap(initializeClusterTopology[F](_, config.topologyViewRefreshStrategy))
.flatTap(initializeClusterTopology[F](_, config.topologyViewRefreshStrategy, config.nodeFilter))
.map(new RedisClusterClient(_) {})

val release: RedisClusterClient => F[Unit] = client =>
Expand All @@ -69,11 +69,18 @@ object RedisClusterClient {

private[redis4cats] def initializeClusterTopology[F[_]: Functor: RedisExecutor](
client: JClusterClient,
topologyViewRefreshStrategy: TopologyViewRefreshStrategy
topologyViewRefreshStrategy: TopologyViewRefreshStrategy,
nodeFilter: RedisClusterNode => Boolean
): F[Unit] =
RedisExecutor[F].lift {
topologyViewRefreshStrategy match {
case NoRefresh =>
client.setOptions(
ClusterClientOptions
.builder()
.nodeFilter(nodeFilter(_))
.build()
)
client.getPartitions
case Periodic(interval) =>
client.setOptions(
Expand All @@ -86,6 +93,7 @@ object RedisClusterClient {
.enablePeriodicRefresh(Duration.ofMillis(interval.toMillis))
.build()
)
.nodeFilter(nodeFilter(_))
.build()
)
client.getPartitions
Expand All @@ -101,6 +109,7 @@ object RedisClusterClient {
.adaptiveRefreshTriggersTimeout(Duration.ofMillis(timeout.toMillis))
.build()
)
.nodeFilter(nodeFilter(_))
.build()
)
client.getPartitions
Expand Down