Project has been created

.gitignore

@@ -0,0 +1,68 @@
+# Created by .ignore support plugin (hsz.mobi)
+### JetBrains template
+# Covers JetBrains IDEs: IntelliJ, RubyMine, PhpStorm, AppCode, PyCharm, CLion, Android Studio and Webstorm
+# Reference: https://intellij-support.jetbrains.com/hc/en-us/articles/206544839
+
+# User-specific stuff:
+.idea/**/workspace.xml
+.idea/**/tasks.xml
+.idea/dictionaries
+
+# Sensitive or high-churn files:
+.idea/**/dataSources/
+.idea/**/dataSources.ids
+.idea/**/dataSources.xml
+.idea/**/dataSources.local.xml
+.idea/**/sqlDataSources.xml
+.idea/**/dynamic.xml
+.idea/**/uiDesigner.xml
+
+# Gradle:
+.idea/**/gradle.xml
+.idea/**/libraries
+
+# Mongo Explorer plugin:
+.idea/**/mongoSettings.xml
+
+## File-based project format:
+*.iws
+
+## Plugin-specific files:
+
+# IntelliJ
+/out/
+
+# mpeltonen/sbt-idea plugin
+.idea_modules/
+
+# JIRA plugin
+atlassian-ide-plugin.xml
+
+# Crashlytics plugin (for Android Studio and IntelliJ)
+com_crashlytics_export_strings.xml
+crashlytics.properties
+crashlytics-build.properties
+fabric.properties
+### Go template
+# Binaries for programs and plugins
+*.exe
+*.dll
+*.so
+*.dylib
+
+# Test binary, build with `go test -c`
+*.test
+
+# Output of the go coverage tool, specifically when used with LiteIDE
+*.out
+
+# Project-local glide cache, RE: https://github.com/Masterminds/glide/issues/736
+.glide/
+.idea/
+*.iml
+
+vendor/
+glide.lock
+.DS_Store
+dist/
+debug

.vscode/launch.json

@@ -0,0 +1,32 @@
+{
+    "version": "0.2.0",
+    "configurations": [
+        {
+            "name": "Debug",
+            "type": "go",
+            "request": "launch",
+            "mode": "debug",
+            "remotePath": "",
+            "port": 2345,
+            "host": "127.0.0.1",
+            "program": "${workspaceRoot}/main.go",
+            "env": {},
+            "args": [],
+            "showLog": true
+        },
+        {
+            "name": "File Debug",
+            "type": "go",
+            "request": "launch",
+            "mode": "debug",
+            "remotePath": "",
+            "port": 2345,
+            "host": "127.0.0.1",
+            "program": "${fileDirname}",
+            "env": {},
+            "args": [],
+            "showLog": true
+        }
+
+    ]
+}

.vscode/settings.json

@@ -0,0 +1,11 @@
+// Place your settings in this file to overwrite default and user settings.
+{
+  // Specifies Lint tool name.
+  "go.lintTool": "gometalinter",
+
+  // Flags to pass to Lint tool (e.g. ["-min_confidence=.8"])
+  "go.lintFlags": [
+    "--config=${workspaceRoot}/golint.json"
+  ]
+
+}

glide.yaml

@@ -0,0 +1,6 @@
+package: git.loafle.net/commons_go/redis_pool
+import:
+- package: github.com/garyburd/redigo
+  version: v1.1.0
+  subpackages:
+  - redis

internal/commandinfo.go

@@ -0,0 +1,54 @@
+// Copyright 2014 Gary Burd
+//
+// Licensed under the Apache License, Version 2.0 (the "License"): you may
+// not use this file except in compliance with the License. You may obtain
+// a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+// License for the specific language governing permissions and limitations
+// under the License.
+
+package internal // import "git.loafle.net/commons_go/redis_pool/internal"
+
+import (
+	"strings"
+)
+
+const (
+	WatchState = 1 << iota
+	MultiState
+	SubscribeState
+	MonitorState
+)
+
+type CommandInfo struct {
+	Set, Clear int
+}
+
+var commandInfos = map[string]CommandInfo{
+	"WATCH":      {Set: WatchState},
+	"UNWATCH":    {Clear: WatchState},
+	"MULTI":      {Set: MultiState},
+	"EXEC":       {Clear: WatchState | MultiState},
+	"DISCARD":    {Clear: WatchState | MultiState},
+	"PSUBSCRIBE": {Set: SubscribeState},
+	"SUBSCRIBE":  {Set: SubscribeState},
+	"MONITOR":    {Set: MonitorState},
+}
+
+func init() {
+	for n, ci := range commandInfos {
+		commandInfos[strings.ToLower(n)] = ci
+	}
+}
+
+func LookupCommandInfo(commandName string) CommandInfo {
+	if ci, ok := commandInfos[commandName]; ok {
+		return ci
+	}
+	return commandInfos[strings.ToUpper(commandName)]
+}

