Handle closing frame properly in datagramBuffer

internal/multiplex/datagramBuffer.go

@@ -64,7 +64,13 @@ func (d *datagramBuffer) Write(f Frame) error {
 		}
 		d.rwCond.Wait()
 	}
-	// TODO: deal with closing frame here
+
+	if f.Closing == 1 {
+		d.closed = true
+		d.rwCond.Broadcast()
+		return nil
+	}
+
 	data := make([]byte, len(f.Payload))
 	copy(data, f.Payload)
 	d.buf = append(d.buf, data)

internal/multiplex/datagramBuffer_test.go

@@ -7,48 +7,66 @@ import (
 )
 
 func TestDatagramBuffer_RW(t *testing.T) {
-	pipe := NewDatagramBuffer()
 	b := []byte{0x01, 0x02, 0x03}
-	err := pipe.Write(Frame{Payload: b})
+	t.Run("simple write", func(t *testing.T) {
-	if err != nil {
+		pipe := NewDatagramBuffer()
-		t.Error(
+		err := pipe.Write(Frame{Payload: b})
-			"For", "simple write",
+		if err != nil {
-			"expecting", "nil error",
+			t.Error(
-			"got", err,
+				"expecting", "nil error",
-		)
+				"got", err,
-		return
+			)
-	}
+			return
+		}
+	})
 
-	b2 := make([]byte, len(b))
+	t.Run("simple read", func(t *testing.T) {
-	n, err := pipe.Read(b2)
+		pipe := NewDatagramBuffer()
-	if n != len(b) {
+		_ = pipe.Write(Frame{Payload: b})
-		t.Error(
+		b2 := make([]byte, len(b))
-			"For", "number of bytes read",
+		n, err := pipe.Read(b2)
-			"expecting", len(b),
+		if n != len(b) {
-			"got", n,
+			t.Error(
-		)
+				"For", "number of bytes read",
-		return
+				"expecting", len(b),
-	}
+				"got", n,
-	if err != nil {
+			)
-		t.Error(
+			return
-			"For", "simple read",
+		}
-			"expecting", "nil error",
+		if err != nil {
-			"got", err,
+			t.Error(
-		)
+				"expecting", "nil error",
-		return
+				"got", err,
-	}
+			)
-	if !bytes.Equal(b, b2) {
+			return
-		t.Error(
+		}
-			"For", "simple read",
+		if !bytes.Equal(b, b2) {
-			"expecting", b,
+			t.Error(
-			"got", b2,
+				"expecting", b,
-		)
+				"got", b2,
-	}
+			)
-	if len(pipe.buf) != 0 {
+		}
-		t.Error("buf len is not 0 after finished reading")
+		if len(pipe.buf) != 0 {
-		return
+			t.Error("buf len is not 0 after finished reading")
-	}
+			return
+		}
 
+	})
+
+	t.Run("writing closing frame", func(t *testing.T) {
+		pipe := NewDatagramBuffer()
+		err := pipe.Write(Frame{Closing: 1})
+		if err != nil {
+			t.Error(
+				"expecting", "nil error",
+				"got", err,
+			)
+			return
+		}
+		if !pipe.closed {
+			t.Error("expecting closed pipe, not closed")
+		}
+	})
 }
 
 func TestDatagramBuffer_BlockingRead(t *testing.T) {