streaming: Listen on shutdown gossip callback

When a node shutdown itself, it will send a shutdown status to peer
nodes. When peer nodes receives the shtudown status update, they are
supposed to close all the sessions with that node becasue the node is
shutdown, no need to wait and timeout, then fail the session.

This change can speed up the closing of sessions.

streaming/stream_manager.cc

@@ -291,4 +291,15 @@ void stream_manager::on_restart(inet_address endpoint, endpoint_state ep_state)
     }
 }
 
+void stream_manager::on_dead(inet_address endpoint, endpoint_state ep_state) {
+    if (has_peer(endpoint) && ep_state.is_shutdown()) {
+        sslog.info("stream_manager: Close all stream_session with peer = {} in on_dead", endpoint);
+        get_stream_manager().invoke_on_all([endpoint] (auto& sm) {
+            sm.fail_sessions(endpoint);
+        }).handle_exception([endpoint] (auto ep) {
+            sslog.warn("stream_manager: Fail to close sessions peer = {} in on_dead", endpoint);
+        });
+    }
+}
+
 } // namespace streaming

streaming/stream_manager.hh

@@ -156,7 +156,7 @@ public:
     virtual void before_change(inet_address endpoint, endpoint_state current_state, application_state new_state_key, const versioned_value& new_value) override {}
     virtual void on_change(inet_address endpoint, application_state state, const versioned_value& value) override {}
     virtual void on_alive(inet_address endpoint, endpoint_state state) override {}
-    virtual void on_dead(inet_address endpoint, endpoint_state state) override {}
+    virtual void on_dead(inet_address endpoint, endpoint_state state) override;
     virtual void on_remove(inet_address endpoint) override;
     virtual void on_restart(inet_address endpoint, endpoint_state ep_state) override;