Implement deadlines for mux

internal/mux/endpoint.go

@@ -80,14 +80,16 @@ func (e *Endpoint) RemoteAddr() net.Addr {
 	return e.mux.nextConn.RemoteAddr()
 }
 
-// SetDeadline is a stub.
-func (e *Endpoint) SetDeadline(time.Time) error {
-	return nil
+// SetDeadline sets the read deadline for this Endpoint.
+// Write deadlines are not supported because writes go directly to the shared
+// underlying connection and are non-blocking for this endpoint.
+func (e *Endpoint) SetDeadline(t time.Time) error {
+	return e.buffer.SetReadDeadline(t)
 }
 
-// SetReadDeadline is a stub.
-func (e *Endpoint) SetReadDeadline(time.Time) error {
-	return nil
+// SetReadDeadline sets the read deadline for this Endpoint.
+func (e *Endpoint) SetReadDeadline(t time.Time) error {
+	return e.buffer.SetReadDeadline(t)
 }
 
 // SetWriteDeadline is a stub.

internal/mux/mux_test.go

@@ -32,6 +32,52 @@ func TestNoEndpoints(t *testing.T) {
 	require.NoError(t, ca.Close())
 }
 
+func TestEndpointDeadline(t *testing.T) {
+	tests := []struct {
+		name        string
+		setDeadline func(*Endpoint, time.Time) error
+	}{
+		{
+			name:        "SetReadDeadline",
+			setDeadline: (*Endpoint).SetReadDeadline,
+		},
+		{
+			name:        "SetDeadline",
+			setDeadline: (*Endpoint).SetDeadline,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			lim := test.TimeOut(2 * time.Second)
+			defer lim.Stop()
+
+			ca, cb := net.Pipe()
+			defer func() {
+				_ = ca.Close()
+				_ = cb.Close()
+			}()
+
+			mux := NewMux(Config{
+				Conn:          ca,
+				BufferSize:    testPipeBufferSize,
+				LoggerFactory: logging.NewDefaultLoggerFactory(),
+			})
+
+			endpoint := mux.NewEndpoint(MatchAll)
+			require.NoError(t, tt.setDeadline(endpoint, time.Now().Add(10*time.Millisecond)))
+
+			_, err := endpoint.Read(make([]byte, testPipeBufferSize))
+			require.Error(t, err)
+			var netErr interface{ Timeout() bool }
+			require.ErrorAs(t, err, &netErr)
+			require.True(t, netErr.Timeout())
+
+			require.NoError(t, mux.Close())
+		})
+	}
+}
+
 type muxErrorConnReadResult struct {
 	err  error
 	data []byte