protocol: add webrtc support

Author: cihangir

client.go

@@ -711,6 +711,16 @@ func (c *Client) Tell(method string, args ...interface{}) (result *dnode.Partial
 	return c.TellWithTimeout(method, 0, args...)
 }
 
+// SendWebRTCRequest sends requests to kontrol for signalling purposes.
+func (c *Client) SendWebRTCRequest(req *protocol.WebRTCSignalMessage) error {
+	timeout := time.Duration(0)
+	if c.Config != nil {
+		timeout = c.Config.Timeout
+	}
+	_, err := c.TellWithTimeout(WebRTCHandlerName, timeout, req)
+	return err
+}
+
 // TellWithTimeout does the same thing with Tell() method except it takes an
 // extra argument that is the timeout for waiting reply from the remote Kite.
 // If timeout is given 0, the behavior is same as Tell().

config/config.go

@@ -19,8 +19,8 @@ import (
 	"github.com/igm/sockjs-go/sockjs"
 )
 
-// the implementation of New() doesn't have any error to be returned yet it
-// returns, so it's totally safe to neglect the error
+// CookieJar ignoring err: the implementation of New() doesn't have any error to
+// be returned yet it returns, so it's totally safe to neglect the error
 var CookieJar, _ = cookiejar.New(nil)
 
 // Options is passed to kite.New when creating new instance.
@@ -110,6 +110,9 @@ type Config struct {
 	KontrolURL  string
 	KontrolKey  string
 	KontrolUser string
+
+	// UseWebRTC is the flag for Kite's to communicate over WebRTC if possible.
+	UseWebRTC bool
 }
 
 // DefaultConfig contains the default settings.

handlers.go

@@ -1,19 +1,85 @@
 package kite
 
 import (
+	"errors"
 	"fmt"
 	"net/http"
 	"net/url"
 	"os"
 	"os/exec"
 	"runtime"
+	"time"
 
 	"github.com/gorilla/websocket"
+	"github.com/koding/cache"
+	"github.com/koding/kite/protocol"
 	"github.com/koding/kite/sockjsclient"
 	"github.com/koding/kite/systeminfo"
 	"golang.org/x/crypto/ssh/terminal"
 )
 
