From fa959a25708ae9e5a03954aab67c748d364d257d Mon Sep 17 00:00:00 2001
From: bel <me@me.me>
Date: Thu, 7 May 2020 19:39:27 -0600
Subject: [PATCH] video call with imperfect reconnect, persistent id, possibly
 many clients

---
 config.go         |  6 ++---
 main.go           | 10 +++++++--
 pool.go           | 53 ++++++++++++++++++++++++++++++++------------
 public/index.html |  7 ++++--
 public/webrtc.js  | 31 ++++++++++++++++++++++++--
 server.go         |  7 +++---
 ws.go             | 56 ++++++++++++++---------------------------------
 7 files changed, 105 insertions(+), 65 deletions(-)

diff --git a/config.go b/config.go
index 93fdaa2..7db9db4 100644
--- a/config.go
+++ b/config.go
@@ -6,9 +6,9 @@ func config() *args.ArgSet {
 	as := args.NewArgSet()
 
 	as.Append(args.INT, "p", "port to listen on", "58080")
-	as.Append(args.STRING, "d", "root dir to serve static", ".")
-	as.Append(args.STRING, "crt", "path to crt", "./cert.crt")
-	as.Append(args.STRING, "key", "path to key", "./cert.key")
+	as.Append(args.STRING, "d", "root dir to serve static", "./public")
+	as.Append(args.STRING, "crt", "path to crt", "./testdata/scratch.crt")
+	as.Append(args.STRING, "key", "path to key", "./testdata/scratch.key")
 
 	if err := as.Parse(); err != nil {
 		panic(err)
diff --git a/main.go b/main.go
index b52577d..a4a7282 100644
--- a/main.go
+++ b/main.go
@@ -28,7 +28,13 @@ func main() {
 	c := as.GetString("crt")
 	k := as.GetString("key")
 	log.Printf("listening on %q", httpsServer.Addr)
-	if err := httpsServer.ListenAndServeTLS(c, k); err != nil {
-		panic(err)
+	if c == "" && k == "" {
+		if err := httpsServer.ListenAndServe(); err != nil {
+			panic(err)
+		}
+	} else {
+		if err := httpsServer.ListenAndServeTLS(c, k); err != nil {
+			panic(err)
+		}
 	}
 }
diff --git a/pool.go b/pool.go
index 399db03..51b49aa 100644
--- a/pool.go
+++ b/pool.go
@@ -1,26 +1,51 @@
 package main
 
-import "io"
+import (
+	"io"
+	"io/ioutil"
+	"log"
+	"sync"
+
+	"github.com/gorilla/websocket"
+)
 
 type Pool struct {
-	writers []io.Writer
-	r       *io.PipeReader
-	w       *io.PipeWriter
+	conns *sync.Map //map[string]*websocket.Conn
 }
 
 func NewPool() *Pool {
-	r, w := io.Pipe()
 	return &Pool{
-		writers: make([]io.Writer, 0),
-		r:       r,
-		w:       w,
+		conns: &sync.Map{}, //map[string]*websocket.Conn{},
 	}
 }
 
-func (p *Pool) Read(b []byte) (int, error) {
-	return p.r.Read(b)
-}
-
-func (p *Pool) Write(b []byte) (int, error) {
-	return p.w.Write(b)
+func (p *Pool) Broadcast(mt int, r io.Reader) error {
+	b, err := ioutil.ReadAll(r)
+	if err != nil {
+		return err
+	}
+	n := 1000000
+	cnt := 0
+	p.conns.Range(func(k, v interface{}) bool {
+		k = k.(string)
+		conn := v.(*websocket.Conn)
+		cnt += 1
+		w, err := conn.NextWriter(mt)
+		if err != nil {
+			p.conns.Delete(k)
+			return true
+		}
+		defer w.Close()
+		m, err := w.Write(b)
+		if err != nil {
+			p.conns.Delete(k)
+			return true
+		}
+		if m < n {
+			n = m
+		}
+		return true
+	})
+	log.Printf("%d writes, %d min size, %d wanted size", cnt, n, len(b))
+	return nil
 }
diff --git a/public/index.html b/public/index.html
index 0f1299f..1b0ef31 100755
--- a/public/index.html
+++ b/public/index.html
@@ -5,9 +5,12 @@
     <script src="webrtc.js"></script>
     <style>
       video {
-         max-width: 100%;
-         min-width: 100%;
+         width: 100%;
+         width: 100%;
+         max-width: 400px;
+         min-width: 400px;
          border: 1px solid black;
+         display: inline-block;
       }
     </style>
   </head>
diff --git a/public/webrtc.js b/public/webrtc.js
index 97c860d..580cc49 100755
--- a/public/webrtc.js
+++ b/public/webrtc.js
@@ -12,13 +12,40 @@ var peerConnectionConfig = {
   ]
 };
 
+function getCookie(cname) {
+  var name = cname + "=";
+  var decodedCookie = decodeURIComponent(document.cookie);
+  var ca = decodedCookie.split(';');
+  for(var i = 0; i <ca.length; i++) {
+    var c = ca[i];
+    while (c.charAt(0) == ' ') {
+      c = c.substring(1);
+    }
+    if (c.indexOf(name) == 0) {
+      return c.substring(name.length, c.length);
+    }
+  }
+  return "";
+}
+
+function setCookie(cname, cvalue) {
+  var d = new Date();
+  d.setTime(d.getTime() + (1*24*60*60*1000));
+  var expires = "expires="+ d.toUTCString();
+  document.cookie = cname + "=" + cvalue + ";" + expires + ";path=/";
+}
+
 function pageReady() {
-  uuid = createUUID();
+  uuid = getCookie("uuid");
+  if (!uuid) {
+    uuid = createUUID();
+    setCookie("uuid", uuid);
+  }
 
   localVideo = document.getElementById('localVideo');
   remoteVideo = document.getElementById('remoteVideo');
 
-  serverConnection = new WebSocket('wss://' + window.location.hostname + '/abc');
+  serverConnection = new WebSocket('wss://' + window.location.hostname + '/abc?uuid=' + uuid);
   serverConnection.onmessage = gotMessageFromServer;
 
   var constraints = {
diff --git a/server.go b/server.go
index 86eab94..2c691d7 100644
--- a/server.go
+++ b/server.go
@@ -21,11 +21,12 @@ func New() *Server {
 }
 
 func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
-	log.Println("serving", r.URL)
-	if _, err := os.Stat(path.Join(config().GetString("d"), r.URL.Path[1:])); os.IsNotExist(err) {
+	log.Println("ext", path.Ext(r.URL.Path))
+	if path.Ext(r.URL.Path) != "" {
+		s.fs.ServeHTTP(w, r)
+	} else if _, err := os.Stat(path.Join(config().GetString("d"), r.URL.Path[1:])); os.IsNotExist(err) {
 		s.ws.ServeHTTP(w, r)
 	} else {
-		log.Printf("Serving static %q from %q", r.URL.Path, config().GetString("d"))
 		s.fs.ServeHTTP(w, r)
 	}
 }
diff --git a/ws.go b/ws.go
index 570f4f0..f63f1b3 100644
--- a/ws.go
+++ b/ws.go
@@ -1,11 +1,11 @@
 package main
 
 import (
-	"io"
 	"log"
 	"net/http"
 	"sync"
 
+	"github.com/google/uuid"
 	"github.com/gorilla/websocket"
 )
 
@@ -17,8 +17,8 @@ type WS struct {
 func NewWS() *WS {
 	return &WS{
 		upgrader: websocket.Upgrader{
-			ReadBufferSize:  10240,
-			WriteBufferSize: 10240,
+			ReadBufferSize:  16384,
+			WriteBufferSize: 16384,
 			CheckOrigin:     func(_ *http.Request) bool { return true },
 		},
 		pools: &sync.Map{},
@@ -33,51 +33,29 @@ func (ws *WS) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 }
 
 func (ws *WS) serveHTTP(w http.ResponseWriter, r *http.Request) error {
+	id := r.URL.Query().Get("uuid")
+	if len(id) == 0 {
+		id = uuid.New().String()
+	}
 	log.Println("ws serve http", r.URL.Path)
 	pooli, ok := ws.pools.Load(r.URL.Path)
 	if !ok {
 		pooli = NewPool()
+		ws.pools.Store(r.URL.Path, pooli.(*Pool))
 	}
 	pool := pooli.(*Pool)
-	log.Println("ws upgrade")
 	conn, err := ws.upgrader.Upgrade(w, r, nil)
 	if err != nil {
 		return err
 	}
-	log.Println("ws next reader")
-	mt, _, err := conn.NextReader()
-	if err != nil {
-		return err
+	pool.conns.Store(id, conn)
+	for {
+		mt, reader, err := conn.NextReader()
+		if err != nil {
+			return err
+		}
+		if err := pool.Broadcast(mt, reader); err != nil {
+			return err
+		}
 	}
-	log.Println("funcs")
-	go func() {
-		for {
-			_, r, err := conn.NextReader()
-			if err != nil {
-				panic(err)
-			}
-			log.Println("copying to pool...")
-			if _, err := io.Copy(pool, r); err != nil {
-				panic(err)
-			}
-			log.Println("/copying to pool...")
-		}
-	}()
-	func() {
-		for {
-			w, err := conn.NextWriter(mt)
-			if err != nil {
-				panic(err)
-			}
-			log.Println("copying from pool...")
-			if _, err := io.Copy(w, pool); err != nil { // todo impl broadcast to channel;; sync map to all channels, goes to a forking reader-writer pipe, all listeners to broadcast read from pipe
-				panic(err)
-			}
-			log.Println("/copying from pool...")
-			if err := w.Close(); err != nil {
-				panic(err)
-			}
-		}
-	}()
-	return nil
 }