packer/rpc: hard close all streams when underlying conn closes

packer/rpc/muxconn.go

@@ -234,7 +234,13 @@ func (m *MuxConn) openStream(id uint32) (*Stream, error) {
 }
 
 func (m *MuxConn) loop() {
-	defer m.Close()
+	defer func() {
+		m.mu.Lock()
+		defer m.mu.Unlock()
+		for _, w := range m.streams {
+			w.remoteClose()
+		}
+	}()
 
 	var id uint32
 	var packetType muxPacketType
@@ -384,16 +390,13 @@ func (s *Stream) Close() error {
 		return fmt.Errorf("Stream in bad state: %d", s.state)
 	}
 
-	if _, err := s.mux.write(s.id, muxPacketFin, nil); err != nil {
-		return err
-	}
-
 	if s.state == streamStateEstablished {
 		s.setState(streamStateFinWait1)
 	} else {
 		s.remoteClose()
 	}
 
+	s.mux.write(s.id, muxPacketFin, nil)
 	return nil
 }
 

packer/rpc/muxconn_test.go

@@ -114,6 +114,32 @@ func TestMuxConn(t *testing.T) {
 	<-doneCh
 }
 
+func TestMuxConn_socketClose(t *testing.T) {
+	client, server := testMux(t)
+	defer client.Close()
+	defer server.Close()
+
+	go func() {
+		_, err := server.Accept(0)
+		if err != nil {
+			t.Fatalf("err: %s", err)
+		}
+
+		server.rwc.Close()
+	}()
+
+	s0, err := client.Dial(0)
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	var data [1024]byte
+	_, err = s0.Read(data[:])
+	if err != io.EOF {
+		t.Fatalf("err: %s", err)
+	}
+}
+
 func TestMuxConn_clientClosesStreams(t *testing.T) {
 	client, server := testMux(t)
 	defer client.Close()