Remove unnecessary Len check

internal/multiplex/datagramBuffer.go

@@ -80,9 +80,3 @@ func (d *datagramBuffer) Close() error {
 	d.rwCond.Broadcast()
 	return nil
 }
-
-func (d *datagramBuffer) Len() int {
-	d.rwCond.L.Lock()
-	defer d.rwCond.L.Unlock()
-	return len(d.buf)
-}

internal/multiplex/datagramBuffer_test.go

@@ -44,7 +44,7 @@ func TestDatagramBuffer_RW(t *testing.T) {
 			"got", b2,
 		)
 	}
-	if pipe.Len() != 0 {
+	if len(pipe.buf) != 0 {
 		t.Error("buf len is not 0 after finished reading")
 		return
 	}

internal/multiplex/recvBuffer.go

@@ -5,5 +5,4 @@ import "io"
 type recvBuffer interface {
 	io.ReadCloser
 	Write(Frame) error
-	Len() int
 }

internal/multiplex/stream.go

@@ -15,13 +15,6 @@ import (
 
 var ErrBrokenStream = errors.New("broken stream")
 
-// ReadWriteCloseLener is io.ReadWriteCloser with Len()
-// used for bufferedPipe and datagramBuffer
-type ReadWriteCloseLener interface {
-	io.ReadWriteCloser
-	Len() int
-}
-
 type Stream struct {
 	id uint32
 
@@ -83,20 +76,13 @@ func (s *Stream) Read(buf []byte) (n int, err error) {
 		}
 	}
 
-	if s.isClosed() {
-		// TODO: Len check may not be necessary as this can be offloaded to buffer implementation
-		if s.recvBuf.Len() == 0 {
-			return 0, ErrBrokenStream
-		} else {
-			n, err = s.recvBuf.Read(buf)
-			log.Tracef("%v read from stream %v with err %v", n, s.id, err)
-			return
-		}
-	} else {
-		n, err = s.recvBuf.Read(buf)
-		log.Tracef("%v read from stream %v with err %v", n, s.id, err)
-		return
+	n, err = s.recvBuf.Read(buf)
+	if err == io.EOF {
+		return n, ErrBrokenStream
 	}
+	log.Tracef("%v read from stream %v with err %v", n, s.id, err)
+	return
+
 }
 
 // Write implements io.Write

internal/multiplex/streamBuffer.go

@@ -107,7 +107,3 @@ func (sb *streamBuffer) Read(buf []byte) (int, error) {
 func (sb *streamBuffer) Close() error {
 	return sb.buf.Close()
 }
-
-func (sb *streamBuffer) Len() int {
-	return sb.buf.Len()
-}