You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 

745 lines
22 KiB

  1. package redis
  2. import (
  3. "context"
  4. "crypto/tls"
  5. "errors"
  6. "fmt"
  7. "net"
  8. "net/url"
  9. "runtime"
  10. "sort"
  11. "strconv"
  12. "strings"
  13. "time"
  14. "github.com/redis/go-redis/v9/auth"
  15. "github.com/redis/go-redis/v9/internal/pool"
  16. "github.com/redis/go-redis/v9/internal/proto"
  17. "github.com/redis/go-redis/v9/internal/util"
  18. "github.com/redis/go-redis/v9/maintnotifications"
  19. "github.com/redis/go-redis/v9/push"
  20. )
  21. // Limiter is the interface of a rate limiter or a circuit breaker.
  22. type Limiter interface {
  23. // Allow returns nil if operation is allowed or an error otherwise.
  24. // If operation is allowed client must ReportResult of the operation
  25. // whether it is a success or a failure.
  26. Allow() error
  27. // ReportResult reports the result of the previously allowed operation.
  28. // nil indicates a success, non-nil error usually indicates a failure.
  29. ReportResult(result error)
  30. }
  31. // Options keeps the settings to set up redis connection.
  32. type Options struct {
  33. // Network type, either tcp or unix.
  34. //
  35. // default: is tcp.
  36. Network string
  37. // Addr is the address formated as host:port
  38. Addr string
  39. // ClientName will execute the `CLIENT SETNAME ClientName` command for each conn.
  40. ClientName string
  41. // Dialer creates new network connection and has priority over
  42. // Network and Addr options.
  43. Dialer func(ctx context.Context, network, addr string) (net.Conn, error)
  44. // Hook that is called when new connection is established.
  45. OnConnect func(ctx context.Context, cn *Conn) error
  46. // Protocol 2 or 3. Use the version to negotiate RESP version with redis-server.
  47. //
  48. // default: 3.
  49. Protocol int
  50. // Username is used to authenticate the current connection
  51. // with one of the connections defined in the ACL list when connecting
  52. // to a Redis 6.0 instance, or greater, that is using the Redis ACL system.
  53. Username string
  54. // Password is an optional password. Must match the password specified in the
  55. // `requirepass` server configuration option (if connecting to a Redis 5.0 instance, or lower),
  56. // or the User Password when connecting to a Redis 6.0 instance, or greater,
  57. // that is using the Redis ACL system.
  58. Password string
  59. // CredentialsProvider allows the username and password to be updated
  60. // before reconnecting. It should return the current username and password.
  61. CredentialsProvider func() (username string, password string)
  62. // CredentialsProviderContext is an enhanced parameter of CredentialsProvider,
  63. // done to maintain API compatibility. In the future,
  64. // there might be a merge between CredentialsProviderContext and CredentialsProvider.
  65. // There will be a conflict between them; if CredentialsProviderContext exists, we will ignore CredentialsProvider.
  66. CredentialsProviderContext func(ctx context.Context) (username string, password string, err error)
  67. // StreamingCredentialsProvider is used to retrieve the credentials
  68. // for the connection from an external source. Those credentials may change
  69. // during the connection lifetime. This is useful for managed identity
  70. // scenarios where the credentials are retrieved from an external source.
  71. //
  72. // Currently, this is a placeholder for the future implementation.
  73. StreamingCredentialsProvider auth.StreamingCredentialsProvider
  74. // DB is the database to be selected after connecting to the server.
  75. DB int
  76. // MaxRetries is the maximum number of retries before giving up.
  77. // -1 (not 0) disables retries.
  78. //
  79. // default: 3 retries
  80. MaxRetries int
  81. // MinRetryBackoff is the minimum backoff between each retry.
  82. // -1 disables backoff.
  83. //
  84. // default: 8 milliseconds
  85. MinRetryBackoff time.Duration
  86. // MaxRetryBackoff is the maximum backoff between each retry.
  87. // -1 disables backoff.
  88. // default: 512 milliseconds;
  89. MaxRetryBackoff time.Duration
  90. // DialTimeout for establishing new connections.
  91. //
  92. // default: 5 seconds
  93. DialTimeout time.Duration
  94. // DialerRetries is the maximum number of retry attempts when dialing fails.
  95. //
  96. // default: 5
  97. DialerRetries int
  98. // DialerRetryTimeout is the backoff duration between retry attempts.
  99. //
  100. // default: 100 milliseconds
  101. DialerRetryTimeout time.Duration
  102. // ReadTimeout for socket reads. If reached, commands will fail
  103. // with a timeout instead of blocking. Supported values:
  104. //
  105. // - `-1` - no timeout (block indefinitely).
  106. // - `-2` - disables SetReadDeadline calls completely.
  107. //
  108. // default: 3 seconds
  109. ReadTimeout time.Duration
  110. // WriteTimeout for socket writes. If reached, commands will fail
  111. // with a timeout instead of blocking. Supported values:
  112. //
  113. // - `-1` - no timeout (block indefinitely).
  114. // - `-2` - disables SetWriteDeadline calls completely.
  115. //
  116. // default: 3 seconds
  117. WriteTimeout time.Duration
  118. // ContextTimeoutEnabled controls whether the client respects context timeouts and deadlines.
  119. // See https://redis.uptrace.dev/guide/go-redis-debugging.html#timeouts
  120. ContextTimeoutEnabled bool
  121. // ReadBufferSize is the size of the bufio.Reader buffer for each connection.
  122. // Larger buffers can improve performance for commands that return large responses.
  123. // Smaller buffers can improve memory usage for larger pools.
  124. //
  125. // default: 32KiB (32768 bytes)
  126. ReadBufferSize int
  127. // WriteBufferSize is the size of the bufio.Writer buffer for each connection.
  128. // Larger buffers can improve performance for large pipelines and commands with many arguments.
  129. // Smaller buffers can improve memory usage for larger pools.
  130. //
  131. // default: 32KiB (32768 bytes)
  132. WriteBufferSize int
  133. // PoolFIFO type of connection pool.
  134. //
  135. // - true for FIFO pool
  136. // - false for LIFO pool.
  137. //
  138. // Note that FIFO has slightly higher overhead compared to LIFO,
  139. // but it helps closing idle connections faster reducing the pool size.
  140. // default: false
  141. PoolFIFO bool
  142. // PoolSize is the base number of socket connections.
  143. // Default is 10 connections per every available CPU as reported by runtime.GOMAXPROCS.
  144. // If there is not enough connections in the pool, new connections will be allocated in excess of PoolSize,
  145. // you can limit it through MaxActiveConns
  146. //
  147. // default: 10 * runtime.GOMAXPROCS(0)
  148. PoolSize int
  149. // PoolTimeout is the amount of time client waits for connection if all connections
  150. // are busy before returning an error.
  151. //
  152. // default: ReadTimeout + 1 second
  153. PoolTimeout time.Duration
  154. // MinIdleConns is the minimum number of idle connections which is useful when establishing
  155. // new connection is slow. The idle connections are not closed by default.
  156. //
  157. // default: 0
  158. MinIdleConns int
  159. // MaxIdleConns is the maximum number of idle connections.
  160. // The idle connections are not closed by default.
  161. //
  162. // default: 0
  163. MaxIdleConns int
  164. // MaxActiveConns is the maximum number of connections allocated by the pool at a given time.
  165. // When zero, there is no limit on the number of connections in the pool.
  166. // If the pool is full, the next call to Get() will block until a connection is released.
  167. MaxActiveConns int
  168. // ConnMaxIdleTime is the maximum amount of time a connection may be idle.
  169. // Should be less than server's timeout.
  170. //
  171. // Expired connections may be closed lazily before reuse.
  172. // If d <= 0, connections are not closed due to a connection's idle time.
  173. // -1 disables idle timeout check.
  174. //
  175. // default: 30 minutes
  176. ConnMaxIdleTime time.Duration
  177. // ConnMaxLifetime is the maximum amount of time a connection may be reused.
  178. //
  179. // Expired connections may be closed lazily before reuse.
  180. // If <= 0, connections are not closed due to a connection's age.
  181. //
  182. // default: 0
  183. ConnMaxLifetime time.Duration
  184. // TLSConfig to use. When set, TLS will be negotiated.
  185. TLSConfig *tls.Config
  186. // Limiter interface used to implement circuit breaker or rate limiter.
  187. Limiter Limiter
  188. // readOnly enables read only queries on slave/follower nodes.
  189. readOnly bool
  190. // DisableIndentity - Disable set-lib on connect.
  191. //
  192. // default: false
  193. //
  194. // Deprecated: Use DisableIdentity instead.
  195. DisableIndentity bool
  196. // DisableIdentity is used to disable CLIENT SETINFO command on connect.
  197. //
  198. // default: false
  199. DisableIdentity bool
  200. // Add suffix to client name. Default is empty.
  201. // IdentitySuffix - add suffix to client name.
  202. IdentitySuffix string
  203. // UnstableResp3 enables Unstable mode for Redis Search module with RESP3.
  204. // When unstable mode is enabled, the client will use RESP3 protocol and only be able to use RawResult
  205. UnstableResp3 bool
  206. // Push notifications are always enabled for RESP3 connections (Protocol: 3)
  207. // and are not available for RESP2 connections. No configuration option is needed.
  208. // PushNotificationProcessor is the processor for handling push notifications.
  209. // If nil, a default processor will be created for RESP3 connections.
  210. PushNotificationProcessor push.NotificationProcessor
  211. // FailingTimeoutSeconds is the timeout in seconds for marking a cluster node as failing.
  212. // When a node is marked as failing, it will be avoided for this duration.
  213. // Default is 15 seconds.
  214. FailingTimeoutSeconds int
  215. // MaintNotificationsConfig provides custom configuration for maintnotifications.
  216. // When MaintNotificationsConfig.Mode is not "disabled", the client will handle
  217. // cluster upgrade notifications gracefully and manage connection/pool state
  218. // transitions seamlessly. Requires Protocol: 3 (RESP3) for push notifications.
  219. // If nil, maintnotifications are in "auto" mode and will be enabled if the server supports it.
  220. MaintNotificationsConfig *maintnotifications.Config
  221. }
  222. func (opt *Options) init() {
  223. if opt.Addr == "" {
  224. opt.Addr = "localhost:6379"
  225. }
  226. if opt.Network == "" {
  227. if strings.HasPrefix(opt.Addr, "/") {
  228. opt.Network = "unix"
  229. } else {
  230. opt.Network = "tcp"
  231. }
  232. }
  233. if opt.Protocol < 2 {
  234. opt.Protocol = 3
  235. }
  236. if opt.DialTimeout == 0 {
  237. opt.DialTimeout = 5 * time.Second
  238. }
  239. if opt.DialerRetries == 0 {
  240. opt.DialerRetries = 5
  241. }
  242. if opt.DialerRetryTimeout == 0 {
  243. opt.DialerRetryTimeout = 100 * time.Millisecond
  244. }
  245. if opt.Dialer == nil {
  246. opt.Dialer = NewDialer(opt)
  247. }
  248. if opt.PoolSize == 0 {
  249. opt.PoolSize = 10 * runtime.GOMAXPROCS(0)
  250. }
  251. if opt.ReadBufferSize == 0 {
  252. opt.ReadBufferSize = proto.DefaultBufferSize
  253. }
  254. if opt.WriteBufferSize == 0 {
  255. opt.WriteBufferSize = proto.DefaultBufferSize
  256. }
  257. switch opt.ReadTimeout {
  258. case -2:
  259. opt.ReadTimeout = -1
  260. case -1:
  261. opt.ReadTimeout = 0
  262. case 0:
  263. opt.ReadTimeout = 3 * time.Second
  264. }
  265. switch opt.WriteTimeout {
  266. case -2:
  267. opt.WriteTimeout = -1
  268. case -1:
  269. opt.WriteTimeout = 0
  270. case 0:
  271. opt.WriteTimeout = opt.ReadTimeout
  272. }
  273. if opt.PoolTimeout == 0 {
  274. if opt.ReadTimeout > 0 {
  275. opt.PoolTimeout = opt.ReadTimeout + time.Second
  276. } else {
  277. opt.PoolTimeout = 30 * time.Second
  278. }
  279. }
  280. if opt.ConnMaxIdleTime == 0 {
  281. opt.ConnMaxIdleTime = 30 * time.Minute
  282. }
  283. switch opt.MaxRetries {
  284. case -1:
  285. opt.MaxRetries = 0
  286. case 0:
  287. opt.MaxRetries = 3
  288. }
  289. switch opt.MinRetryBackoff {
  290. case -1:
  291. opt.MinRetryBackoff = 0
  292. case 0:
  293. opt.MinRetryBackoff = 8 * time.Millisecond
  294. }
  295. switch opt.MaxRetryBackoff {
  296. case -1:
  297. opt.MaxRetryBackoff = 0
  298. case 0:
  299. opt.MaxRetryBackoff = 512 * time.Millisecond
  300. }
  301. opt.MaintNotificationsConfig = opt.MaintNotificationsConfig.ApplyDefaultsWithPoolConfig(opt.PoolSize, opt.MaxActiveConns)
  302. // auto-detect endpoint type if not specified
  303. endpointType := opt.MaintNotificationsConfig.EndpointType
  304. if endpointType == "" || endpointType == maintnotifications.EndpointTypeAuto {
  305. // Auto-detect endpoint type if not specified
  306. endpointType = maintnotifications.DetectEndpointType(opt.Addr, opt.TLSConfig != nil)
  307. }
  308. opt.MaintNotificationsConfig.EndpointType = endpointType
  309. }
  310. func (opt *Options) clone() *Options {
  311. clone := *opt
  312. // Deep clone MaintNotificationsConfig to avoid sharing between clients
  313. if opt.MaintNotificationsConfig != nil {
  314. configClone := *opt.MaintNotificationsConfig
  315. clone.MaintNotificationsConfig = &configClone
  316. }
  317. return &clone
  318. }
  319. // NewDialer returns a function that will be used as the default dialer
  320. // when none is specified in Options.Dialer.
  321. func (opt *Options) NewDialer() func(context.Context, string, string) (net.Conn, error) {
  322. return NewDialer(opt)
  323. }
  324. // NewDialer returns a function that will be used as the default dialer
  325. // when none is specified in Options.Dialer.
  326. func NewDialer(opt *Options) func(context.Context, string, string) (net.Conn, error) {
  327. return func(ctx context.Context, network, addr string) (net.Conn, error) {
  328. netDialer := &net.Dialer{
  329. Timeout: opt.DialTimeout,
  330. KeepAlive: 5 * time.Minute,
  331. }
  332. if opt.TLSConfig == nil {
  333. return netDialer.DialContext(ctx, network, addr)
  334. }
  335. return tls.DialWithDialer(netDialer, network, addr, opt.TLSConfig)
  336. }
  337. }
  338. // ParseURL parses a URL into Options that can be used to connect to Redis.
  339. // Scheme is required.
  340. // There are two connection types: by tcp socket and by unix socket.
  341. // Tcp connection:
  342. //
  343. // redis://<user>:<password>@<host>:<port>/<db_number>
  344. //
  345. // Unix connection:
  346. //
  347. // unix://<user>:<password>@</path/to/redis.sock>?db=<db_number>
  348. //
  349. // Most Option fields can be set using query parameters, with the following restrictions:
  350. // - field names are mapped using snake-case conversion: to set MaxRetries, use max_retries
  351. // - only scalar type fields are supported (bool, int, time.Duration)
  352. // - for time.Duration fields, values must be a valid input for time.ParseDuration();
  353. // additionally a plain integer as value (i.e. without unit) is interpreted as seconds
  354. // - to disable a duration field, use value less than or equal to 0; to use the default
  355. // value, leave the value blank or remove the parameter
  356. // - only the last value is interpreted if a parameter is given multiple times
  357. // - fields "network", "addr", "username" and "password" can only be set using other
  358. // URL attributes (scheme, host, userinfo, resp.), query parameters using these
  359. // names will be treated as unknown parameters
  360. // - unknown parameter names will result in an error
  361. // - use "skip_verify=true" to ignore TLS certificate validation
  362. //
  363. // Examples:
  364. //
  365. // redis://user:password@localhost:6789/3?dial_timeout=3&db=1&read_timeout=6s&max_retries=2
  366. // is equivalent to:
  367. // &Options{
  368. // Network: "tcp",
  369. // Addr: "localhost:6789",
  370. // DB: 1, // path "/3" was overridden by "&db=1"
  371. // DialTimeout: 3 * time.Second, // no time unit = seconds
  372. // ReadTimeout: 6 * time.Second,
  373. // MaxRetries: 2,
  374. // }
  375. func ParseURL(redisURL string) (*Options, error) {
  376. u, err := url.Parse(redisURL)
  377. if err != nil {
  378. return nil, err
  379. }
  380. switch u.Scheme {
  381. case "redis", "rediss":
  382. return setupTCPConn(u)
  383. case "unix":
  384. return setupUnixConn(u)
  385. default:
  386. return nil, fmt.Errorf("redis: invalid URL scheme: %s", u.Scheme)
  387. }
  388. }
  389. func setupTCPConn(u *url.URL) (*Options, error) {
  390. o := &Options{Network: "tcp"}
  391. o.Username, o.Password = getUserPassword(u)
  392. h, p := getHostPortWithDefaults(u)
  393. o.Addr = net.JoinHostPort(h, p)
  394. f := strings.FieldsFunc(u.Path, func(r rune) bool {
  395. return r == '/'
  396. })
  397. switch len(f) {
  398. case 0:
  399. o.DB = 0
  400. case 1:
  401. var err error
  402. if o.DB, err = strconv.Atoi(f[0]); err != nil {
  403. return nil, fmt.Errorf("redis: invalid database number: %q", f[0])
  404. }
  405. default:
  406. return nil, fmt.Errorf("redis: invalid URL path: %s", u.Path)
  407. }
  408. if u.Scheme == "rediss" {
  409. o.TLSConfig = &tls.Config{
  410. ServerName: h,
  411. MinVersion: tls.VersionTLS12,
  412. }
  413. }
  414. return setupConnParams(u, o)
  415. }
  416. // getHostPortWithDefaults is a helper function that splits the url into
  417. // a host and a port. If the host is missing, it defaults to localhost
  418. // and if the port is missing, it defaults to 6379.
  419. func getHostPortWithDefaults(u *url.URL) (string, string) {
  420. host, port, err := net.SplitHostPort(u.Host)
  421. if err != nil {
  422. host = u.Host
  423. }
  424. if host == "" {
  425. host = "localhost"
  426. }
  427. if port == "" {
  428. port = "6379"
  429. }
  430. return host, port
  431. }
  432. func setupUnixConn(u *url.URL) (*Options, error) {
  433. o := &Options{
  434. Network: "unix",
  435. }
  436. if strings.TrimSpace(u.Path) == "" { // path is required with unix connection
  437. return nil, errors.New("redis: empty unix socket path")
  438. }
  439. o.Addr = u.Path
  440. o.Username, o.Password = getUserPassword(u)
  441. return setupConnParams(u, o)
  442. }
  443. type queryOptions struct {
  444. q url.Values
  445. err error
  446. }
  447. func (o *queryOptions) has(name string) bool {
  448. return len(o.q[name]) > 0
  449. }
  450. func (o *queryOptions) string(name string) string {
  451. vs := o.q[name]
  452. if len(vs) == 0 {
  453. return ""
  454. }
  455. delete(o.q, name) // enable detection of unknown parameters
  456. return vs[len(vs)-1]
  457. }
  458. func (o *queryOptions) strings(name string) []string {
  459. vs := o.q[name]
  460. delete(o.q, name)
  461. return vs
  462. }
  463. func (o *queryOptions) int(name string) int {
  464. s := o.string(name)
  465. if s == "" {
  466. return 0
  467. }
  468. i, err := strconv.Atoi(s)
  469. if err == nil {
  470. return i
  471. }
  472. if o.err == nil {
  473. o.err = fmt.Errorf("redis: invalid %s number: %s", name, err)
  474. }
  475. return 0
  476. }
  477. func (o *queryOptions) duration(name string) time.Duration {
  478. s := o.string(name)
  479. if s == "" {
  480. return 0
  481. }
  482. // try plain number first
  483. if i, err := strconv.Atoi(s); err == nil {
  484. if i <= 0 {
  485. // disable timeouts
  486. return -1
  487. }
  488. return time.Duration(i) * time.Second
  489. }
  490. dur, err := time.ParseDuration(s)
  491. if err == nil {
  492. return dur
  493. }
  494. if o.err == nil {
  495. o.err = fmt.Errorf("redis: invalid %s duration: %w", name, err)
  496. }
  497. return 0
  498. }
  499. func (o *queryOptions) bool(name string) bool {
  500. switch s := o.string(name); s {
  501. case "true", "1":
  502. return true
  503. case "false", "0", "":
  504. return false
  505. default:
  506. if o.err == nil {
  507. o.err = fmt.Errorf("redis: invalid %s boolean: expected true/false/1/0 or an empty string, got %q", name, s)
  508. }
  509. return false
  510. }
  511. }
  512. func (o *queryOptions) remaining() []string {
  513. if len(o.q) == 0 {
  514. return nil
  515. }
  516. keys := make([]string, 0, len(o.q))
  517. for k := range o.q {
  518. keys = append(keys, k)
  519. }
  520. sort.Strings(keys)
  521. return keys
  522. }
  523. // setupConnParams converts query parameters in u to option value in o.
  524. func setupConnParams(u *url.URL, o *Options) (*Options, error) {
  525. q := queryOptions{q: u.Query()}
  526. // compat: a future major release may use q.int("db")
  527. if tmp := q.string("db"); tmp != "" {
  528. db, err := strconv.Atoi(tmp)
  529. if err != nil {
  530. return nil, fmt.Errorf("redis: invalid database number: %w", err)
  531. }
  532. o.DB = db
  533. }
  534. o.Protocol = q.int("protocol")
  535. o.ClientName = q.string("client_name")
  536. o.MaxRetries = q.int("max_retries")
  537. o.MinRetryBackoff = q.duration("min_retry_backoff")
  538. o.MaxRetryBackoff = q.duration("max_retry_backoff")
  539. o.DialTimeout = q.duration("dial_timeout")
  540. o.ReadTimeout = q.duration("read_timeout")
  541. o.WriteTimeout = q.duration("write_timeout")
  542. o.PoolFIFO = q.bool("pool_fifo")
  543. o.PoolSize = q.int("pool_size")
  544. o.PoolTimeout = q.duration("pool_timeout")
  545. o.MinIdleConns = q.int("min_idle_conns")
  546. o.MaxIdleConns = q.int("max_idle_conns")
  547. o.MaxActiveConns = q.int("max_active_conns")
  548. if q.has("conn_max_idle_time") {
  549. o.ConnMaxIdleTime = q.duration("conn_max_idle_time")
  550. } else {
  551. o.ConnMaxIdleTime = q.duration("idle_timeout")
  552. }
  553. if q.has("conn_max_lifetime") {
  554. o.ConnMaxLifetime = q.duration("conn_max_lifetime")
  555. } else {
  556. o.ConnMaxLifetime = q.duration("max_conn_age")
  557. }
  558. if q.err != nil {
  559. return nil, q.err
  560. }
  561. if o.TLSConfig != nil && q.has("skip_verify") {
  562. o.TLSConfig.InsecureSkipVerify = q.bool("skip_verify")
  563. }
  564. // any parameters left?
  565. if r := q.remaining(); len(r) > 0 {
  566. return nil, fmt.Errorf("redis: unexpected option: %s", strings.Join(r, ", "))
  567. }
  568. return o, nil
  569. }
  570. func getUserPassword(u *url.URL) (string, string) {
  571. var user, password string
  572. if u.User != nil {
  573. user = u.User.Username()
  574. if p, ok := u.User.Password(); ok {
  575. password = p
  576. }
  577. }
  578. return user, password
  579. }
  580. func newConnPool(
  581. opt *Options,
  582. dialer func(ctx context.Context, network, addr string) (net.Conn, error),
  583. ) (*pool.ConnPool, error) {
  584. poolSize, err := util.SafeIntToInt32(opt.PoolSize, "PoolSize")
  585. if err != nil {
  586. return nil, err
  587. }
  588. minIdleConns, err := util.SafeIntToInt32(opt.MinIdleConns, "MinIdleConns")
  589. if err != nil {
  590. return nil, err
  591. }
  592. maxIdleConns, err := util.SafeIntToInt32(opt.MaxIdleConns, "MaxIdleConns")
  593. if err != nil {
  594. return nil, err
  595. }
  596. maxActiveConns, err := util.SafeIntToInt32(opt.MaxActiveConns, "MaxActiveConns")
  597. if err != nil {
  598. return nil, err
  599. }
  600. return pool.NewConnPool(&pool.Options{
  601. Dialer: func(ctx context.Context) (net.Conn, error) {
  602. return dialer(ctx, opt.Network, opt.Addr)
  603. },
  604. PoolFIFO: opt.PoolFIFO,
  605. PoolSize: poolSize,
  606. PoolTimeout: opt.PoolTimeout,
  607. DialTimeout: opt.DialTimeout,
  608. DialerRetries: opt.DialerRetries,
  609. DialerRetryTimeout: opt.DialerRetryTimeout,
  610. MinIdleConns: minIdleConns,
  611. MaxIdleConns: maxIdleConns,
  612. MaxActiveConns: maxActiveConns,
  613. ConnMaxIdleTime: opt.ConnMaxIdleTime,
  614. ConnMaxLifetime: opt.ConnMaxLifetime,
  615. ReadBufferSize: opt.ReadBufferSize,
  616. WriteBufferSize: opt.WriteBufferSize,
  617. PushNotificationsEnabled: opt.Protocol == 3,
  618. }), nil
  619. }
  620. func newPubSubPool(opt *Options, dialer func(ctx context.Context, network, addr string) (net.Conn, error),
  621. ) (*pool.PubSubPool, error) {
  622. poolSize, err := util.SafeIntToInt32(opt.PoolSize, "PoolSize")
  623. if err != nil {
  624. return nil, err
  625. }
  626. minIdleConns, err := util.SafeIntToInt32(opt.MinIdleConns, "MinIdleConns")
  627. if err != nil {
  628. return nil, err
  629. }
  630. maxIdleConns, err := util.SafeIntToInt32(opt.MaxIdleConns, "MaxIdleConns")
  631. if err != nil {
  632. return nil, err
  633. }
  634. maxActiveConns, err := util.SafeIntToInt32(opt.MaxActiveConns, "MaxActiveConns")
  635. if err != nil {
  636. return nil, err
  637. }
  638. return pool.NewPubSubPool(&pool.Options{
  639. PoolFIFO: opt.PoolFIFO,
  640. PoolSize: poolSize,
  641. PoolTimeout: opt.PoolTimeout,
  642. DialTimeout: opt.DialTimeout,
  643. DialerRetries: opt.DialerRetries,
  644. DialerRetryTimeout: opt.DialerRetryTimeout,
  645. MinIdleConns: minIdleConns,
  646. MaxIdleConns: maxIdleConns,
  647. MaxActiveConns: maxActiveConns,
  648. ConnMaxIdleTime: opt.ConnMaxIdleTime,
  649. ConnMaxLifetime: opt.ConnMaxLifetime,
  650. ReadBufferSize: 32 * 1024,
  651. WriteBufferSize: 32 * 1024,
  652. PushNotificationsEnabled: opt.Protocol == 3,
  653. }, dialer), nil
  654. }