Added logic to reset the client SessionID if the client disconnects a…

…nd does not reconnect before the session timeout time elapses.

conn.go

@@ -420,7 +420,12 @@ func (c *Conn) sendRequest(
 	return rq.recvChan, nil
 }
 
+func (c *Conn) sessionExpired(d time.Time) bool {
+	return d.Add(time.Duration(c.sessionTimeoutMs) * time.Millisecond).Before(time.Now())
+}
+
 func (c *Conn) loop(ctx context.Context) {
+	var disconnectTime time.Time
 	for {
 		if err := c.connect(); err != nil {
 			// c.Close() was called
@@ -430,11 +435,15 @@ func (c *Conn) loop(ctx context.Context) {
 		err := c.authenticate()
 		switch {
 		case err == ErrSessionExpired:
-			c.logger.Printf("authentication failed: %s", err)
+			c.logger.Printf("authentication expired: %s", err)
+			c.resetSession()
 			c.invalidateWatches(err)
 		case err != nil && c.conn != nil:
 			c.logger.Printf("authentication failed: %s", err)
 			c.conn.Close()
+			if !disconnectTime.IsZero() && c.sessionExpired(disconnectTime) {
+				c.resetSession()
+			}
 		case err == nil:
 			if c.logInfo {
 				c.logger.Printf("authenticated: id=%d, timeout=%d", c.SessionID(), c.sessionTimeoutMs)
@@ -483,6 +492,7 @@ func (c *Conn) loop(ctx context.Context) {
 		}
 
 		c.setState(StateDisconnected)
+		disconnectTime = time.Now()
 
 		select {
 		case <-c.shouldQuit:
@@ -672,9 +682,6 @@ func (c *Conn) authenticate() error {
 		return err
 	}
 	if r.SessionID == 0 {
-		atomic.StoreInt64(&c.sessionID, int64(0))
-		c.passwd = emptyPassword
-		c.lastZxid = 0
 		c.setState(StateExpired)
 		return ErrSessionExpired
 	}
@@ -687,6 +694,12 @@ func (c *Conn) authenticate() error {
 	return nil
 }
 
+func (c *Conn) resetSession() {
+	atomic.StoreInt64(&c.sessionID, int64(0))
+	c.passwd = emptyPassword
+	c.lastZxid = 0
+}
+
 func (c *Conn) sendData(req *request) error {
 	header := &requestHeader{req.xid, req.opcode}
 	n, err := encodePacket(c.buf[4:], header)