From 2ebb9927aee14f1777357d3af1858e114c53f7d4 Mon Sep 17 00:00:00 2001
From: chrislu <chris.lu@gmail.com>
Date: Tue, 25 Feb 2025 01:10:42 -0800
Subject: [PATCH] check cancelled request while waiting

---
 weed/mq/broker/broker_grpc_sub.go | 13 +++++++++++++
 1 file changed, 13 insertions(+)

diff --git a/weed/mq/broker/broker_grpc_sub.go b/weed/mq/broker/broker_grpc_sub.go
index fb8878b2d..a67898423 100644
--- a/weed/mq/broker/broker_grpc_sub.go
+++ b/weed/mq/broker/broker_grpc_sub.go
@@ -177,6 +177,19 @@ func (b *MessageQueueBroker) SubscribeMessage(stream mq_pb.SeaweedMessaging_Subs
 
 		for imt.IsInflight(logEntry.Key) {
 			time.Sleep(137 * time.Millisecond)
+			// Check if the client has disconnected by monitoring the context
+			select {
+			case <-ctx.Done():
+				err := ctx.Err()
+				if err == context.Canceled {
+					// Client disconnected
+					return false, nil
+				}
+				glog.V(0).Infof("Subscriber %s disconnected: %v", clientName, err)
+				return false, nil
+			default:
+				// Continue processing the request
+			}
 		}
 		if logEntry.Key != nil {
 			imt.EnflightMessage(logEntry.Key, logEntry.TsNs)