src/server/pkg/pool/pool.go
func (p *Pool) Do(ctx context.Context, f func(cc *grpc.ClientConn) error) error {
	var conn *connCount
	if err := func() error {
		p.connsLock.Lock()
		defer p.connsLock.Unlock()
		for {
			for addr, mapConn := range p.conns {
				if mapConn.cc == nil {
					cc, err := grpc.DialContext(ctx, addr, p.opts...)
					if err != nil {
						return fmt.Errorf("failed to connect to %s: %+v", addr, err)
					}
					mapConn.cc = cc
					conn = mapConn
					// We break because this conn has a count of 0 which we know
					// we're not beating
					break
				} else {
					mapConnCount := atomic.LoadInt64(&mapConn.count)
					if mapConnCount < p.queueSize && (conn == nil || mapConnCount < atomic.LoadInt64(&conn.count)) {
						conn = mapConn
					}
				}
			}
			if conn == nil {
				p.connsCond.Wait()
			} else {
				atomic.AddInt64(&conn.count, 1)
				break
			}
		}
		return nil
	}(); err != nil {
		return err
	}
	defer p.connsCond.Broadcast()
	defer atomic.AddInt64(&conn.count, -1)
	return f(conn.cc)
}
func (p *Pool) Do(ctx context.Context, f func(cc *grpc.ClientConn) error) error {
	var conn *connCount
	if err := func() error {
		p.connsLock.Lock()
		defer p.connsLock.Unlock()
		for {
			for addr, mapConn := range p.conns {
				if mapConn.cc == nil {
					cc, err := grpc.DialContext(ctx, addr, p.opts...)
					if err != nil {
						return fmt.Errorf("failed to connect to %s: %+v", addr, err)
					}
					mapConn.cc = cc
					conn = mapConn
					// We break because this conn has a count of 0 which we know
					// we're not beating
					break
				} else {
					mapConnCount := atomic.LoadInt64(&mapConn.count)
					if mapConnCount < p.queueSize && (conn == nil || mapConnCount < atomic.LoadInt64(&conn.count)) {
						conn = mapConn
					}
				}
			}
			if conn == nil {
				p.connsCond.Wait()
			} else {
				atomic.AddInt64(&conn.count, 1)
				break
			}
		}
		return nil
	}(); err != nil {
		return err
	}
	defer p.connsCond.Broadcast()
	defer atomic.AddInt64(&conn.count, -1)
	return f(conn.cc)
}
func ( p * Pool ) Do ( ctx context . Context , f func ( cc * grpc . ClientConn ) error ) error { var conn * connCount if err := func ( ) error { p . connsLock . Lock ( ) defer p . connsLock . Unlock ( ) for { for addr , mapConn := range p . conns { if mapConn . cc == nil { cc , err := grpc . DialContext ( ctx , addr , p . opts ... ) if err != nil { return fmt . Errorf ( " " , addr , err ) } mapConn . cc = cc conn = mapConn // We break because this conn has a count of 0 which we know // we're not beating break } else { mapConnCount := atomic . LoadInt64 ( & mapConn . count ) if mapConnCount < p . queueSize && ( conn == nil || mapConnCount < atomic . LoadInt64 ( & conn . count ) ) { conn = mapConn } } } if conn == nil { p . connsCond . Wait ( ) } else { atomic . AddInt64 ( & conn . count , 1 ) break } } return nil } ( ) ; err != nil { return err } defer p . connsCond . Broadcast ( ) defer atomic . AddInt64 ( & conn . count , - 1 ) return f ( conn . cc ) }
// Do allows you to do something with a grpc.ClientConn.
// Errors returned from f will be returned by Do.
Do allows you to do something with a grpc . ClientConn . Errors returned from f will be returned by Do .