shrpx: Send RST_STREAM when downstream becomes stale

src/shrpx_spdy_downstream_connection.cc

@@ -59,7 +59,11 @@ SpdyDownstreamConnection::~SpdyDownstreamConnection()
   if(request_body_buf_) {
     evbuffer_free(request_body_buf_);
   }
-  // TODO need RST_STREAM?
+  if(downstream_) {
+    if(submit_rst_stream(downstream_) == 0) {
+      spdy_->notify();
+    }
+  }
   spdy_->remove_downstream_connection(this);
   // Downstream and DownstreamConnection may be deleted
   // asynchronously.
@@ -110,14 +114,37 @@ void SpdyDownstreamConnection::detach_downstream(Downstream *downstream)
   if(LOG_ENABLED(INFO)) {
     DCLOG(INFO, this) << "Detaching from DOWNSTREAM:" << downstream;
   }
+  if(submit_rst_stream(downstream) == 0) {
+    spdy_->notify();
+  }
   downstream->set_downstream_connection(0);
   downstream_ = 0;
 
-  // TODO do something to SpdySession? RST_STREAM?
-
   client_handler_->pool_downstream_connection(this);
 }
 
+int SpdyDownstreamConnection::submit_rst_stream(Downstream *downstream)
+{
+  int rv = -1;
+  if(spdy_->get_state() == SpdySession::CONNECTED &&
+     downstream->get_downstream_stream_id() != -1) {
+    switch(downstream->get_response_state()) {
+    case Downstream::MSG_RESET:
+    case Downstream::MSG_COMPLETE:
+      break;
+    default:
+      if(LOG_ENABLED(INFO)) {
+        DCLOG(INFO, this) << "Submit RST_STREAM for DOWNSTREAM:"
+                          << downstream;
+      }
+      rv = spdy_->submit_rst_stream(this,
+                                    downstream->get_downstream_stream_id(),
+                                    SPDYLAY_CANCEL);
+    }
+  }
+  return rv;
+}
+
 namespace {
 ssize_t spdy_data_read_callback(spdylay_session *session,
                                 int32_t stream_id,

src/shrpx_spdy_downstream_connection.h

@@ -66,6 +66,8 @@ public:
   void attach_stream_data(StreamData *sd);
   StreamData* detach_stream_data();
 
+  int submit_rst_stream(Downstream *downstream);
+
   int32_t get_recv_window_size() const;
   void inc_recv_window_size(int32_t amount);
 private: