func (this *pubPool) buildPools() {
	// idleTimeout=0 means each kafka conn will last forever
	this.syncPool = pool.NewResourcePool(this.syncProducerFactory,
		this.size, this.size, 0)
	this.syncAllPool = pool.NewResourcePool(this.syncAllProducerFactory,
		30, 30, 0) // should be enough TODO
	this.asyncPool = pool.NewResourcePool(this.asyncProducerFactory,
		this.size, this.size, 0)
}

// Open must be call before starting to use the pool.
func (cp *ConnectionPool) Open(connFactory CreateConnectionFunc) {
	cp.mu.Lock()
	defer cp.mu.Unlock()
	f := func() (pools.Resource, error) {
		return connFactory(cp)
	}
	cp.connections = pools.NewResourcePool(f, cp.capacity, cp.capacity, cp.idleTimeout)
}

func newPool(server string) *pools.ResourcePool {
	f := func() (pools.Resource, error) {
		c, err := redis.Dial("tcp", server)
		return &resourceConn{c}, err
	}

	cap := redisConnParams()

	return pools.NewResourcePool(f, cap, redisMaxCap, time.Minute)
}

// Open must be called before starting to use the pool.
func (cp *ConnPool) Open(appParams, dbaParams *sqldb.ConnParams) {
	cp.mu.Lock()
	defer cp.mu.Unlock()

	f := func() (pools.Resource, error) {
		return NewDBConn(cp, appParams, dbaParams, cp.queryServiceStats)
	}
	cp.connections = pools.NewResourcePool(f, cp.capacity, cp.capacity, cp.idleTimeout)
	cp.dbaPool.Open(dbconnpool.DBConnectionCreator(dbaParams, cp.queryServiceStats.MySQLStats))
}

func (cp *ConnectionPool) Open(connFactory CreateConnectionFunc) {
	f := func() (pools.Resource, error) {
		c, err := connFactory()
		if err != nil {
			return nil, err
		}
		return &pooledConnection{c, cp}, nil
	}
	cp.pool = pools.NewResourcePool(f, cp.capacity, cp.capacity, cp.idleTimeout)
}

func (cp *CachePool) Open(connFactory CreateCacheFunc) {
	if connFactory == nil {
		return
	}
	f := func() (pools.Resource, error) {
		c, err := connFactory()
		if err != nil {
			return nil, err
		}
		return &Cache{c, cp}, nil
	}
	cp.pool = pools.NewResourcePool(f, cp.capacity, cp.capacity, cp.idleTimeout)
}

func (cp *CachePool) Open() {
	if len(cp.commandLine) == 0 {
		log.Infof("rowcache not enabled")
		return
	}
	cp.startMemcache()
	log.Infof("rowcache is enabled")
	f := func() (pools.Resource, error) {
		c, err := memcache.Connect(cp.port)
		if err != nil {
			return nil, err
		}
		return &Cache{c, cp}, nil
	}
	cp.pool = pools.NewResourcePool(f, cp.capacity, cp.capacity, cp.idleTimeout)
}

func (cp *CachePool) Open() {
	if cp.rowCacheConfig.Binary == "" {
		panic(NewTabletError(FATAL, "rowcache binary not specified"))
	}
	cp.startMemcache()
	log.Infof("rowcache is enabled")
	f := func() (pools.Resource, error) {
		return memcache.Connect(cp.port, 10*time.Second)
	}
	cp.mu.Lock()
	defer cp.mu.Unlock()
	cp.pool = pools.NewResourcePool(f, cp.capacity, cp.capacity, cp.idleTimeout)
	if cp.memcacheStats != nil {
		cp.memcacheStats.Open()
	}
}

func (cp *CachePool) Open() {
	if cp.rowCacheConfig.Binary == "" {
		log.Infof("rowcache not enabled")
		return
	}
	cp.startMemcache()
	log.Infof("rowcache is enabled")
	f := func() (pools.Resource, error) {
		c, err := memcache.Connect(cp.port)
		if err != nil {
			return nil, err
		}
		return &Cache{c, cp}, nil
	}
	cp.mu.Lock()
	defer cp.mu.Unlock()
	cp.pool = pools.NewResourcePool(f, cp.capacity, cp.capacity, cp.idleTimeout)
	if cp.memcacheStats != nil {
		cp.memcacheStats.Open()
	}
}

// Open opens the pool. It launches memcache and waits till it's up.
func (cp *CachePool) Open() {
	cp.mu.Lock()
	defer cp.mu.Unlock()
	if cp.pool != nil {
		panic(NewTabletError(ErrFatal, vtrpc.ErrorCode_INTERNAL_ERROR, "rowcache is already open"))
	}
	if cp.rowCacheConfig.Binary == "" {
		panic(NewTabletError(ErrFatal, vtrpc.ErrorCode_INTERNAL_ERROR, "rowcache binary not specified"))
	}
	cp.socket = generateFilename(cp.rowCacheConfig.Socket)
	cp.startCacheService()
	log.Infof("rowcache is enabled")
	f := func() (pools.Resource, error) {
		return cacheservice.Connect(cacheservice.Config{
			Address: cp.socket,
			Timeout: 10 * time.Second,
		})
	}
	cp.pool = pools.NewResourcePool(f, cp.capacity, cp.capacity, cp.idleTimeout)
	if cp.memcacheStats != nil {
		cp.memcacheStats.Open()
	}
}

func newRedisPool(uri string, capacity int, maxCapacity int, idleTimout time.Duration) *pools.ResourcePool {
	return pools.NewResourcePool(newRedisFactory(uri), capacity, maxCapacity, idleTimout)
}

// NewPool returns a new short service client pool
func NewPool(addr AddressFunc, capacity, maxCap int, connectionTimeout, idleTimeout time.Duration) (*Pool, error) {
	p := pools.NewResourcePool(createFactoryFunc(addr, connectionTimeout), capacity, maxCap, idleTimeout)

	return &Pool{p}, nil
}