package main
import (
        "context"
        "flag"
        "fmt"
        "log"
        "time"
        "mypoolclient/pool"
        "mypoolclient/pb"
)
var addr = flag.String("addr", "127.0.0.1:8888", "the address to connect to")
func main() {
        flag.Parse()
        p, err := pool.New(*addr, pool.DefaultOptions)
        if err != nil {
                log.Fatalf("failed to new pool: %v", err)
        }
        defer p.Close()
        conn, err := p.Get()
        if err != nil {
                log.Fatalf("failed to get conn: %v", err)
        }
        defer conn.Close()
        client := pb.NewTestsvrClient(conn.Value())
        ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
        defer cancel()
        res, err := client.Say(ctx, &pb.TestReq{Message: []byte("hi")})
        if err != nil {
                log.Fatalf("unexpected error from Say: %v", err)
        }
        fmt.Println("rpc response:", res)
}

package main
import (
        "context"
        "flag"
        "fmt"
        "log"
        "net"
        "google.golang.org/grpc"
        "mypoolserver/pb"
)
var port = flag.Int("port", 8888, "port number")
// server implements EchoServer.
type server struct{}
func (s *server) Say(context.Context, *pb.TestReq) (*pb.TestRsp, error) {
        fmt.Println("call  Say ... ")
        return &pb.TestRsp{Message: []byte("hello world")}, nil
}
func main() {
        flag.Parse()
        listen, err := net.Listen("tcp", fmt.Sprintf("127.0.0.1:%v", *port))
        if err != nil {
                log.Fatalf("failed to listen: %v", err)
        }
        s := grpc.NewServer()
        pb.RegisterTestsvrServer(s, &server{})
        fmt.Println("start server ...")
        if err := s.Serve(listen); err != nil {
                log.Fatalf("failed to serve: %v", err)
        }
        fmt.Println("over server ...")
}

type Pool interface {
   Get() (Conn, error)
   Close() error
   Status() string
}

func New(address string, option Options) (Pool, error) {
   if address == "" {
      return nil, errors.New("invalid address settings")
   }
   if option.Dial == nil {
      return nil, errors.New("invalid dial settings")
   }
   if option.MaxIdle <= 0 || option.MaxActive <= 0 || option.MaxIdle > option.MaxActive {
      return nil, errors.New("invalid maximum settings")
   }
   if option.MaxConcurrentStreams <= 0 {
      return nil, errors.New("invalid maximun settings")
   }
   p := &pool{
      index:   0,
      current: int32(option.MaxIdle),
      ref:     0,
      opt:     option,
      conns:   make([]*conn, option.MaxActive),
      address: address,
      closed:  0,
   }
   for i := 0; i < p.opt.MaxIdle; i++ {
      c, err := p.opt.Dial(address)
      if err != nil {
         p.Close()
         return nil, fmt.Errorf("dial is not able to fill the pool: %s", err)
      }
      p.conns[i] = p.wrapConn(c, false)
   }
   log.Printf("new pool success: %v\n", p.Status())
   return p, nil
}

// 關閉連接池
func (p *pool) Close() error {
   atomic.StoreInt32(&p.closed, 1)
   atomic.StoreUint32(&p.index, 0)
   atomic.StoreInt32(&p.current, 0)
   atomic.StoreInt32(&p.ref, 0)
   p.deleteFrom(0)
   log.Printf("close pool success: %v\n", p.Status())
   return nil
}
// 清除從 指定位置開始到 MaxActive 之間的連接
func (p *pool) deleteFrom(begin int) {
   for i := begin; i < p.opt.MaxActive; i++ {
      p.reset(i)
   }
}
// 清除具體的連接
func (p *pool) reset(index int) {
   conn := p.conns[index]
   if conn == nil {
      return
   }
   conn.reset()
   p.conns[index] = nil
}

func (c *conn) reset() error {
   cc := c.cc
   c.cc = nil
   c.once = false
   if cc != nil {
      return cc.Close()
   }
   return nil
}
func (c *conn) Close() error {
   c.pool.decrRef()
   if c.once {
      return c.reset()
   }
   return nil
}

func (p *pool) Get() (Conn, error) {
   // the first selected from the created connections
   nextRef := p.incrRef()
   p.RLock()
   current := atomic.LoadInt32(&p.current)
   p.RUnlock()
   if current == 0 {
      return nil, ErrClosed
   }
   if nextRef <= current*int32(p.opt.MaxConcurrentStreams) {
      next := atomic.AddUint32(&p.index, 1) % uint32(current)
      return p.conns[next], nil
   }
   // the number connection of pool is reach to max active
   if current == int32(p.opt.MaxActive) {
      // the second if reuse is true, select from pool's connections
      if p.opt.Reuse {
         next := atomic.AddUint32(&p.index, 1) % uint32(current)
         return p.conns[next], nil
      }
      // the third create one-time connection
      c, err := p.opt.Dial(p.address)
      return p.wrapConn(c, true), err
   }
   // the fourth create new connections given back to pool
   p.Lock()
   current = atomic.LoadInt32(&p.current)
   if current < int32(p.opt.MaxActive) && nextRef > current*int32(p.opt.MaxConcurrentStreams) {
      // 2 times the incremental or the remain incremental
      increment := current
      if current+increment > int32(p.opt.MaxActive) {
         increment = int32(p.opt.MaxActive) - current
      }
      var i int32
      var err error
      for i = 0; i < increment; i++ {
         c, er := p.opt.Dial(p.address)
         if er != nil {
            err = er
            break
         }
         p.reset(int(current + i))
         p.conns[current+i] = p.wrapConn(c, false)
      }
      current += i
      log.Printf("grow pool: %d ---> %d, increment: %d, maxActive: %d\n",
         p.current, current, increment, p.opt.MaxActive)
      atomic.StoreInt32(&p.current, current)
      if err != nil {
         p.Unlock()
         return nil, err
      }
   }
   p.Unlock()
   next := atomic.AddUint32(&p.index, 1) % uint32(current)
   return p.conns[next], nil
}