Implement connection topic

Author: jmaralo

backend/cmd/main.go

@@ -37,6 +37,7 @@ import (
 	"github.com/HyperloopUPV-H8/h9-backend/internal/ws_handle"
 	"github.com/HyperloopUPV-H8/h9-backend/pkg/abstraction"
 	"github.com/HyperloopUPV-H8/h9-backend/pkg/broker"
+	connection_topic "github.com/HyperloopUPV-H8/h9-backend/pkg/broker/topics/connection"
 	data_topic "github.com/HyperloopUPV-H8/h9-backend/pkg/broker/topics/data"
 	"github.com/HyperloopUPV-H8/h9-backend/pkg/transport"
 	"github.com/HyperloopUPV-H8/h9-backend/pkg/transport/network/sniffer"
@@ -161,10 +162,12 @@ func main() {
 		OnUserPush: func(push abstraction.BrokerPush) {},
 	})
 
-	dataTopic := data_topic.NewUpdateTopic(time.Second / 60)
+	dataTopic := data_topic.NewUpdateTopic(time.Second / 10)
 	defer dataTopic.Stop()
+	connectionTopic := connection_topic.NewUpdateTopic()
 
 	broker.AddTopic(data_topic.UpdateName, dataTopic)
+	broker.AddTopic(connection_topic.UpdateName, connectionTopic)
 
 	connections := make(chan *websocket.Client)
 	upgrader := websocket.NewUpgrader(connections)
@@ -226,7 +229,7 @@ func main() {
 		},
 
 		OnConnectionUpdate: func(target abstraction.TransportTarget, isConnected bool) {
-			connectionTransfer.Update(string(target), isConnected)
+			connectionTopic.Push(connection_topic.NewConnection(string(target), isConnected))
 		},
 	})
 

backend/pkg/broker/topics/connection/push.go

@@ -0,0 +1,19 @@
+package data
+
+import "github.com/HyperloopUPV-H8/h9-backend/pkg/abstraction"
+
+type Connection struct {
+	Name        string `json:"name"`
+	IsConnected bool   `json:"isConnected"`
+}
+
+func NewConnection(name string, isConnected bool) *Connection {
+	return &Connection{
+		Name:        name,
+		IsConnected: isConnected,
+	}
+}
+
+func (connection *Connection) Topic() abstraction.BrokerTopic {
+	return UpdateName
+}

backend/pkg/broker/topics/connection/update.go

@@ -1,27 +1,118 @@
 package data
 
 import (
+	"encoding/json"
+	"fmt"
+	"sync"
+
 	"github.com/HyperloopUPV-H8/h9-backend/pkg/abstraction"
+	"github.com/HyperloopUPV-H8/h9-backend/pkg/broker/topics"
 	"github.com/HyperloopUPV-H8/h9-backend/pkg/websocket"
+	"github.com/google/uuid"
+	ws "github.com/gorilla/websocket"
 )
 
 const UpdateName abstraction.BrokerTopic = "connection/update"
+const SubscribeName abstraction.BrokerTopic = "connection/update"
 
 type Update struct {
-	pool *websocket.Pool
-	api  abstraction.BrokerAPI
+	subscriberMx *sync.Mutex
+	subscribers  map[websocket.ClientId]struct{}
+	connectionMx *sync.Mutex
+	connections  map[string]Connection
+	pool         *websocket.Pool
+	api          abstraction.BrokerAPI
+}
+
+func NewUpdateTopic() *Update {
+	topic := &Update{
+		subscriberMx: &sync.Mutex{},
+		subscribers:  make(map[websocket.ClientId]struct{}),
+		connectionMx: &sync.Mutex{},
+		connections:  make(map[string]Connection),
+	}
+
+	return topic
 }
 
 func (update *Update) Topic() abstraction.BrokerTopic {
 	return UpdateName
 }
 
 func (update *Update) Push(push abstraction.BrokerPush) error {
+	connection, ok := push.(*Connection)
+	if !ok {
+		return topics.ErrUnexpectedPush{Push: push}
+	}
+
+	update.connectionMx.Lock()
+	defer update.connectionMx.Unlock()
+	update.connections[connection.Name] = *connection
+
+	rawPayload, err := json.Marshal(update.connections)
+	if err != nil {
+		return err
+	}
+
+	message := websocket.Message{
+		Topic:   UpdateName,
+		Payload: rawPayload,
+	}
+
+	update.subscriberMx.Lock()
+	defer update.subscriberMx.Unlock()
+	flagged := make([]websocket.ClientId, 0, len(update.subscribers))
+	for id := range update.subscribers {
+		err := update.pool.Write(id, message)
+		if err != nil {
+			flagged = append(flagged, id)
+		}
+	}
+
+	for _, id := range flagged {
+		update.pool.Disconnect(id, ws.CloseNormalClosure, "client disconnected")
+		delete(update.subscribers, id)
+		fmt.Printf("connection/update unsubscribed %s\n", uuid.UUID(id).String())
+	}
+
 	return nil
 }
 
 func (update *Update) Pull(request abstraction.BrokerRequest) (abstraction.BrokerResponse, error) {
-	return nil, nil
+	return nil, topics.ErrOpNotSupported{}
+}
+
+func (update *Update) ClientMessage(id websocket.ClientId, message *websocket.Message) {
+	update.subscriberMx.Lock()
+	defer update.subscriberMx.Unlock()
+
+	switch message.Topic {
+	case SubscribeName:
+		update.subscribers[id] = struct{}{}
+		fmt.Printf("connection/update subscribed %s\n", uuid.UUID(id).String())
+
+		update.connectionMx.Lock()
+		defer update.connectionMx.Unlock()
+		rawPayload, err := json.Marshal(update.connections)
+		if err != nil {
+			fmt.Println(err)
+		}
+
+		message := websocket.Message{
+			Topic:   UpdateName,
+			Payload: rawPayload,
+		}
+
+		err = update.pool.Write(id, message)
+		if err != nil {
+			fmt.Println(err)
+		}
+
+	default:
+		update.pool.Disconnect(id, ws.CloseUnsupportedData, "unsupported topic")
+		delete(update.subscribers, id)
+		fmt.Printf("connection/update unsubscribed %s\n", uuid.UUID(id).String())
+	}
 }
 
 func (update *Update) SetPool(pool *websocket.Pool) {