pool.go

@@ -0,0 +1,385 @@
+package overflow_redis_pool
+
+import (
+	"bytes"
+	"container/list"
+	"context"
+	"crypto/rand"
+	"crypto/sha1"
+	"errors"
+	"io"
+	"strconv"
+	"sync"
+	"time"
+
+	"git.loafle.net/commons_go/redis_pool/internal"
+	"github.com/garyburd/redigo/redis"
+)
+
+var nowFunc = time.Now // for testing
+
+// ErrPoolExhausted is returned from a pool connection method (Do, Send,
+// Receive, Flush, Err) when the maximum number of database connections in the
+// pool has been reached.
+var ErrPoolExhausted = errors.New("redigo: connection pool exhausted")
+
+var (
+	errPoolClosed = errors.New("redigo: connection pool closed")
+	errConnClosed = errors.New("redigo: connection closed")
+)
+
+// Pool maintains a pool of connections. The application calls the Get method
+// to get a connection from the pool and the connection's Close method to
+// return the connection's resources to the pool.
+//
+// The following example shows how to use a pool in a web application. The
+// application creates a pool at application startup and makes it available to
+// request handlers using a package level variable. The pool configuration used
+// here is an example, not a recommendation.
+//
+//  func newPool(addr string) *redis.Pool {
+//    return &redis.Pool{
+//      MaxIdle: 3,
+//      IdleTimeout: 240 * time.Second,
+//      Dial: func () (redis.Conn, error) { return redis.Dial("tcp", addr) },
+//    }
+//  }
+//
+//  var (
+//    pool *redis.Pool
+//    redisServer = flag.String("redisServer", ":6379", "")
+//  )
+//
+//  func main() {
+//    flag.Parse()
+//    pool = newPool(*redisServer)
+//    ...
+//  }
+//
+// A request handler gets a connection from the pool and closes the connection
+// when the handler is done:
+//
+//  func serveHome(w http.ResponseWriter, r *http.Request) {
+//      conn := pool.Get()
+//      defer conn.Close()
+//      ...
+//  }
+//
+// Use the Dial function to authenticate connections with the AUTH command or
+// select a database with the SELECT command:
+//
+//  pool := &redis.Pool{
+//    // Other pool configuration not shown in this example.
+//    Dial: func () (redis.Conn, error) {
+//      c, err := redis.Dial("tcp", server)
+//      if err != nil {
+//        return nil, err
+//      }
+//      if _, err := c.Do("AUTH", password); err != nil {
+//        c.Close()
+//        return nil, err
+//      }
+//      if _, err := c.Do("SELECT", db); err != nil {
+//        c.Close()
+//        return nil, err
+//      }
+//      return c, nil
+//    }
+//  }
+//
+// Use the TestOnBorrow function to check the health of an idle connection
+// before the connection is returned to the application. This example PINGs
+// connections that have been idle more than a minute:
+//
+//  pool := &redis.Pool{
+//    // Other pool configuration not shown in this example.
+//    TestOnBorrow: func(c redis.Conn, t time.Time) error {
+//      if time.Since(t) < time.Minute {
+//        return nil
+//      }
+//      _, err := c.Do("PING")
+//      return err
+//    },
+//  }
+//
+type Pool interface {
+	Get() redis.Conn
+	ActiveCount() int
+	Close() error
+}
+
+type pool struct {
+	ctx     context.Context
+	handler PoolHandler
+	// mu protects fields defined below.
+	mu     sync.Mutex
+	cond   *sync.Cond
+	closed bool
+	active int
+
+	// Stack of idleConn with most recently used at the front.
+	idle list.List
+}
+
+type idleConn struct {
+	c redis.Conn
+	t time.Time
+}
+
+// NewPool creates a new pool.
+//
+// Deprecated: Initialize the Pool directory as shown in the example.
+func NewPool(ctx context.Context, handler PoolHandler) Pool {
+	p := &pool{
+		ctx:     ctx,
+		handler: handler,
+	}
+	return p
+}
+
+// Get gets a connection. The application must close the returned connection.
+// This method always returns a valid connection so that applications can defer
+// error handling to the first use of the connection. If there is an error
+// getting an underlying connection, then the connection Err, Do, Send, Flush
+// and Receive methods return that error.
+func (p *pool) Get() redis.Conn {
+	c, err := p.get()
+	if err != nil {
+		return errorConnection{err}
+	}
+	return &pooledConnection{p: p, c: c}
+}
+
+// ActiveCount returns the number of active connections in the pool.
+func (p *pool) ActiveCount() int {
+	p.mu.Lock()
+	active := p.active
+	p.mu.Unlock()
+	return active
+}
+
+// Close releases the resources used by the pool.
+func (p *pool) Close() error {
+	p.mu.Lock()
+	idle := p.idle
+	p.idle.Init()
+	p.closed = true
+	p.active -= idle.Len()
+	if p.cond != nil {
+		p.cond.Broadcast()
+	}
+	p.mu.Unlock()
+	for e := idle.Front(); e != nil; e = e.Next() {
+		e.Value.(idleConn).c.Close()
+	}
+	return nil
+}
+
+// release decrements the active count and signals waiters. The caller must
+// hold p.mu during the call.
+func (p *pool) release() {
+	p.active -= 1
+	if p.cond != nil {
+		p.cond.Signal()
+	}
+}
+
+// get prunes stale connections and returns a connection from the idle list or
+// creates a new connection.
+func (p *pool) get() (redis.Conn, error) {
+	p.mu.Lock()
+
+	// Prune stale connections.
+
+	if timeout := p.handler.GetIdleTimeout(); timeout > 0 {
+		for i, n := 0, p.idle.Len(); i < n; i++ {
+			e := p.idle.Back()
+			if e == nil {
+				break
+			}
+			ic := e.Value.(idleConn)
+			if ic.t.Add(timeout).After(nowFunc()) {
+				break
+			}
+			p.idle.Remove(e)
+			p.release()
+			p.mu.Unlock()
+			ic.c.Close()
+			p.mu.Lock()
+		}
+	}
+
+	for {
+
+		// Get idle connection.
+
+		for i, n := 0, p.idle.Len(); i < n; i++ {
+			e := p.idle.Front()
+			if e == nil {
+				break
+			}
+			ic := e.Value.(idleConn)
+			p.idle.Remove(e)
+			test := p.handler.TestOnBorrow
+			p.mu.Unlock()
+			if test == nil || test(ic.c, ic.t) == nil {
+				return ic.c, nil
+			}
+			ic.c.Close()
+			p.mu.Lock()
+			p.release()
+		}
+
+		// Check for pool closed before dialing a new connection.
+
+		if p.closed {
+			p.mu.Unlock()
+			return nil, errors.New("redigo: get on closed pool")
+		}
+
+		// Dial new connection if under limit.
+
+		if p.handler.GetMaxActive() == 0 || p.active < p.handler.GetMaxActive() {
+			dial := p.handler.Dial
+			p.active += 1
+			p.mu.Unlock()
+			c, err := dial()
+			if err != nil {
+				p.mu.Lock()
+				p.release()
+				p.mu.Unlock()
+				c = nil
+			}
+			return c, err
+		}
+
+		if !p.handler.IsWait() {
+			p.mu.Unlock()
+			return nil, ErrPoolExhausted
+		}
+
+		if p.cond == nil {
+			p.cond = sync.NewCond(&p.mu)
+		}
+		p.cond.Wait()
+	}
+}
+
+func (p *pool) put(c redis.Conn, forceClose bool) error {
+	err := c.Err()
+	p.mu.Lock()
+	if !p.closed && err == nil && !forceClose {
+		p.idle.PushFront(idleConn{t: nowFunc(), c: c})
+		if p.idle.Len() > p.handler.GetMaxIdle() {
+			c = p.idle.Remove(p.idle.Back()).(idleConn).c
+		} else {
+			c = nil
+		}
+	}
+
+	if c == nil {
+		if p.cond != nil {
+			p.cond.Signal()
+		}
+		p.mu.Unlock()
+		return nil
+	}
+
+	p.release()
+	p.mu.Unlock()
+	return c.Close()
+}
+
+type pooledConnection struct {
+	p     *pool
+	c     redis.Conn
+	state int
+}
+
+var (
+	sentinel     []byte
+	sentinelOnce sync.Once
+)
+
+func initSentinel() {
+	p := make([]byte, 64)
+	if _, err := rand.Read(p); err == nil {
+		sentinel = p
+	} else {
+		h := sha1.New()
+		io.WriteString(h, "Oops, rand failed. Use time instead.")
+		io.WriteString(h, strconv.FormatInt(time.Now().UnixNano(), 10))
+		sentinel = h.Sum(nil)
+	}
+}
+
+func (pc *pooledConnection) Close() error {
+	c := pc.c
+	if _, ok := c.(errorConnection); ok {
+		return nil
+	}
+	pc.c = errorConnection{errConnClosed}
+
+	if pc.state&internal.MultiState != 0 {
+		c.Send("DISCARD")
+		pc.state &^= (internal.MultiState | internal.WatchState)
+	} else if pc.state&internal.WatchState != 0 {
+		c.Send("UNWATCH")
+		pc.state &^= internal.WatchState
+	}
+	if pc.state&internal.SubscribeState != 0 {
+		c.Send("UNSUBSCRIBE")
+		c.Send("PUNSUBSCRIBE")
+		// To detect the end of the message stream, ask the server to echo
+		// a sentinel value and read until we see that value.
+		sentinelOnce.Do(initSentinel)
+		c.Send("ECHO", sentinel)
+		c.Flush()
+		for {
+			p, err := c.Receive()
+			if err != nil {
+				break
+			}
+			if p, ok := p.([]byte); ok && bytes.Equal(p, sentinel) {
+				pc.state &^= internal.SubscribeState
+				break
+			}
+		}
+	}
+	c.Do("")
+	pc.p.put(c, pc.state != 0)
+	return nil
+}
+
+func (pc *pooledConnection) Err() error {
+	return pc.c.Err()
+}
+
+func (pc *pooledConnection) Do(commandName string, args ...interface{}) (reply interface{}, err error) {
+	ci := internal.LookupCommandInfo(commandName)
+	pc.state = (pc.state | ci.Set) &^ ci.Clear
+	return pc.c.Do(commandName, args...)
+}
+
+func (pc *pooledConnection) Send(commandName string, args ...interface{}) error {
+	ci := internal.LookupCommandInfo(commandName)
+	pc.state = (pc.state | ci.Set) &^ ci.Clear
+	return pc.c.Send(commandName, args...)
+}
+
+func (pc *pooledConnection) Flush() error {
+	return pc.c.Flush()
+}
+
+func (pc *pooledConnection) Receive() (reply interface{}, err error) {
+	return pc.c.Receive()
+}
+
+type errorConnection struct{ err error }
+
+func (ec errorConnection) Do(string, ...interface{}) (interface{}, error) { return nil, ec.err }
+func (ec errorConnection) Send(string, ...interface{}) error              { return ec.err }
+func (ec errorConnection) Err() error                                     { return ec.err }
+func (ec errorConnection) Close() error                                   { return ec.err }
+func (ec errorConnection) Flush() error                                   { return ec.err }
+func (ec errorConnection) Receive() (interface{}, error)                  { return nil, ec.err }

