support multiple clients and reorg (#10)

## Problem
Previously, 1+ connected CDP clients would race for the events coming from event.
The backend would randomly "load balance" between them.

 ## Solution
Copy events to each connected client.

 ## Also
 - some cleanup and reorg, still WIP

Author: gmarik

examples/helloworld/main.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"context"
 	"flag"
 	"fmt"
 	"log"
@@ -15,21 +16,26 @@ import (
 )
 
 var (
-	HTTP_CDP_Addr    = "localhost:9229"
-	HTTP_Server_Addr = "localhost:8081"
+	HTTP_CDP_HostPort   = "localhost:9229"
+	HTTP_Proxy_HostPort = "localhost:8081"
 )
 
 func main() {
-	flag.StringVar(&HTTP_CDP_Addr, "http-cdp-addr", HTTP_CDP_Addr, "chrome devtools protocol listener address")
-	flag.StringVar(&HTTP_Server_Addr, "http-proxy-addr", HTTP_Server_Addr, "HTTP server listener address")
+	flag.StringVar(&HTTP_CDP_HostPort, "http-cdp-hostport", HTTP_CDP_HostPort, "chrome devtools protocol listener address(host:port)")
+	flag.StringVar(&HTTP_Proxy_HostPort, "http-proxy-hostport", HTTP_Proxy_HostPort, "HTTP proxy listener address(host:port)")
 	flag.Parse()
 
 	var eb = httpcdp.NewEventBus()
+	var ctx = context.Background()
 
 	go func() {
-		log.Printf("http.devtools: ListenAndServe.address=%q", HTTP_CDP_Addr)
-		if err := http.ListenAndServe(HTTP_CDP_Addr, httpcdp.Devtools(eb)); err != nil {
-			log.Fatalf("http.devtools: ListenAndServe.error=%q", err)
+		log.Printf("devtools: http.ListenAndServe: hostport=%q", HTTP_CDP_HostPort)
+		s := httpcdp.Server{
+			Eventbus: eb,
+			HostPort: HTTP_CDP_HostPort,
+		}
+		if err := s.ListenAndServe(ctx); err != nil {
+			log.Fatalf("devtools: http.ListenAndServe: error=%q", err)
 		}
 	}()
 
@@ -39,9 +45,9 @@ func main() {
 			fmt.Fprintf(w, "hello world")
 		})
 
-		log.Printf("http.proxy: ListenAndServe.address=%q", HTTP_Server_Addr)
-		if err := http.ListenAndServe(HTTP_Server_Addr, cdphttp.Handler(eb, handler)); err != nil {
-			log.Fatalf("http.proxy: ListenAndServe.error=%q", err)
+		log.Printf("proxy: http.ListenAndServe: hostport=%q", HTTP_Proxy_HostPort)
+		if err := http.ListenAndServe(HTTP_Proxy_HostPort, cdphttp.Handler(eb, handler)); err != nil {
+			log.Fatalf("proxy: http.ListenAndServe: error=%q", err)
 		}
 	}()
 

main/cdp-proxy/http.go

@@ -0,0 +1,118 @@
+package main
+
+import (
+	"fmt"
+	"io"
+	"log"
+	"net"
+	"net/http"
+	"net/http/httputil"
+	"net/url"
+	"time"
+)
+
+var proxy = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+	u := *r.URL
+	u.Host = r.Host
+	u.Scheme = "http"
+
+	if _, p, _ := net.SplitHostPort(r.Host); p == "443" {
+		u.Scheme = "https"
+	}
+
+	log.Printf("[proxy] %s", u.String())
+
+	if r.Method == http.MethodConnect {
+		newTunnel(&u).ServeHTTP(w, r)
+	} else {
+		newForwardProxy(&u).ServeHTTP(w, r)
+	}
+})
+
+func newTunnel(u *url.URL) http.Handler {
+
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		httpErr := func(code int, err error) {
+			log.Printf("[CONNECT]: error=%v", err)
+			http.Error(w, http.StatusText(code), code)
+		}
+		log.Printf("[CONNECT]: start:%s", u.Host)
+		defer log.Printf("[CONNECT]: done: %s", u.Host)
+
+		dconn, err := net.DialTimeout("tcp", r.Host, 5*time.Second)
+		if err != nil {
+			httpErr(http.StatusServiceUnavailable, err)
+			return
+		}
+
+		w.WriteHeader(http.StatusOK)
+		if f, ok := w.(http.Flusher); ok {
+			f.Flush()
+		} else {
+			log.Println("http.Flusher: unavailable")
+		}
+
+		var sconn net.Conn
+		if h, ok := w.(http.Hijacker); !ok {
+			httpErr(http.StatusInternalServerError, fmt.Errorf("http.Hijacker: unavailable"))
+			return
+		} else {
+			conn, _, err := h.Hijack()
+			if err != nil {
+				httpErr(http.StatusServiceUnavailable, err)
+				return
+			}
+			sconn = conn
+		}
+
+		type readWriteCloser interface {
+			net.Conn
+			CloseRead() error
+			CloseWrite() error
+		}
+
+		var src, dst = sconn.(readWriteCloser), dconn.(readWriteCloser)
+		// TODO:
+		var done = make(chan struct{})
+		go func() {
+			n, err := io.Copy(src, dst)
+			log.Printf("src<-dst: n=%d error=%v", n, err)
+			dst.CloseRead()
+			src.CloseWrite()
+			done <- struct{}{}
+		}()
+		go func() {
+			n, err := io.Copy(dst, src)
+			log.Printf("src->dst: n=%d error=%v", n, err)
+			dst.CloseWrite()
+			src.CloseRead()
+			done <- struct{}{}
+		}()
+		<-done
+		<-done
+	})
+}
+
+func newForwardProxy(target *url.URL) *httputil.ReverseProxy {
+	return &httputil.ReverseProxy{
+		Director: func(req *http.Request) {
+			// TODO:
+			req.Host = target.Host
+			if _, ok := req.Header["User-Agent"]; !ok {
+				req.Header.Set("User-Agent", "")
+			}
+		},
+		// Transport: loggingTransport(http.DefaultTransport.RoundTrip),
+	}
+}
+
+type loggingTransport func(*http.Request) (*http.Response, error)
+
+func (lt loggingTransport) RoundTrip(r *http.Request) (*http.Response, error) {
+	bs, _ := httputil.DumpRequestOut(r, false)
+	log.Println("REQ:", string(bs))
+	re, err := lt(r)
+	bs, _ = httputil.DumpResponse(re, false)
+	log.Println("RES:", string(bs))
+	return re, err
+}

