v2

README.md

@@ -11,76 +11,65 @@ _Warning_ This is an experimental project, backward compatibility is not guarant
 Here is an example of using:
 
 ```go
-package main
-
-import (
-	"context"
-	"log"
-	"time"
-
-	"github.com/ecwid/control"
-	"github.com/ecwid/control/chrome"
-)
-
 func main() {
-	chromium, err := chrome.Launch(context.TODO(), "--disable-popup-blocking") // you can specify more startup parameters for chrome
-	if err != nil {
-		panic(err)
-	}
-	defer chromium.Close()
-	ctrl := control.New(chromium.GetClient())
-
-	session, err := ctrl.CreatePageTarget("")
+	session, cancel, err := control.Take("--no-startup-window")
 	if err != nil {
 		panic(err)
 	}
+	defer cancel()
 
-	var page = session.Page() // main frame 
-	err = page.Navigate("https://surfparadise.ecwid.com/", control.LifecycleIdleNetwork, time.Second*60)
-	if err != nil {
-		panic(err)
+	retrier := retry.Static{
+		Timeout: 10 * time.Second,
+		Delay:   500 * time.Millisecond, // delay between attempts
 	}
 
-	items, err := page.QuerySelectorAll(".grid-product__title-inner")
+	session.Frame.MustNavigate("https://zoid.ecwid.com")
+
+	var products []string
+	err = retry.Func(retrier, func() error {
+		products = []string{}
+		return session.Frame.QueryAll(".grid-product__title-inner").Then(func(nl control.NodeList) error {
+			return nl.Foreach(func(n *control.Node) error {
+				return n.GetText().Then(func(s string) error {
+					products = append(products, s)
+					return nil
+				})
+			})
+		})
+	})
 	if err != nil {
 		panic(err)
 	}
-	for _, i := range items {
-		title, err := i.GetText()
-		if err != nil {
-			panic(err)
-		}
-		log.Print(title)
+
+    // "must way" throws panic on an error
+
+	for _, node := range session.Frame.MustQueryAll(".grid-product__title-inner") {
+		log.Println(node.MustGetText())
 	}
 }
 ```
 
 You can call any CDP method implemented in protocol package using a session
 ```go
-err = security.SetIgnoreCertificateErrors(sess, security.SetIgnoreCertificateErrorsArgs{
+err = security.SetIgnoreCertificateErrors(session, security.SetIgnoreCertificateErrorsArgs{
     Ignore: true,
 })
 ```
 
-or even call a custom method
+or call a custom unimplemented method
 ```go
-err = sess.Call("Security.setIgnoreCertificateErrors", sendStruct, receiveStruct)
+err = session.Call("Security.setIgnoreCertificateErrors", sendStruct, receiveStruct)
 ```
 
-Subscribe on domain event
+Subscribe on the domain event
 ```go
-cancel := sess.Subscribe("Overlay.screenshotRequested", func(e observe.Value) {
-    v := overlay.ScreenshotRequested{}
-    _= json.Unmarshal(e.Params, &v)
-    doSomething(v.Viewport.Height)
+future := control.Subscribe(session, "Target.targetCreated", func(t target.TargetCreated) bool {
+    return t.TargetInfo.Type == "page"
 })
-defer cancel()
+defer future.Cancel()
 
-// Subscribe on all incoming events
-sess.Subscribe("*", func(e observe.Value) {
-    switch e.Method {
-        case "Overlay.screenshotRequested":
-    }
-})
+// do something here ...
 
-```
+ctx, cancel := context.WithTimeout(context.TODO(), time.Second*10)
+result /* target.TargetCreated */, err := future.Get(ctx)
+```

cdp/broker.go

@@ -0,0 +1,100 @@
+package cdp
+
+import (
+	"sync"
+)
+
+var BrokerChannelSize = 50000
+
+type subscriber struct {
+	sessionID string
+	channel   chan Message
+}
+
+type broker struct {
+	cancel   chan struct{}
+	messages chan Message
+	sub      chan subscriber
+	unsub    chan chan Message
+	lock     *sync.Mutex
+}
+
+func makeBroker() broker {
+	return broker{
+		cancel:   make(chan struct{}),
+		messages: make(chan Message),
+		sub:      make(chan subscriber),
+		unsub:    make(chan chan Message),
+		lock:     &sync.Mutex{},
+	}
+}
+
+func (b broker) run() {
+	var value = map[chan Message]subscriber{}
+	for {
+		select {
+
+		case sub := <-b.sub:
+			value[sub.channel] = sub
+
+		case channel := <-b.unsub:
+			if _, ok := value[channel]; ok {
+				delete(value, channel)
+				close(channel)
+			}
+
+		case <-b.cancel:
+			for msgCh := range value {
+				close(msgCh)
+			}
+			close(b.sub)
+			close(b.unsub)
+			close(b.messages)
+			return
+
+		case message := <-b.messages:
+			for _, subscriber := range value {
+				if message.SessionID == "" || subscriber.sessionID == "" || message.SessionID == subscriber.sessionID {
+					subscriber.channel <- message
+				}
+			}
+		}
+	}
+}
+
+func (b broker) subscribe(sessionID string) chan Message {
+	b.lock.Lock()
+	defer b.lock.Unlock()
+
+	select {
+	case <-b.cancel:
+		return nil
+	default:
+		sub := subscriber{
+			sessionID: sessionID,
+			channel:   make(chan Message, BrokerChannelSize),
+		}
+		b.sub <- sub
+		return sub.channel
+	}
+}
+
+func (b broker) unsubscribe(value chan Message) {
+	b.lock.Lock()
+	select {
+	case <-b.cancel:
+	default:
+		b.unsub <- value
+	}
+	b.lock.Unlock()
+}
+
+func (b broker) publish(msg Message) {
+	b.messages <- msg
+}
+
+func (b broker) Cancel() {
+	b.lock.Lock()
+	close(b.cancel)
+	b.lock.Unlock()
+}