From bfaea09df9f146f13a536e08d30a606172aea830 Mon Sep 17 00:00:00 2001
From: dcorbacho <dparracorbacho@piotal.io>
Date: Mon, 1 Feb 2021 16:49:10 +0100
Subject: Implement `is_unresponsive` for stream queues

---
 deps/rabbit/src/rabbit_amqqueue.erl | 11 +++++++++++
 1 file changed, 11 insertions(+)

diff --git a/deps/rabbit/src/rabbit_amqqueue.erl b/deps/rabbit/src/rabbit_amqqueue.erl
index dccc725a9d..3704cf6ecc 100644
--- a/deps/rabbit/src/rabbit_amqqueue.erl
+++ b/deps/rabbit/src/rabbit_amqqueue.erl
@@ -1207,6 +1207,17 @@ is_unresponsive(Q, Timeout) when ?amqqueue_is_quorum(Q) ->
     catch
         exit:{timeout, _} ->
             true
+    end;
+is_unresponsive(Q, Timeout) when ?amqqueue_is_stream(Q) ->
+    try
+        #{leader_pid := LeaderPid} = amqqueue:get_type_state(Q),
+        case gen_batch_server:call(LeaderPid, get_reader_context, Timeout) of
+            #{dir := _} -> false;
+            _ -> true
+        end
+    catch
+        exit:{timeout, _} ->
+            true
     end.
 
 format(Q) when ?amqqueue_is_quorum(Q) -> rabbit_quorum_queue:format(Q);
-- 
cgit v1.2.1