Convert Session Moved Error and Closing Error to Connection Error

client.go

@@ -1009,7 +1009,7 @@ func (c *Client) handleNormalResponse(res responseHeader, buf []byte, blen int)
 	}
 
 	output := connNormal()
-	if res.Err == errSessionExpired {
+	if res.Err == errSessionExpired || res.Err == errSessionMoved || res.Err == errClosing {
 		err = ErrConnectionClosed
 		output = connError(err)
 	}

client_test.go

@@ -866,6 +866,74 @@ func TestClient_RecvData(t *testing.T) {
 		assert.Equal(t, ErrConnectionClosed, queue[0].err)
 	})
 
+	t.Run("receive session moved error", func(t *testing.T) {
+		c := newClientTest(t)
+		c.doAuthenticate()
+
+		c.client.enqueueRequest(
+			opGetData, &getDataRequest{}, &getDataResponse{},
+			nil,
+		)
+		c.client.getFromSendQueue()
+
+		buf := make([]byte, 2048)
+		n1, _ := encodePacket(buf[4:], &responseHeader{
+			Xid:  1,
+			Zxid: 73,
+			Err:  errSessionMoved,
+		})
+		binary.BigEndian.PutUint32(buf[:4], uint32(n1))
+
+		c.conn.readBuf.Write(buf[:4+n1])
+
+		output := c.client.readSingleData(c.conn)
+		assert.Equal(t, connIOOutput{
+			closed: false,
+			broken: true,
+			err:    ErrConnectionClosed,
+		}, output)
+
+		// Check Handle Queue
+		queue := c.client.handleQueue
+
+		assert.Equal(t, 1, len(queue))
+		assert.Equal(t, ErrConnectionClosed, queue[0].err)
+	})
+
+	t.Run("receive zookeeper is closing error", func(t *testing.T) {
+		c := newClientTest(t)
+		c.doAuthenticate()
+
+		c.client.enqueueRequest(
+			opGetData, &getDataRequest{}, &getDataResponse{},
+			nil,
+		)
+		c.client.getFromSendQueue()
+
+		buf := make([]byte, 2048)
+		n1, _ := encodePacket(buf[4:], &responseHeader{
+			Xid:  1,
+			Zxid: 73,
+			Err:  errClosing,
+		})
+		binary.BigEndian.PutUint32(buf[:4], uint32(n1))
+
+		c.conn.readBuf.Write(buf[:4+n1])
+
+		output := c.client.readSingleData(c.conn)
+		assert.Equal(t, connIOOutput{
+			closed: false,
+			broken: true,
+			err:    ErrConnectionClosed,
+		}, output)
+
+		// Check Handle Queue
+		queue := c.client.handleQueue
+
+		assert.Equal(t, 1, len(queue))
+		assert.Equal(t, ErrConnectionClosed, queue[0].err)
+	})
+
 	t.Run("read header len error", func(t *testing.T) {
 		c := newClientTest(t)
 

todolist

@@ -1,5 +1,3 @@
-*) Add Lint
-*) Convert errSessionMoved to Connection Error
 *) Batching Read & Write to TCP (Need to do or not?)
 *) Stress Tests with Race Detector
 *) Add Multi-Ops Transactions