API: Review DataChannel

Resolves #427

Author: backkem

datachannel.go

@@ -6,7 +6,6 @@ import (
 	"sync"
 
 	"github.com/pions/datachannel"
-	sugar "github.com/pions/webrtc/pkg/datachannel"
 	"github.com/pions/webrtc/pkg/rtcerr"
 	"github.com/pkg/errors"
 )
@@ -86,12 +85,10 @@ type DataChannel struct {
 	// "blob". This attribute controls how binary data is exposed to scripts.
 	// binaryType                 string
 
-	// OnOpen              func()
 	// OnBufferedAmountLow func()
 	// OnError             func()
-	// OnClose             func()
 
-	onMessageHandler func(sugar.Payload)
+	onMessageHandler func(DataChannelMessage)
 	onOpenHandler    func()
 	onCloseHandler   func()
 
@@ -265,35 +262,36 @@ func (d *DataChannel) onClose() (done chan struct{}) {
 	return
 }
 
-// OnMessage sets an event handler which is invoked on a message
-// arrival over the sctp transport from a remote peer.
+// DataChannelMessage represents a message received from the
+// data channel. IsString will be set to true if the incoming
+// message is of the string type. Otherwise the message is of
+// a binary type.
+type DataChannelMessage struct {
+	IsString bool
+	Data     []byte
+}
+
+// OnMessage sets an event handler which is invoked on a binary
+// message arrival over the sctp transport from a remote peer.
 // OnMessage can currently receive messages up to 16384 bytes
 // in size. Check out the detach API if you want to use larger
 // message sizes. Note that browser support for larger messages
 // is also limited.
-func (d *DataChannel) OnMessage(f func(p sugar.Payload)) {
+func (d *DataChannel) OnMessage(f func(msg DataChannelMessage)) {
 	d.mu.Lock()
 	defer d.mu.Unlock()
 	d.onMessageHandler = f
 }
 
-func (d *DataChannel) onMessage(p sugar.Payload) {
+func (d *DataChannel) onMessage(msg DataChannelMessage) {
 	d.mu.RLock()
 	hdlr := d.onMessageHandler
 	d.mu.RUnlock()
 
-	if hdlr == nil || p == nil {
+	if hdlr == nil {
 		return
 	}
-	hdlr(p)
-}
-
-// Onmessage sets an event handler which is invoked on a message
-// arrival over the sctp transport from a remote peer.
-//
-// Deprecated: use OnMessage instead.
-func (d *DataChannel) Onmessage(f func(p sugar.Payload)) {
-	d.OnMessage(f)
+	hdlr(msg)
 }
 
 func (d *DataChannel) handleOpen(dc *datachannel.DataChannel) {
@@ -331,39 +329,38 @@ func (d *DataChannel) readLoop() {
 			return
 		}
 
-		if isString {
-			d.onMessage(&sugar.PayloadString{Data: buffer[:n]})
-			continue
-		}
-		d.onMessage(&sugar.PayloadBinary{Data: buffer[:n]})
+		d.onMessage(DataChannelMessage{Data: buffer[:n], IsString: isString})
 	}
 }
 
-// Send sends the passed message to the DataChannel peer
-func (d *DataChannel) Send(payload sugar.Payload) error {
+// Send sends the binary message to the DataChannel peer
+func (d *DataChannel) Send(data []byte) error {
 	err := d.ensureOpen()
 	if err != nil {
 		return err
 	}
 
-	var data []byte
-	isString := false
+	if len(data) == 0 {
+		data = []byte{0}
+	}
 
-	switch p := payload.(type) {
-	case sugar.PayloadString:
-		data = p.Data
-		isString = true
-	case sugar.PayloadBinary:
-		data = p.Data
-	default:
-		return errors.Errorf("unknown DataChannel Payload (%s)", payload.PayloadType())
+	_, err = d.dataChannel.WriteDataChannel(data, false)
+	return err
+}
+
+// SendText sends the text message to the DataChannel peer
+func (d *DataChannel) SendText(s string) error {
+	err := d.ensureOpen()
+	if err != nil {
+		return err
 	}
 
+	data := []byte(s)
 	if len(data) == 0 {
 		data = []byte{0}
 	}
 
-	_, err = d.dataChannel.WriteDataChannel(data, isString)
+	_, err = d.dataChannel.WriteDataChannel(data, true)
 	return err
 }
 

datachannel_ortc_test.go

@@ -5,7 +5,6 @@ import (
 	"time"
 
 	"github.com/pions/transport/test"
-	"github.com/pions/webrtc/pkg/datachannel"
 )
 
 func TestDataChannel_ORTCE2E(t *testing.T) {
@@ -26,11 +25,11 @@ func TestDataChannel_ORTCE2E(t *testing.T) {
 	awaitBinary := make(chan struct{})
 	stackB.sctp.OnDataChannel(func(d *DataChannel) {
 		close(awaitSetup)
-		d.OnMessage(func(payload datachannel.Payload) {
-			switch payload.(type) {
-			case *datachannel.PayloadString:
+
+		d.OnMessage(func(msg DataChannelMessage) {
+			if msg.IsString {
 				close(awaitString)
-			case *datachannel.PayloadBinary:
+			} else {
 				close(awaitBinary)
 			}
 		})
@@ -52,11 +51,11 @@ func TestDataChannel_ORTCE2E(t *testing.T) {
 
 	<-awaitSetup
 
-	err = channelA.Send(datachannel.PayloadString{Data: []byte("ABC")})
+	err = channelA.SendText("ABC")
 	if err != nil {
 		t.Fatal(err)
 	}
-	err = channelA.Send(datachannel.PayloadBinary{Data: []byte("ABC")})
+	err = channelA.Send([]byte("ABC"))
 	if err != nil {
 		t.Fatal(err)
 	}

datachannel_test.go

@@ -8,8 +8,6 @@ import (
 	"testing"
 	"time"
 
-	sugar "github.com/pions/webrtc/pkg/datachannel"
-
 	"github.com/pions/transport/test"
 	"github.com/stretchr/testify/assert"
 )
@@ -85,20 +83,20 @@ func TestDataChannel_Send(t *testing.T) {
 	assert.True(t, dc.Ordered, "Ordered should be set to true")
 
 	dc.OnOpen(func() {
-		e := dc.Send(sugar.PayloadString{Data: []byte("Ping")})
+		e := dc.SendText("Ping")
 		if e != nil {
 			t.Fatalf("Failed to send string on data channel")
 		}
 	})
-	dc.OnMessage(func(payload sugar.Payload) {
+	dc.OnMessage(func(msg DataChannelMessage) {
 		done <- true
 	})
 
 	answerPC.OnDataChannel(func(d *DataChannel) {
 		assert.True(t, d.Ordered, "Ordered should be set to true")
 
-		d.OnMessage(func(payload sugar.Payload) {
-			e := d.Send(sugar.PayloadBinary{Data: []byte("Pong")})
+		d.OnMessage(func(msg DataChannelMessage) {
+			e := d.Send([]byte("Pong"))
 			if e != nil {
 				t.Fatalf("Failed to send string on data channel")
 			}
@@ -124,43 +122,27 @@ func TestDataChannel_EventHandlers(t *testing.T) {
 	api := NewAPI()
 	dc := &DataChannel{api: api}
 
-	onOpenCalled := make(chan bool)
-	onMessageCalled := make(chan bool)
+	onOpenCalled := make(chan struct{})
+	onMessageCalled := make(chan struct{})
 
 	// Verify that the noop case works
 	assert.NotPanics(t, func() { dc.onOpen() })
-	assert.NotPanics(t, func() { dc.onMessage(nil) })
 
 	dc.OnOpen(func() {
-		onOpenCalled <- true
+		close(onOpenCalled)
 	})
 
-	dc.OnMessage(func(p sugar.Payload) {
-		go func() {
-			onMessageCalled <- true
-		}()
+	dc.OnMessage(func(p DataChannelMessage) {
+		close(onMessageCalled)
 	})
 
-	// Verify that the handlers deal with nil inputs
-	assert.NotPanics(t, func() { dc.onMessage(nil) })
-
 	// Verify that the set handlers are called
 	assert.NotPanics(t, func() { dc.onOpen() })
-	assert.NotPanics(t, func() { dc.onMessage(&sugar.PayloadString{Data: []byte("o hai")}) })
+	assert.NotPanics(t, func() { dc.onMessage(DataChannelMessage{Data: []byte("o hai")}) })
 
-	allTrue := func(vals []bool) bool {
-		for _, val := range vals {
-			if !val {
-				return false
-			}
-		}
-		return true
-	}
-
-	assert.True(t, allTrue([]bool{
-		<-onOpenCalled,
-		<-onMessageCalled,
-	}))
+	// Wait for all handlers to be called
+	<-onOpenCalled
+	<-onMessageCalled
 }
 
 func TestDataChannel_MessagesAreOrdered(t *testing.T) {
@@ -172,34 +154,31 @@ func TestDataChannel_MessagesAreOrdered(t *testing.T) {
 
 	max := 512
 	out := make(chan int)
-	inner := func(payload sugar.Payload) {
+	inner := func(msg DataChannelMessage) {
 		// randomly sleep
 		// NB: The big.Int/crypto.Rand is overkill but makes the linter happy
 		randInt, err := rand.Int(rand.Reader, big.NewInt(int64(max)))
 		if err != nil {
 			t.Fatalf("Failed to get random sleep duration: %s", err)
 		}
 		time.Sleep(time.Duration(randInt.Int64()) * time.Microsecond)
-		p, ok := payload.(*sugar.PayloadBinary)
-		if ok {
-			s, _ := binary.Varint(p.Data)
-			out <- int(s)
-		}
+		s, _ := binary.Varint(msg.Data)
+		out <- int(s)
 	}
-	dc.OnMessage(func(p sugar.Payload) {
+	dc.OnMessage(func(p DataChannelMessage) {
 		inner(p)
 	})
 
 	go func() {
 		for i := 1; i <= max; i++ {
 			buf := make([]byte, 8)
 			binary.PutVarint(buf, int64(i))
-			dc.onMessage(&sugar.PayloadBinary{Data: buf})
+			dc.onMessage(DataChannelMessage{Data: buf})
 			// Change the registered handler a couple of times to make sure
 			// that everything continues to work, we don't lose messages, etc.
 			if i%2 == 0 {
-				hdlr := func(p sugar.Payload) {
-					inner(p)
+				hdlr := func(msg DataChannelMessage) {
+					inner(msg)
 				}
 				dc.OnMessage(hdlr)
 			}

examples/data-channels-close/main.go

@@ -6,7 +6,6 @@ import (
 	"time"
 
 	"github.com/pions/webrtc"
-	"github.com/pions/webrtc/pkg/datachannel"
 
 	"github.com/pions/webrtc/examples/internal/signal"
 )
@@ -56,9 +55,10 @@ func main() {
 			cnt := *closeAfter
 			for range ticker.C {
 				message := signal.RandSeq(15)
-				fmt.Printf("Sending %s \n", message)
+				fmt.Printf("Sending '%s'\n", message)
 
-				err := d.Send(datachannel.PayloadString{Data: []byte(message)})
+				// Send the message as text
+				err := d.SendText(message)
 				if err != nil {
 					panic(err)
 				}
@@ -76,15 +76,8 @@ func main() {
 		})
 
 		// Register message handling
-		d.OnMessage(func(payload datachannel.Payload) {
-			switch p := payload.(type) {
-			case *datachannel.PayloadString:
-				fmt.Printf("Message '%s' from DataChannel '%s' payload '%s'\n", p.PayloadType().String(), d.Label, string(p.Data))
-			case *datachannel.PayloadBinary:
-				fmt.Printf("Message '%s' from DataChannel '%s' payload '% 02x'\n", p.PayloadType().String(), d.Label, p.Data)
-			default:
-				fmt.Printf("Message '%s' from DataChannel '%s' no payload \n", p.PayloadType().String(), d.Label)
-			}
+		d.OnMessage(func(msg webrtc.DataChannelMessage) {
+			fmt.Printf("Message from DataChannel '%s': '%s'\n", d.Label, string(msg.Data))
 		})
 	})
 

examples/data-channels-create/main.go

@@ -5,7 +5,6 @@ import (
 	"time"
 
 	"github.com/pions/webrtc"
-	sugar "github.com/pions/webrtc/pkg/datachannel"
 
 	"github.com/pions/webrtc/examples/internal/signal"
 )
@@ -46,25 +45,19 @@ func main() {
 
 		for range time.NewTicker(5 * time.Second).C {
 			message := signal.RandSeq(15)
-			fmt.Printf("Sending %s \n", message)
+			fmt.Printf("Sending '%s'\n", message)
 
-			err := dataChannel.Send(sugar.PayloadString{Data: []byte(message)})
+			// Send the message as text
+			err := dataChannel.SendText(message)
 			if err != nil {
 				panic(err)
 			}
 		}
 	})
 
-	// Register the OnMessage to handle incoming messages
-	dataChannel.OnMessage(func(payload sugar.Payload) {
-		switch p := payload.(type) {
-		case *sugar.PayloadString:
-			fmt.Printf("Message '%s' from DataChannel '%s' payload '%s'\n", p.PayloadType().String(), dataChannel.Label, string(p.Data))
-		case *sugar.PayloadBinary:
-			fmt.Printf("Message '%s' from DataChannel '%s' payload '% 02x'\n", p.PayloadType().String(), dataChannel.Label, p.Data)
-		default:
-			fmt.Printf("Message '%s' from DataChannel '%s' no payload \n", p.PayloadType().String(), dataChannel.Label)
-		}
+	// Register text message handling
+	dataChannel.OnMessage(func(msg webrtc.DataChannelMessage) {
+		fmt.Printf("Message from DataChannel '%s': '%s'\n", dataChannel.Label, string(msg.Data))
 	})
 
 	// Create an offer to send to the browser