api: add Connection.CloseGraceful() and ConnectionPool.CloseGraceful()

CHANGELOG.md

@@ -10,6 +10,11 @@ Versioning](http://semver.org/spec/v2.0.0.html) except to the first release.
 
 ### Added
 
+- Connection.CloseGraceful() unlike Connection.Close() waits for all
+  requests to complete (#257)
+- ConnectionPool.CloseGraceful() unlike ConnectionPool.Close() waits for all
+  requests to complete (#257)
+
 ### Changed
 
 ### Fixed

connection.go

@@ -37,10 +37,10 @@ const (
 	Disconnected
 	// ReconnectFailed signals that attempt to reconnect has failed.
 	ReconnectFailed
-	// Either reconnect attempts exhausted, or explicit Close is called.
-	Closed
 	// Shutdown signals that shutdown callback is processing.
 	Shutdown
+	// Either reconnect attempts exhausted, or explicit Close is called.
+	Closed
 
 	// LogReconnectFailed is logged when reconnect attempt failed.
 	LogReconnectFailed ConnLogKind = iota + 1
@@ -109,6 +109,11 @@ func (d defaultLogger) Report(event ConnLogKind, conn *Connection, v ...interfac
 // - In "Disconnected" state it rejects queries with ClientError{Code:
 // ErrConnectionNotReady}
 //
+// - In "Shutdown" state it rejects queries with ClientError{Code:
+// ErrConnectionShutdown}. The state indicates that a graceful shutdown
+// in progress. The connection waits for all active requests to
+// complete.
+//
 // - In "Closed" state it rejects queries with ClientError{Code:
 // ErrConnectionClosed}. Connection could become "Closed" when
 // Connection.Close() method called, or when Tarantool disconnected and
@@ -457,6 +462,13 @@ func (conn *Connection) Close() error {
 	return conn.closeConnection(err, true)
 }
 
+// CloseGraceful closes Connection gracefully. It waits for all requests to
+// complete.
+// After this method called, there is no way to reopen this Connection.
+func (conn *Connection) CloseGraceful() error {
+	return conn.shutdown(true)
+}
+
 // Addr returns a configured address of Tarantool socket.
 func (conn *Connection) Addr() string {
 	return conn.addr
@@ -1527,17 +1539,27 @@ func shutdownEventCallback(event WatchEvent) {
 	// step 2.
 	val, ok := event.Value.(bool)
 	if ok && val {
-		go event.Conn.shutdown()
+		go event.Conn.shutdown(false)
 	}
 }
 
-func (conn *Connection) shutdown() {
+func (conn *Connection) shutdown(forever bool) error {
 	// Forbid state changes.
 	conn.mutex.Lock()
 	defer conn.mutex.Unlock()
 
 	if !atomic.CompareAndSwapUint32(&conn.state, connConnected, connShutdown) {
-		return
+		if forever {
+			err := ClientError{ErrConnectionClosed, "connection closed by client"}
+			return conn.closeConnection(err, true)
+		}
+		return nil
+	}
+
+	if forever {
+		// We don't want to reconnect any more.
+		conn.opts.Reconnect = 0
+		conn.opts.MaxReconnects = 0
 	}
 
 	conn.cond.Broadcast()
@@ -1546,7 +1568,7 @@ func (conn *Connection) shutdown() {
 	c := conn.c
 	for {
 		if (atomic.LoadUint32(&conn.state) != connShutdown) || (c != conn.c) {
-			return
+			return nil
 		}
 		if atomic.LoadInt64(&conn.requestCnt) == 0 {
 			break
@@ -1558,14 +1580,19 @@ func (conn *Connection) shutdown() {
 		conn.cond.Wait()
 	}
 
-	// Start to reconnect based on common rules, same as in net.box.
-	// Reconnect also closes the connection: server waits until all
-	// subscribed connections are terminated.
-	// See https://www.tarantool.io/en/doc/latest/dev_guide/internals/iproto/graceful_shutdown/
-	// step 3.
-	conn.reconnectImpl(
-		ClientError{
+	if forever {
+		err := ClientError{ErrConnectionClosed, "connection closed by client"}
+		return conn.closeConnection(err, true)
+	} else {
+		// Start to reconnect based on common rules, same as in net.box.
+		// Reconnect also closes the connection: server waits until all
+		// subscribed connections are terminated.
+		// See https://www.tarantool.io/en/doc/latest/dev_guide/internals/iproto/graceful_shutdown/
+		// step 3.
+		conn.reconnectImpl(ClientError{
 			ErrConnectionClosed,
 			"connection closed after server shutdown",
 		}, conn.c)
+		return nil
+	}
 }