main/cdp-proxy/httpcdp/body_store.go

@@ -0,0 +1,30 @@
+package httpcdp
+
+import (
+	"bytes"
+	"sync"
+)
+
+type bodyStore struct {
+	m sync.Map
+}
+
+func newStore() *bodyStore {
+	return &bodyStore{}
+}
+
+func (bs *bodyStore) Load(key string) (*bytes.Buffer, bool) {
+	val, ok := bs.m.Load(key)
+	if !ok {
+		return nil, false
+	}
+	if buf, ok := val.(*bytes.Buffer); ok {
+		return buf, true
+	}
+	return nil, false
+}
+
+func (bs *bodyStore) LoadOrStore(key string, buf *bytes.Buffer) (actual *bytes.Buffer, loaded bool) {
+	val, ok := bs.m.LoadOrStore(key, buf)
+	return val.(*bytes.Buffer), ok
+}

main/cdp-proxy/httpcdp/cdp.go

@@ -0,0 +1,89 @@
+package httpcdp
+
+import (
+	"context"
+	"encoding/base64"
+	"encoding/json"
+	"fmt"
+	"log"
+	"net/http"
+
+	"github.com/gorilla/websocket"
+)
+
+// devtools://devtools/bundled/inspector.html?experiments=true&v8only=true&ws=localhost:9229/cdp-proxy
+
+// https://medium.com/@paul_irish/debugging-node-js-nightlies-with-chrome-devtools-7c4a1b95ae27
+// conn.WriteMessage(websocket.TextMessage, []byte(fmt.Sprintf(`{"method": "Page.disable","params":{}}`)))
+func handleCDP(ctx context.Context, conn *websocket.Conn, e event) error {
+	switch m := e.Method; {
+	case m == "Page.canScreencast" ||
+		m == "Network.canEmulateNetworkConditions" ||
+		m == "Emulation.canEmulate":
+		respond(conn, e.ID, `{"result":false}`)
+	case m == "Page.getResourceTree":
+		// Window decoration
+		respond(conn, e.ID,
+			`{"frameTree": { "frame":{"id":1,"url":"http://cdp-proxy","mimeType":"other"},"childFrames":[],"resources":[]}}`,
+		)
+	case m == "Network.getResponseBody":
+		params, ok := e.Params.(map[string]interface{})
+		if !ok {
+			return nil
+		}
+		e.reqID, ok = params["requestId"].(string)
+		if !ok {
+			return nil
+		}
+
+		if buf, ok := store.Load(e.reqID); !ok {
+			respond(conn, e.ID, `{"body":"","base64Encoded":true}`)
+		} else {
+			result := map[string]interface{}{
+				"base64Encoded": true,
+				"body":          base64.StdEncoding.Strict().EncodeToString(buf.Bytes()),
+			}
+
+			data, err := json.Marshal(result)
+			if err != nil {
+				log.Printf("json.Marshal: error=%q", err)
+				return nil
+			}
+
+			// https://chromedevtools.github.io/devtools-protocol/1-2/Network#method-getResponseBody
+			respond(conn, e.ID, string(data))
+		}
+	default:
+		respond(conn, e.ID, `{}`)
+	}
+
+	return nil
+}
+
+func (s *Server) metadata(w http.ResponseWriter, r *http.Request) {
+	// NOTE:
+	// devtoolsFrontendUrl is not respected when opened from `chrome://inspect`
+	var (
+		hostPort = s.HostPort
+		wsURL    = hostPort + "/cdp"
+	)
+	fmt.Fprintf(w, `[{
+			"id": "cdp-proxy",
+			"title": "cdp-proxy",
+			"type": "proxy",
+			"description": "cdp-proxy requests",
+			"faviconUrl": "https://nodejs.org/static/favicon.ico",
+			"url": %q,
+			"devtoolsFrontendUrl": "ws=%s",
+			"webSocketDebuggerUrl": "ws://%s"
+		}]`, hostPort, wsURL, wsURL)
+}
+
+func writeConn(conn *websocket.Conn, p []byte) (int, error) {
+	log.Printf("[CDP<-] %.120s", string(p))
+	return len(p), conn.WriteMessage(websocket.TextMessage, p)
+}
+
+func respond(conn *websocket.Conn, id int, p string) (int, error) {
+	return writeConn(conn, []byte(fmt.Sprintf(`{"id":%d,"result":%s}`, id, p)))
+}