Event refactor

rutcast · Jul 18, 2017 · 4731d8e · 4731d8e
1 parent 6eaece9
commit 4731d8e
Show file tree

Hide file tree

Showing 17 changed files with 197 additions and 124 deletions.
diff --git a/api.go b/api.go
@@ -6,6 +6,7 @@ import (
 	"os"
 	"time"
 
+	"github.com/googollee/go-socket.io"
 	gh "github.com/gorilla/handlers"
 	"github.com/gorilla/mux"
 	"github.com/play-with-docker/play-with-docker/config"
@@ -21,7 +22,14 @@ func main() {
 
 	bypassCaptcha := len(os.Getenv("GOOGLE_RECAPTCHA_DISABLED")) > 0
 
-	server := handlers.Broadcast.GetHandler()
+	server, err := socketio.NewServer(nil)
+	if err != nil {
+		log.Fatal(err)
+	}
+	server.On("connection", handlers.WS)
+	server.On("error", handlers.WSError)
+
+	handlers.RegisterEvents(server)
 
 	r := mux.NewRouter()
 	corsRouter := mux.NewRouter()

diff --git a/event/event.go b/event/event.go
@@ -2,15 +2,24 @@ package event
 
 type EventType string
 
+func (e EventType) String() string {
+	return string(e)
+}
+
 const INSTANCE_VIEWPORT_RESIZE EventType = "instance viewport resize"
 const INSTANCE_DELETE EventType = "instance delete"
 const INSTANCE_NEW EventType = "instance new"
+const INSTANCE_STATS EventType = "instance stats"
+const INSTANCE_TERMINAL_OUT EventType = "instance terminal out"
 const SESSION_END EventType = "session end"
 const SESSION_READY EventType = "session ready"
+const SESSION_BUILDER_OUT EventType = "session builder out"
 
-type Handler func(args ...interface{})
+type Handler func(sessionId string, args ...interface{})
+type AnyHandler func(eventType EventType, sessionId string, args ...interface{})
 
 type EventApi interface {
-	Emit(name EventType, args ...interface{})
+	Emit(name EventType, sessionId string, args ...interface{})
 	On(name EventType, handler Handler)
+	OnAny(handler AnyHandler)
 }
diff --git a/event/local_broker.go b/event/local_broker.go
@@ -5,11 +5,12 @@ import "sync"
 type localBroker struct {
 	sync.Mutex
 
-	handlers map[EventType][]Handler
+	handlers    map[EventType][]Handler
+	anyHandlers []AnyHandler
 }
 
 func NewLocalBroker() *localBroker {
-	return &localBroker{handlers: map[EventType][]Handler{}}
+	return &localBroker{handlers: map[EventType][]Handler{}, anyHandlers: []AnyHandler{}}
 }
 
 func (b *localBroker) On(name EventType, handler Handler) {
@@ -22,13 +23,25 @@ func (b *localBroker) On(name EventType, handler Handler) {
 	b.handlers[name] = append(b.handlers[name], handler)
 }
 
-func (b *localBroker) Emit(name EventType, args ...interface{}) {
+func (b *localBroker) OnAny(handler AnyHandler) {
 	b.Lock()
 	defer b.Unlock()
 
-	if b.handlers[name] != nil {
-		for _, handler := range b.handlers[name] {
-			handler(args...)
+	b.anyHandlers = append(b.anyHandlers, handler)
+}
+
+func (b *localBroker) Emit(name EventType, sessionId string, args ...interface{}) {
+	go func() {
+		b.Lock()
+		defer b.Unlock()
+
+		for _, handler := range b.anyHandlers {
+			handler(name, sessionId, args...)
 		}
-	}
+		if b.handlers[name] != nil {
+			for _, handler := range b.handlers[name] {
+				handler(sessionId, args...)
+			}
+		}
+	}()
 }
diff --git a/event/local_broker_test.go b/event/local_broker_test.go
@@ -7,25 +7,54 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
-func TestLocalBroker(t *testing.T) {
+func TestLocalBroker_On(t *testing.T) {
 	broker := NewLocalBroker()
 
 	called := 0
+	receivedSessionId := ""
 	receivedArgs := []interface{}{}
 
 	wg := sync.WaitGroup{}
 	wg.Add(1)
 
-	broker.On(INSTANCE_NEW, func(args ...interface{}) {
+	broker.On(INSTANCE_NEW, func(sessionId string, args ...interface{}) {
 		called++
+		receivedSessionId = sessionId
 		receivedArgs = args
 		wg.Done()
 	})
-	broker.Emit(SESSION_READY)
-	broker.Emit(INSTANCE_NEW, "foo", "bar")
+	broker.Emit(SESSION_READY, "1")
+	broker.Emit(INSTANCE_NEW, "2", "foo", "bar")
 
 	wg.Wait()
 
 	assert.Equal(t, 1, called)
+	assert.Equal(t, "2", receivedSessionId)
 	assert.Equal(t, []interface{}{"foo", "bar"}, receivedArgs)
 }
+
+func TestLocalBroker_OnAny(t *testing.T) {
+	broker := NewLocalBroker()
+
+	var receivedEvent EventType
+	receivedSessionId := ""
+	receivedArgs := []interface{}{}
+
+	wg := sync.WaitGroup{}
+	wg.Add(1)
+
+	broker.OnAny(func(eventType EventType, sessionId string, args ...interface{}) {
+		receivedSessionId = sessionId
+		receivedArgs = args
+		receivedEvent = eventType
+		wg.Done()
+	})
+	broker.Emit(SESSION_READY, "1")
+
+	wg.Wait()
+
+	var expectedArgs []interface{}
+	assert.Equal(t, SESSION_READY, receivedEvent)
+	assert.Equal(t, "1", receivedSessionId)
+	assert.Equal(t, expectedArgs, receivedArgs)
+}
diff --git a/handlers/bootstrap.go b/handlers/bootstrap.go
@@ -5,14 +5,17 @@ import (
 	"os"
 
 	"github.com/docker/docker/client"
+	"github.com/googollee/go-socket.io"
 	"github.com/play-with-docker/play-with-docker/config"
 	"github.com/play-with-docker/play-with-docker/docker"
+	"github.com/play-with-docker/play-with-docker/event"
 	"github.com/play-with-docker/play-with-docker/pwd"
 	"github.com/play-with-docker/play-with-docker/storage"
 )
 
 var core pwd.PWDApi
-var Broadcast pwd.BroadcastApi
+var e event.EventApi
+var ws *socketio.Server
 
 func Bootstrap() {
 	c, err := client.NewEnvClient()
@@ -22,18 +25,24 @@ func Bootstrap() {
 
 	d := docker.NewDocker(c)
 
-	Broadcast, err = pwd.NewBroadcast(WS, WSError)
-	if err != nil {
-		log.Fatal(err)
-	}
+	e = event.NewLocalBroker()
 
-	t := pwd.NewScheduler(Broadcast, d)
+	t := pwd.NewScheduler(e, d)
 
 	s, err := storage.NewFileStorage(config.SessionsFile)
 
 	if err != nil && !os.IsNotExist(err) {
 		log.Fatal("Error decoding sessions from disk ", err)
 	}
-	core = pwd.NewPWD(d, t, Broadcast, s)
+	core = pwd.NewPWD(d, t, e, s)
+
+}
+
+func RegisterEvents(s *socketio.Server) {
+	ws = s
+	e.OnAny(broadcastEvent)
+}
 
+func broadcastEvent(eventType event.EventType, sessionId string, args ...interface{}) {
+	ws.BroadcastTo(sessionId, eventType.String(), args...)
 }
diff --git a/pwd/broadcast.go b/pwd/broadcast.go
diff --git a/pwd/broadcast_mock_test.go b/pwd/broadcast_mock_test.go
diff --git a/pwd/client.go b/pwd/client.go
@@ -4,6 +4,7 @@ import (
 	"log"
 	"time"
 
+	"github.com/play-with-docker/play-with-docker/event"
 	"github.com/play-with-docker/play-with-docker/pwd/types"
 )
 
@@ -41,7 +42,7 @@ func (p *pwd) ClientClose(client *types.Client) {
 func (p *pwd) notifyClientSmallestViewPort(session *types.Session) {
 	vp := p.SessionGetSmallestViewPort(session)
 	// Resize all terminals in the session
-	p.broadcast.BroadcastTo(session.Id, "viewport resize", vp.Cols, vp.Rows)
+	p.event.Emit(event.INSTANCE_VIEWPORT_RESIZE, session.Id, vp.Cols, vp.Rows)
 	for _, instance := range session.Instances {
 		err := p.InstanceResizeTerminal(instance, vp.Rows, vp.Cols)
 		if err != nil {

diff --git a/pwd/client_test.go b/pwd/client_test.go
@@ -1,20 +1,22 @@
 package pwd
 
 import (
+	"sync"
 	"testing"
 	"time"
 
+	"github.com/play-with-docker/play-with-docker/event"
 	"github.com/play-with-docker/play-with-docker/pwd/types"
 	"github.com/stretchr/testify/assert"
 )
 
 func TestClientNew(t *testing.T) {
 	docker := &mockDocker{}
 	tasks := &mockTasks{}
-	broadcast := &mockBroadcast{}
+	e := event.NewLocalBroker()
 	storage := &mockStorage{}
 
-	p := NewPWD(docker, tasks, broadcast, storage)
+	p := NewPWD(docker, tasks, e, storage)
 
 	session, err := p.SessionNew(time.Hour, "", "", "")
 	assert.Nil(t, err)
@@ -26,33 +28,34 @@ func TestClientNew(t *testing.T) {
 }
 
 func TestClientResizeViewPort(t *testing.T) {
+	wg := sync.WaitGroup{}
+	wg.Add(1)
 	docker := &mockDocker{}
 	tasks := &mockTasks{}
-	broadcast := &mockBroadcast{}
+	e := event.NewLocalBroker()
 
 	broadcastedSessionId := ""
-	broadcastedEventName := ""
 	broadcastedArgs := []interface{}{}
 
-	broadcast.broadcastTo = func(sessionId, eventName string, args ...interface{}) {
+	e.On(event.INSTANCE_VIEWPORT_RESIZE, func(sessionId string, args ...interface{}) {
 		broadcastedSessionId = sessionId
-		broadcastedEventName = eventName
 		broadcastedArgs = args
-	}
+		wg.Done()
+	})
 
 	storage := &mockStorage{}
 
-	p := NewPWD(docker, tasks, broadcast, storage)
+	p := NewPWD(docker, tasks, e, storage)
 
 	session, err := p.SessionNew(time.Hour, "", "", "")
 	assert.Nil(t, err)
 	client := p.ClientNew("foobar", session)
 
 	p.ClientResizeViewPort(client, 80, 24)
+	wg.Wait()
 
 	assert.Equal(t, types.ViewPort{Cols: 80, Rows: 24}, client.ViewPort)
 	assert.Equal(t, session.Id, broadcastedSessionId)
-	assert.Equal(t, "viewport resize", broadcastedEventName)
 	assert.Equal(t, uint(80), broadcastedArgs[0])
 	assert.Equal(t, uint(24), broadcastedArgs[1])
 }
diff --git a/pwd/instance.go b/pwd/instance.go
@@ -11,6 +11,7 @@ import (
 
 	"github.com/play-with-docker/play-with-docker/config"
 	"github.com/play-with-docker/play-with-docker/docker"
+	"github.com/play-with-docker/play-with-docker/event"
 	"github.com/play-with-docker/play-with-docker/pwd/types"
 
 	"golang.org/x/text/encoding"
@@ -19,11 +20,11 @@ import (
 type sessionWriter struct {
 	sessionId    string
 	instanceName string
-	broadcast    BroadcastApi
+	event        event.EventApi
 }
 
 func (s *sessionWriter) Write(p []byte) (n int, err error) {
-	s.broadcast.BroadcastTo(s.sessionId, "terminal out", s.instanceName, string(p))
+	s.event.Emit(event.INSTANCE_TERMINAL_OUT, s.sessionId, s.instanceName, string(p))
 	return len(p), nil
 }
 
@@ -52,7 +53,7 @@ func (p *pwd) InstanceAttachTerminal(instance *types.Instance) error {
 	}
 
 	encoder := encoding.Replacement.NewEncoder()
-	sw := &sessionWriter{sessionId: instance.Session.Id, instanceName: instance.Name, broadcast: p.broadcast}
+	sw := &sessionWriter{sessionId: instance.Session.Id, instanceName: instance.Name, event: p.event}
 	instance.Terminal = conn
 	io.Copy(encoder.Writer(sw), conn)
 
@@ -139,7 +140,7 @@ func (p *pwd) InstanceDelete(session *types.Session, instance *types.Instance) e
 		return err
 	}
 
-	p.broadcast.BroadcastTo(session.Id, "delete instance", instance.Name)
+	p.event.Emit(event.INSTANCE_DELETE, session.Id, instance.Name)
 
 	delete(session.Instances, instance.Name)
 	if err := p.storage.SessionPut(session); err != nil {
@@ -238,7 +239,7 @@ func (p *pwd) InstanceNew(session *types.Session, conf InstanceConfig) (*types.I
 		return nil, err
 	}
 
-	p.broadcast.BroadcastTo(session.Id, "new instance", instance.Name, instance.IP, instance.Hostname)
+	p.event.Emit(event.INSTANCE_NEW, session.Id, instance.Name, instance.IP, instance.Hostname)
 
 	p.setGauges()