base test

p2p/transport/webrtc_w3c/handlers.go

@@ -101,10 +101,12 @@ func handleIncoming(ctx context.Context, config webrtc.Configuration, stream net
 		log.Warn("failed to read SDP offer: %v", err)
 		return nil, err
 	}
-	if msg.Type != pb.Message_SDP_OFFER.Enum() {
+	if msg.Type == nil || msg.GetType() != pb.Message_SDP_OFFER {
+		log.Warn("expected SDP offer, instead got: %v", msg.GetType())
 		return nil, errExpectedOffer
 	}
 	if msg.Data == nil {
+		log.Warn("message is empty")
 		return nil, errEmptyData
 	}
 	offer := webrtc.SessionDescription{
@@ -154,24 +156,27 @@ func handleIncoming(ctx context.Context, config webrtc.Configuration, stream net
 				readErr <- err
 				return
 			}
-			if msg.Type != pb.Message_ICE_CANDIDATE.Enum() {
+			if msg.Type == nil || msg.GetType() != pb.Message_ICE_CANDIDATE {
 				readErr <- errors.New("got non-candidate message")
 				return
 			}
 			if msg.Data == nil {
 				readErr <- errEmptyData
 				return
 			}
+			if *msg.Data == "" {
+				return
+			}
 
 			// unmarshal IceCandidateInit
 			var init webrtc.ICECandidateInit
 			if err := json.Unmarshal([]byte(*msg.Data), &init); err != nil {
-				log.Warn("could not unmarshal candidate: %v", err)
+				log.Debugf("could not unmarshal candidate: %v, %s", err, *msg.Data)
 				readErr <- err
 				return
 			}
 			if err := pc.AddICECandidate(init); err != nil {
-				log.Warn("bad candidate: %v", err)
+				log.Debugf("bad candidate: %v", err)
 				readErr <- err
 				return
 			}
@@ -189,8 +194,9 @@ func handleIncoming(ctx context.Context, config webrtc.Configuration, stream net
 		}
 		break
 	case err := <-readErr:
-		if err != nil {
-			panic("nil error should never be written to this channel")
+		if err == nil {
+			log.Error("err: %v", err)
+			panic("nil error should never be written to this channel %v")
 		}
 		_ = pc.Close()
 		return nil, err
@@ -301,14 +307,14 @@ func connect(ctx context.Context, config webrtc.Configuration, stream network.St
 		log.Warn("failed to read SDP answer: %v", err)
 		return nil, err
 	}
-	if msg.Type != pb.Message_SDP_ANSWER.Enum() {
+	if msg.Type == nil || msg.GetType() != pb.Message_SDP_ANSWER {
 		return nil, errExpectedAnswer
 	}
 	if msg.Data == nil {
 		return nil, errEmptyData
 	}
 	answer := webrtc.SessionDescription{
-		Type: webrtc.SDPTypeOffer,
+		Type: webrtc.SDPTypeAnswer,
 		SDP:  *msg.Data,
 	}
 	if err := pc.SetRemoteDescription(answer); err != nil {
@@ -334,7 +340,7 @@ func connect(ctx context.Context, config webrtc.Configuration, stream network.St
 				readErr <- err
 				return
 			}
-			if msg.Type != pb.Message_ICE_CANDIDATE.Enum() {
+			if msg.Type == nil || msg.GetType() != pb.Message_ICE_CANDIDATE {
 				readErr <- errors.New("got non-candidate message")
 				return
 			}
@@ -345,13 +351,14 @@ func connect(ctx context.Context, config webrtc.Configuration, stream network.St
 
 			// unmarshal IceCandidateInit
 			var init webrtc.ICECandidateInit
+			if *msg.Data == "" {
+				return
+			}
 			if err := json.Unmarshal([]byte(*msg.Data), &init); err != nil {
-				log.Warn("could not unmarshal candidate: %v", err)
 				readErr <- err
 				return
 			}
 			if err := pc.AddICECandidate(init); err != nil {
-				log.Warn("bad candidate: %v", err)
 				readErr <- err
 				return
 			}
@@ -369,9 +376,10 @@ func connect(ctx context.Context, config webrtc.Configuration, stream network.St
 		}
 		break
 	case err := <-readErr:
-		if err != nil {
+		if err == nil {
 			panic("nil error should never be written to this channel")
 		}
+		log.Warn("error: %v", err)
 		_ = pc.Close()
 		return nil, err
 	}

p2p/transport/webrtc_w3c/handlers_test.go

@@ -0,0 +1,117 @@
+package webrtc_w3c
+
+import (
+	"context"
+	"io"
+	"testing"
+	"time"
+
+	"github.com/libp2p/go-libp2p/core/network"
+	"github.com/libp2p/go-libp2p/core/protocol"
+	"github.com/pion/webrtc/v3"
+	"github.com/stretchr/testify/require"
+)
+
+var _ network.Stream = &stream{}
+
+type stream struct {
+	r io.Reader
+	w io.Writer
+}
+
+// Read implements network.Stream
+func (s *stream) Read(p []byte) (n int, err error) {
+	return s.r.Read(p)
+}
+
+// Write implements network.Stream
+func (s *stream) Write(p []byte) (n int, err error) {
+	return s.w.Write(p)
+}
+
+// Close implements network.Stream
+func (s *stream) Close() error {
+	return nil
+}
+
+// CloseRead implements network.Stream
+func (*stream) CloseRead() error {
+	return nil
+}
+
+// CloseWrite implements network.Stream
+func (*stream) CloseWrite() error {
+	return nil
+}
+
+// Reset implements network.Stream
+func (*stream) Reset() error {
+	return nil
+}
+
+// SetDeadline implements network.Stream
+func (*stream) SetDeadline(time.Time) error {
+	return nil
+}
+
+// SetReadDeadline implements network.Stream
+func (*stream) SetReadDeadline(time.Time) error {
+	return nil
+}
+
+// SetWriteDeadline implements network.Stream
+func (*stream) SetWriteDeadline(time.Time) error {
+	return nil
+}
+
+// Conn implements network.Stream
+func (*stream) Conn() network.Conn {
+	return nil
+}
+
+// ID implements network.Stream
+func (*stream) ID() string {
+	return ""
+}
+
+// Protocol implements network.Stream
+func (*stream) Protocol() protocol.ID {
+	return ""
+}
+
+// Scope implements network.Stream
+func (*stream) Scope() network.StreamScope {
+	panic("unimplemented")
+}
+
+// SetProtocol implements network.Stream
+func (*stream) SetProtocol(id protocol.ID) error {
+	panic("unimplemented")
+}
+
+// Stat implements network.Stream
+func (*stream) Stat() network.Stats {
+	panic("unimplemented")
+}
+
+func makeStreamPair() (network.Stream, network.Stream) {
+	ra, wb := io.Pipe()
+	rb, wa := io.Pipe()
+	return &stream{ra, wa}, &stream{rb, wb}
+}
+
+func TestWebrtcW3Handlers_ShouldConnect(t *testing.T) {
+	sa, sb := makeStreamPair()
+	errC := make(chan error)
+	go func() {
+		defer close(errC)
+		_, err := handleIncoming(context.Background(), webrtc.Configuration{}, sa)
+		if err != nil {
+			errC <- err
+		}
+	}()
+
+	_, err := connect(context.Background(), webrtc.Configuration{}, sb)
+	require.NoError(t, err)
+	require.NoError(t, <-errC)
+}