+var (
+	errDstNotSet        = errors.New("dst not set")
+	errDstNotRegistered = errors.New("dst not registered")
+)
+
+// WebRTCHandlerName provides the naming scheme for the handler
+const WebRTCHandlerName = "kite.handleWebRTC"
+
+type webRTCHandler struct {
+	kitesColl cache.Cache
+}
+
+// NewWebRCTHandler creates a new handler for web rtc signalling services.
+func NewWebRCTHandler() *webRTCHandler {
+	return &webRTCHandler{
+		kitesColl: cache.NewMemory(),
+	}
+}
+
+func (w *webRTCHandler) registerSrc(src *Client) {
+	w.kitesColl.Set(src.ID, src)
+	src.OnDisconnect(func() {
+		time.Sleep(time.Second * 2)
+		id := src.ID
+		// delete from the collection
+		w.kitesColl.Delete(id)
+	})
+}
+
+func (w *webRTCHandler) getDst(dst string) (*Client, error) {
+	if dst == "" {
+		return nil, errDstNotSet
+	}
+
+	dstKite, err := w.kitesColl.Get(dst)
+	if err != nil {
+		return nil, errDstNotRegistered
+	}
+
+	return dstKite.(*Client), nil
+}
+
+// ServeKite implements Hander interface.
+func (w *webRTCHandler) ServeKite(r *Request) (interface{}, error) {
+	var args protocol.WebRTCSignalMessage
+
+	if err := r.Args.One().Unmarshal(&args); err != nil {
+		return nil, fmt.Errorf("invalid query: %s", err)
+	}
+
+	args.Src = r.Client.ID
+
+	w.registerSrc(r.Client)
+
+	dst, err := w.getDst(args.Dst)
+	if err != nil {
+		return nil, err
+	}
+
+	return nil, dst.SendWebRTCRequest(&args)
+}
+
 func (k *Kite) addDefaultHandlers() {
 	// Default RPC methods
 	k.HandleFunc("kite.systemInfo", handleSystemInfo)
@@ -27,6 +93,9 @@ func (k *Kite) addDefaultHandlers() {
 	if runtime.GOOS == "darwin" {
 		k.HandleFunc("kite.notify", handleNotifyDarwin)
 	}
+	if k.WebRTCHandler != nil {
+		k.Handle(WebRTCHandlerName, k.WebRTCHandler)
+	}
 }
 
 // handleSystemInfo returns info about the system (CPU, memory, disk...).

kite.go

@@ -70,6 +70,9 @@ type Kite struct {
 	// Deprecated: Set Config.XHR field instead.
 	ClientFunc func(*sockjsclient.DialOptions) *http.Client
 
+	// WebRTCHandler handles the webrtc responses coming from a signalling server.
+	WebRTCHandler Handler
+
 	// Handlers added with Kite.HandleFunc().
 	handlers     map[string]*Method // method map for exported methods
 	preHandlers  []Handler          // a list of handlers that are executed before any handler
@@ -197,6 +200,10 @@ func NewWithConfig(name, version string, cfg *config.Config) *Kite {
 		muxer:          mux.NewRouter(),
 	}
 
+	if cfg != nil && cfg.UseWebRTC {
+		k.WebRTCHandler = NewWebRCTHandler()
+	}
+
 	// All sockjs communication is done through this endpoint..
 	k.muxer.PathPrefix("/kite").Handler(sockjs.NewHandler("/kite", *cfg.SockJS, k.sockjsHandler))
 

kitetest/kitetest.go

@@ -9,9 +9,9 @@ import (
 	"os/user"
 	"time"
 
-	"github.com/dgrijalva/jwt-go"
+	jwt "github.com/dgrijalva/jwt-go"
 	"github.com/koding/kite/protocol"
-	"github.com/satori/go.uuid"
+	uuid "github.com/satori/go.uuid"
 )
 
 // KeyPair represents PEM encoded RSA key pair.

kontrol/handlers_test.go

@@ -0,0 +1,46 @@
+package kontrol
+
+import (
+	"fmt"
+	"strings"
+	"testing"
+
+	"github.com/koding/kite"
+	"github.com/koding/kite/protocol"
+	"github.com/koding/kite/testkeys"
+)
+
+// createTestKite creates a test kite, caller of this func should close the kite
+func createTestKite(name string, conf *Config, t *testing.T) *HelloKite {
+	k, err := NewHelloKite(name, conf)
+	if err != nil {
+		t.Fatalf("error creating %s: %s", name, err)
+	}
+
+	k.Kite.HandleFunc(kite.WebRTCHandlerName, func(req *kite.Request) (interface{}, error) {
+		return nil, fmt.Errorf("%s is called", name)
+	})
+
+	return k
+}
+
+func TestKontrol_HandleWebRTC(t *testing.T) {
+	kont, conf := startKontrol(testkeys.PrivateThird, testkeys.PublicThird, 5501)
+	defer kont.Close()
+
+	hk1 := createTestKite("kite1", conf, t)
+	defer hk1.Close()
+
+	hk2 := createTestKite("kite2", conf, t)
+	defer hk2.Close()
+
+	err := hk1.Kite.SendWebRTCRequest(&protocol.WebRTCSignalMessage{Dst: hk2.Kite.Id})
+	if err == nil || !strings.Contains(err.Error(), "not registered") {
+		t.Fatalf("expected kite.errDstNotRegistered, got: %+v", err)
+	}
+
+	err = hk2.Kite.SendWebRTCRequest(&protocol.WebRTCSignalMessage{Dst: hk1.Kite.Id})
+	if !strings.Contains(err.Error(), fmt.Sprintf("%s is called", hk1.Kite.Kite().Name)) {
+		t.Fatalf("expected hk1 error, got: %+v", err)
+	}
+}

kontrol/helper_test.go

@@ -51,7 +51,7 @@ func startKontrol(pem, pub string, port int) (*Kontrol, *Config) {
 	conf.KontrolUser = "testuser"
 	conf.KiteKey = testutil.NewToken("testuser", pem, pub).Raw
 	conf.ReadEnvironmentVariables()
-
+	conf.UseWebRTC = true
 	DefaultPort = port
 	kon := New(conf.Copy(), "1.0.0")
 	// kon.Kite.SetLogLevel(kite.DEBUG)

kontrol/kontrol_test.go

@@ -14,7 +14,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/dgrijalva/jwt-go"
+	jwt "github.com/dgrijalva/jwt-go"
 	"github.com/koding/kite"
 	"github.com/koding/kite/kitekey"
 	"github.com/koding/kite/protocol"

kontrolclient.go

@@ -205,6 +205,18 @@ func (k *Kite) GetToken(kite *protocol.Kite) (string, error) {
 	return tkn, nil
 }
 
+// SendWebRTCRequest sends requests to kontrol for signalling purposes.
+func (k *Kite) SendWebRTCRequest(req *protocol.WebRTCSignalMessage) error {
+	if err := k.SetupKontrolClient(); err != nil {
+		return err
+	}
+
+	<-k.kontrol.readyConnected
+
+	_, err := k.kontrol.TellWithTimeout(WebRTCHandlerName, k.Config.Timeout, req)
+	return err
+}
+
 // GetTokenForce is used to obtain a new token for the given kite.
 //
 // It always returns a new token and forces a Kontrol to

protocol/webrtc.go

@@ -0,0 +1,93 @@
+package protocol
+
+import (
+	"encoding/json"
+	"errors"
+	"strings"
+	"sync"
+	"unicode/utf8"
+)
+
+var (
+	errInvalidChar = errors.New("message contains invalid chars")
+	errInvalidOp   = errors.New("invalid start operation")
+)
+
+// WebRTCSignalMessage represents a signalling message between peers and the singalling server
+type WebRTCSignalMessage struct {
+	Type    string          `json:"type,omitempty"`
+	Src     string          `json:"src,omitempty"`
+	Dst     string          `json:"dst,omitempty"`
+	Payload json.RawMessage `json:"payload,omitempty"`
+
+	parsedPayload *Payload
+	isParsed      bool
+	mu            sync.Mutex
+}
+
+// Payload is the content of `payload` in the json
+type Payload struct {
+	Msg *string `json:"msg,omitempty"`
+	Sdp *struct {
+		Type *string `json:"type,omitempty"`
+		Sdp  *string `json:"sdp,omitempty"`
+	} `json:"sdp,omitempty"`
+	Type          *string `json:"type,omitempty"`
+	Label         *string `json:"label,omitempty"`
+	ConnectionID  *string `json:"connectionId,omitempty"`
+	Reliable      *bool   `json:"reliable,omitempty"`
+	Serialization *string `json:"serialization,omitempty"`
+	Browser       *string `json:"browser,omitempty"`
+	Candidate     *struct {
+		Candidate     *string `json:"candidate,omitempty"`
+		SdpMid        *string `json:"sdpMid,omitempty"`
+		SdpMLineIndex *int    `json:"sdpMLineIndex,omitempty"`
+	} `json:"candidate,omitempty"`
+}
+
+// ParsePayload parses the payload if it is not parsed previously. This method
+// can be called concurrently.
+func (w *WebRTCSignalMessage) ParsePayload() (*Payload, error) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+
+	if w.isParsed {
+		return w.parsedPayload, nil
+	}
+
+	payload := &Payload{}
+	if err := json.Unmarshal(w.Payload, payload); err != nil {
+		return nil, err
+	}
+
+	w.parsedPayload = payload
+	return payload, nil
+}
+
+// ParseWebRTCSignalMessage parses the web rtc command/message
+func ParseWebRTCSignalMessage(msg string) (*WebRTCSignalMessage, error) {
+	// All messages are text (utf-8 encoded at present)
+	if !utf8.Valid([]byte(msg)) {
+		return nil, errInvalidChar
+	}
+
+	w := &WebRTCSignalMessage{}
+	if err := json.Unmarshal([]byte(msg), w); err != nil {
+		return nil, err
+	}
+
+	if err := validateOperation(w.Type); err != nil {
+		return nil, err
+	}
+
+	return w, nil
+}
+
+func validateOperation(op string) error {
+	switch strings.ToUpper(op) {
+	case "ANSWER", "OFFER", "CANDIDATE", "LEAVE":
+		return nil
+	default:
+		return errInvalidOp
+	}
+}