Add hub_state

be1-go/.gitignore

@@ -4,4 +4,6 @@ report.json
 coverage.out
 coverage.html
 configServer1.json
-configServer2.json
+configServer2.json
+configServer3.json
+configServer4.json

be1-go/channel/lao/mod.go

@@ -4,6 +4,7 @@ import (
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
+	"golang.org/x/exp/slices"
 	popstellar "popstellar"
 	"popstellar/channel"
 	"popstellar/channel/authentication"
@@ -705,9 +706,12 @@ func (c *Channel) createAndSendLAOGreet() error {
 	peers := []messagedata.Peer{}
 
 	for _, info := range c.hub.GetPeersInfo() {
-		peers = append(peers, messagedata.Peer{
+		peer := messagedata.Peer{
 			Address: info.ClientAddress,
-		})
+		}
+		if !slices.Contains(peers, peer) {
+			peers = append(peers, peer)
+		}
 	}
 
 	msgData := messagedata.LaoGreet{

be1-go/hub/standard_hub/hub_state/channels_by_id.go

@@ -0,0 +1,40 @@
+package hub_state
+
+import (
+	"maps"
+	"popstellar/channel"
+	"sync"
+)
+
+// ChannelsById provides a thread-safe structure that stores channel ids with their corresponding channels
+type ChannelsById struct {
+	sync.RWMutex
+	table map[string]channel.Channel
+}
+
+func NewChannelsById() ChannelsById {
+	return ChannelsById{
+		table: make(map[string]channel.Channel),
+	}
+}
+
+func (c *ChannelsById) Add(channelId string, channel channel.Channel) {
+	c.Lock()
+	defer c.Unlock()
+	c.table[channelId] = channel
+}
+
+func (c *ChannelsById) GetChannel(channelId string) (channel.Channel, bool) {
+	c.RLock()
+	defer c.RUnlock()
+	channel, ok := c.table[channelId]
+	return channel, ok
+}
+
+func (c *ChannelsById) GetAll() map[string]channel.Channel {
+	c.RLock()
+	defer c.RUnlock()
+	channelsCopy := make(map[string]channel.Channel)
+	maps.Copy(channelsCopy, c.table)
+	return channelsCopy
+}

be1-go/hub/standard_hub/hub_state/ids_by_channel.go

@@ -0,0 +1,54 @@
+package hub_state
+
+import (
+	"golang.org/x/exp/slices"
+	"maps"
+	"sync"
+)
+
+// IdsByChannel provides a thread-safe structure that stores a channel id with its corresponding message ids
+type IdsByChannel struct {
+	sync.RWMutex
+	table map[string][]string
+}
+
+func NewIdsByChannel() IdsByChannel {
+	return IdsByChannel{
+		table: make(map[string][]string),
+	}
+}
+
+func (i *IdsByChannel) Add(channel string, id string) {
+	i.Lock()
+	defer i.Unlock()
+	messageIds, channelStored := i.table[channel]
+	if !channelStored {
+		i.table[channel] = append(i.table[channel], id)
+		return
+	}
+	alreadyStoredId := slices.Contains(messageIds, id)
+	if !alreadyStoredId {
+		i.table[channel] = append(i.table[channel], id)
+	}
+}
+
+func (i *IdsByChannel) AddAll(channel string, ids []string) {
+	i.Lock()
+	defer i.Unlock()
+	i.table[channel] = append(i.table[channel], ids...)
+}
+
+func (i *IdsByChannel) GetAll() map[string][]string {
+	i.RLock()
+	defer i.RUnlock()
+	tableCopy := make(map[string][]string)
+	maps.Copy(tableCopy, i.table)
+	return tableCopy
+}
+
+func (i *IdsByChannel) IsEmpty() bool {
+	i.RLock()
+	defer i.RUnlock()
+
+	return len(i.table) == 0
+}

be1-go/hub/standard_hub/hub_state/peers.go

@@ -0,0 +1,54 @@
+package hub_state
+
+import (
+	"golang.org/x/exp/slices"
+	"popstellar/message/query/method"
+	"sync"
+)
+
+// Peers provides a thread-safe structure that stores the peers' information
+type Peers struct {
+	sync.RWMutex
+	// peersInfo stores the info of the peers: public key, client and server endpoints associated with the socket ID
+	peersInfo map[string]method.ServerInfo
+	// peersGreeted stores the peers that were greeted by the socket ID
+	peersGreeted []string
+}
+
+func NewPeers() Peers {
+	return Peers{
+		peersInfo:    make(map[string]method.ServerInfo),
+		peersGreeted: make([]string, 0),
+	}
+}
+
+func (p *Peers) AddPeerInfo(socketId string, info method.ServerInfo) {
+	p.Lock()
+	defer p.Unlock()
+	p.peersInfo[socketId] = info
+}
+
+func (p *Peers) AddPeerGreeted(socketId string) {
+	p.Lock()
+	defer p.Unlock()
+	if slices.Contains(p.peersGreeted, socketId) {
+		return
+	}
+	p.peersGreeted = append(p.peersGreeted, socketId)
+}
+
+func (p *Peers) GetAllPeersInfo() []method.ServerInfo {
+	p.RLock()
+	defer p.RUnlock()
+	peersInfo := make([]method.ServerInfo, 0)
+	for _, info := range p.peersInfo {
+		peersInfo = append(peersInfo, info)
+	}
+	return peersInfo
+}
+
+func (p *Peers) IsPeerGreeted(socketId string) bool {
+	p.RLock()
+	defer p.RUnlock()
+	return slices.Contains(p.peersGreeted, socketId)
+}

be1-go/hub/standard_hub/hub_state/queries.go

@@ -0,0 +1,60 @@
+package hub_state
+
+import (
+	"popstellar/message/query/method"
+	"sync"
+)
+
+// Queries let the hub remember all queries that it sent to other servers
+type Queries struct {
+	sync.Mutex
+	// state stores the ID of the server's queries and their state. False for a
+	// query not yet answered, else true.
+	state map[int]*bool
+	// getMessagesByIdQueries stores the server's getMessagesByIds queries by their ID.
+	getMessagesByIdQueries map[int]method.GetMessagesById
+	// nextID store the ID of the next query
+	nextID int
+}
+
+// NewQueries creates a new queries struct
+func NewQueries() Queries {
+	return Queries{
+		state:                  make(map[int]*bool),
+		getMessagesByIdQueries: make(map[int]method.GetMessagesById),
+	}
+}
+
+// GetQueryState returns a given query's state
+func (q *Queries) GetQueryState(id int) *bool {
+	q.Lock()
+	defer q.Unlock()
+
+	return q.state[id]
+}
+
+// GetNextID returns the next query ID
+func (q *Queries) GetNextID() int {
+	q.Lock()
+	defer q.Unlock()
+
+	id := q.nextID
+	q.nextID++
+	return id
+}
+
+// SetQueryState sets the state of the query with the given ID
+func (q *Queries) SetQueryState(id int, state bool) {
+	q.Lock()
+	defer q.Unlock()
+
+	q.state[id] = &state
+}
+
+// AddQuery adds the given query to the table
+func (q *Queries) AddQuery(id int, query method.GetMessagesById) {
+	q.Lock()
+	defer q.Unlock()
+
+	q.getMessagesByIdQueries[id] = query
+}