1 /***********************************************************************
2  * Copyright (c) 2013-2025 General Atomics Integrated Intelligence, Inc.
3  * All rights reserved. This program and the accompanying materials
4  * are made available under the terms of the Apache License, Version 2.0
5  * which accompanies this distribution and is available at
6  * https://www.apache.org/licenses/LICENSE-2.0
7  ***********************************************************************/
8 
9 package org.locationtech.geomesa.redis.data
10 
11 import org.apache.commons.pool2.impl.GenericObjectPoolConfig
12 import org.apache.commons.pool2.{PooledObject, PooledObjectFactory}
13 import redis.clients.jedis.util.Pool
14 import redis.clients.jedis.{Connection, HostAndPort, JedisClientConfig, JedisCluster, UnifiedJedis}
15 
16 import java.time.Duration;
17 
18 /**
19  * Noop factory for JedisCluster. This is used to create a singleton pool for Geomesa compatibility.
20  */
21 class NoopClusterFactory extends PooledObjectFactory[UnifiedJedis] {
22   override def activateObject(p: PooledObject[UnifiedJedis]): Unit = {}
23 
24   override def destroyObject(p: PooledObject[UnifiedJedis]): Unit = {}
25 
26   override def passivateObject(p: PooledObject[UnifiedJedis]): Unit = {}
27 
28   override def validateObject(p: PooledObject[UnifiedJedis]): Boolean = true
29 
30   override def makeObject(): PooledObject[UnifiedJedis] = null
31 }
32 
33 /**
34  * Singleton pool for JedisCluster. This is used to create a singleton pool for Geomesa compatibility.
35  * @param nodes
36  * @param clientConfig
37  */
38 class SingletonJedisClusterPool(nodes: java.util.Set[HostAndPort], clientConfig: JedisClientConfig, objectPoolConfig: GenericObjectPoolConfig[Connection])
39     extends Pool[UnifiedJedis](new NoopClusterFactory) {
40 
41   private val cluster = new JedisClusterUncloseable(
42     nodes,
43     clientConfig,
44     objectPoolConfig,
45     JedisCluster.DEFAULT_MAX_ATTEMPTS,
46     Duration.ofMillis(clientConfig.getSocketTimeoutMillis * JedisCluster.DEFAULT_MAX_ATTEMPTS)
47   )
48 
49   override def getResource: UnifiedJedis = cluster
50 
51   override def returnResource(resource: UnifiedJedis): Unit = {
52     // Do nothing - we're always returning the same instance
53   }
54 
55   override def returnBrokenResource(resource: UnifiedJedis): Unit = {
56     // Since JedisCluster handles its own connection recovery, we don't need to do anything here
57   }
58 
59   override def close(): Unit = {
60     // Close the cluster connections
61     cluster.closePool()
62     // close the enclosing pool
63     super.close()
64   }
65 }