pool_handler.go

@@ -0,0 +1,39 @@
+package overflow_redis_pool
+
+import (
+	"time"
+
+	"github.com/garyburd/redigo/redis"
+)
+
+type PoolHandler interface {
+	// Dial is an application supplied function for creating and configuring a
+	// connection.
+	//
+	// The connection returned from Dial must not be in a special state
+	// (subscribed to pubsub channel, transaction started, ...).
+	Dial() (redis.Conn, error)
+
+	// TestOnBorrow is an optional application supplied function for checking
+	// the health of an idle connection before the connection is used again by
+	// the application. Argument t is the time that the connection was returned
+	// to the pool. If the function returns an error, then the connection is
+	// closed.
+	TestOnBorrow(c redis.Conn, t time.Time) error
+
+	// Maximum number of idle connections in the pool.
+	GetMaxIdle() int
+
+	// Maximum number of connections allocated by the pool at a given time.
+	// When zero, there is no limit on the number of connections in the pool.
+	GetMaxActive() int
+
+	// Close connections after remaining idle for this duration. If the value
+	// is zero, then idle connections are not closed. Applications should set
+	// the timeout to a value less than the server's timeout.
+	GetIdleTimeout() time.Duration
+
+	// If Wait is true and the pool is at the MaxActive limit, then Get() waits
+	// for a connection to be returned to the pool before returning.
+	IsWait() bool
+}

pool_handlers.go

@@ -0,0 +1,49 @@
+package overflow_redis_pool
+
+import (
+	"time"
+
+	"github.com/garyburd/redigo/redis"
+)
+
+type PoolHandlers struct {
+	// Maximum number of idle connections in the pool.
+	MaxIdle int
+
+	// Maximum number of connections allocated by the pool at a given time.
+	// When zero, there is no limit on the number of connections in the pool.
+	MaxActive int
+
+	// Close connections after remaining idle for this duration. If the value
+	// is zero, then idle connections are not closed. Applications should set
+	// the timeout to a value less than the server's timeout.
+	IdleTimeout time.Duration
+
+	// If Wait is true and the pool is at the MaxActive limit, then Get() waits
+	// for a connection to be returned to the pool before returning.
+	Wait bool
+}
+
+func (h *PoolHandlers) Dial() (redis.Conn, error) {
+	return nil, nil
+}
+
+func (h *PoolHandlers) TestOnBorrow(c redis.Conn, t time.Time) error {
+	return nil
+}
+
+func (h *PoolHandlers) GetMaxIdle() int {
+	return h.MaxIdle
+}
+
+func (h *PoolHandlers) GetMaxActive() int {
+	return h.MaxActive
+}
+
+func (h *PoolHandlers) GetIdleTimeout() time.Duration {
+	return h.IdleTimeout
+}
+
+func (h *PoolHandlers) IsWait() bool {
+	return h.Wait
+}